diff --git a/Cargo.lock b/Cargo.lock
index 300d9e311..0c08091f6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -9972,6 +9972,7 @@ dependencies = [
  "thiserror 2.0.19",
  "time",
  "tracing",
+ "tracing-subscriber",
 ]
 
 [[package]]
diff --git a/crates/config/src/constants/body_limits.rs b/crates/config/src/constants/body_limits.rs
index 4a806045a..7e6adba74 100644
--- a/crates/config/src/constants/body_limits.rs
+++ b/crates/config/src/constants/body_limits.rs
@@ -28,6 +28,15 @@ pub const MAX_ADMIN_REQUEST_BODY_SIZE: usize = 1024 * 1024; // 1 MB
 /// Rationale: ZIP archives with hundreds of IAM entities. 10MB allows ~10,000 small configs.
 pub const MAX_IAM_IMPORT_SIZE: usize = 10 * 1024 * 1024; // 10 MB
 
+/// Maximum total size the members of an IAM import ZIP may expand to (100 MB).
+/// Used for: bounding decompression of `ImportIam` archive members.
+/// Rationale: `MAX_IAM_IMPORT_SIZE` caps the *compressed* upload only. Deflate
+/// reaches ratios far above 100:1, so without a separate budget a 10 MB archive
+/// can expand without bound. 100 MB keeps a 10x headroom over the compressed cap
+/// — ample for legitimate IAM exports, which are small JSON documents — while
+/// keeping the worst case bounded.
+pub const MAX_IAM_IMPORT_EXPANDED_SIZE: u64 = 100 * 1024 * 1024; // 100 MB
+
 /// Maximum size for bucket metadata import operations (100 MB)
 /// Used for: Bucket metadata import containing configurations for many buckets
 /// Rationale: Large deployments may have thousands of buckets with various configs.
@@ -54,3 +63,12 @@ pub const MAX_HEAL_REQUEST_SIZE: usize = 1024 * 1024; // 1 MB
 /// 10MB provides generous headroom for legitimate responses while preventing
 /// memory exhaustion from malicious or misconfigured remote services.
 pub const MAX_S3_CLIENT_RESPONSE_SIZE: usize = 10 * 1024 * 1024; // 10 MB
+
+/// Maximum size for OIDC provider response bodies (1 MB)
+/// Used for: discovery documents, JWKS documents and token endpoint responses
+/// Rationale: a hostile or compromised identity provider must not be able to exhaust
+/// memory through an arbitrarily large or endless response body.
+/// - Discovery documents: typically < 10KB
+/// - JWKS documents: typically < 50KB
+/// - Token responses: typically < 10KB
+pub const MAX_OIDC_RESPONSE_SIZE: usize = 1024 * 1024; // 1 MB
diff --git a/crates/config/src/constants/protocols.rs b/crates/config/src/constants/protocols.rs
index a102b5235..9e9016249 100644
--- a/crates/config/src/constants/protocols.rs
+++ b/crates/config/src/constants/protocols.rs
@@ -57,6 +57,7 @@ pub const ENV_WEBDAV_CERTS_DIR: &str = "RUSTFS_WEBDAV_CERTS_DIR";
 pub const ENV_WEBDAV_CA_FILE: &str = "RUSTFS_WEBDAV_CA_FILE";
 pub const ENV_WEBDAV_MAX_BODY_SIZE: &str = "RUSTFS_WEBDAV_MAX_BODY_SIZE";
 pub const ENV_WEBDAV_REQUEST_TIMEOUT: &str = "RUSTFS_WEBDAV_REQUEST_TIMEOUT";
+pub const ENV_WEBDAV_MAX_CONNECTIONS: &str = "RUSTFS_WEBDAV_MAX_CONNECTIONS";
 
 /// Default SFTP server bind address.
 pub const DEFAULT_SFTP_ADDRESS: &str = "0.0.0.0:2222";
diff --git a/crates/config/src/constants/tls.rs b/crates/config/src/constants/tls.rs
index f775f9933..c675be83e 100644
--- a/crates/config/src/constants/tls.rs
+++ b/crates/config/src/constants/tls.rs
@@ -142,6 +142,10 @@ pub const DEFAULT_H2_KEEP_ALIVE_TIMEOUT: u64 = 10;
 /// proxy's upstream idle-keepalive, or lower the proxy's keepalive below this
 /// value. Environments that expose RustFS directly to untrusted slow clients and
 /// want tighter slowloris protection can lower it via the env var below.
+///
+/// The same budget bounds the TLS handshake on the listener, so an unauthenticated
+/// peer cannot park an accept task and its socket indefinitely by opening a
+/// connection and then stalling the handshake.
 pub const ENV_HTTP1_HEADER_READ_TIMEOUT: &str = "RUSTFS_HTTP1_HEADER_READ_TIMEOUT";
 pub const DEFAULT_HTTP1_HEADER_READ_TIMEOUT: u64 = 75;
 
diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs
index 583947881..a42ab73b7 100644
--- a/crates/ecstore/src/cluster/rpc/remote_disk.rs
+++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs
@@ -1443,7 +1443,7 @@ impl DiskAPI for RemoteDisk {
 
                 Ok(infos)
             },
-            Duration::ZERO,
+            get_max_timeout_duration(),
         )
         .await
     }
@@ -1522,7 +1522,7 @@ impl DiskAPI for RemoteDisk {
 
                 Ok(())
             },
-            Duration::ZERO,
+            get_max_timeout_duration(),
         )
         .await
     }
@@ -5210,4 +5210,89 @@ mod tests {
                 })
         );
     }
+
+    /// Peer that completes the TCP connect and then goes silent, so every RPC issued over the
+    /// cached lazy channel stays pending until the caller's own deadline fires.
+    async fn spawn_stalled_grpc_peer() -> Option<(String, tokio::task::JoinHandle<()>)> {
+        let listener = match TcpListener::bind("127.0.0.1:0").await {
+            Ok(listener) => listener,
+            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
+            Err(err) => panic!("test listener should bind: {err}"),
+        };
+        let addr = listener.local_addr().expect("listener local address should be available");
+        let accept_task = tokio::spawn(async move {
+            let mut accepted = Vec::new();
+            while let Ok((stream, _)) = listener.accept().await {
+                accepted.push(stream);
+            }
+        });
+
+        let base_addr = format!("http://{}:{}", addr.ip(), addr.port());
+        let channel = TonicEndpoint::from_shared(base_addr.clone())
+            .expect("stalled peer endpoint should parse")
+            .connect_lazy();
+        runtime_sources::cache_test_node_channel(base_addr.clone(), channel).await;
+        Some((base_addr, accept_task))
+    }
+
+    async fn remote_disk_for_addr(base_addr: &str) -> RemoteDisk {
+        let url = url::Url::parse(&format!("{base_addr}/data/rustfs0")).expect("endpoint url should parse");
+        let endpoint = Endpoint {
+            url,
+            is_local: false,
+            pool_idx: 0,
+            set_idx: 0,
+            disk_idx: 0,
+        };
+        RemoteDisk::new(
+            &endpoint,
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+            Arc::new(TcpHttpInternodeDataTransport),
+        )
+        .await
+        .expect("remote disk should construct")
+    }
+
+    #[tokio::test]
+    async fn list_volumes_bounds_the_wait_on_a_stalled_peer() {
+        runtime_sources::ensure_test_rpc_secret();
+        let Some((base_addr, accept_task)) = spawn_stalled_grpc_peer().await else {
+            return;
+        };
+        let remote_disk = remote_disk_for_addr(&base_addr).await;
+
+        temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("1"))], async {
+            let err = tokio::time::timeout(Duration::from_secs(10), remote_disk.list_volumes())
+                .await
+                .expect("list_volumes must bound the wait on a stalled peer")
+                .expect_err("a stalled peer must fail list_volumes");
+            assert!(matches!(err, DiskError::Timeout), "expected the operation deadline to fire, got {err:?}");
+        })
+        .await;
+
+        accept_task.abort();
+    }
+
+    #[tokio::test]
+    async fn delete_volume_bounds_the_wait_on_a_stalled_peer() {
+        runtime_sources::ensure_test_rpc_secret();
+        let Some((base_addr, accept_task)) = spawn_stalled_grpc_peer().await else {
+            return;
+        };
+        let remote_disk = remote_disk_for_addr(&base_addr).await;
+
+        temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("1"))], async {
+            let err = tokio::time::timeout(Duration::from_secs(10), remote_disk.delete_volume("bucket", false))
+                .await
+                .expect("delete_volume must bound the wait on a stalled peer")
+                .expect_err("a stalled peer must fail delete_volume");
+            assert!(matches!(err, DiskError::Timeout), "expected the operation deadline to fire, got {err:?}");
+        })
+        .await;
+
+        accept_task.abort();
+    }
 }
diff --git a/crates/iam/src/oidc.rs b/crates/iam/src/oidc.rs
index 3fc9eb0ea..f0f7dcb2a 100644
--- a/crates/iam/src/oidc.rs
+++ b/crates/iam/src/oidc.rs
@@ -28,7 +28,7 @@ use openidconnect::{
 use reqwest::Client;
 use rustfs_config::oidc::*;
 use rustfs_config::server_config::{Config as ServerConfig, KVS};
-use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY, EnableState};
+use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY, EnableState, MAX_OIDC_RESPONSE_SIZE};
 use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive};
 use rustfs_utils::egress::OutboundPolicy;
 use serde::{Deserialize, Serialize};
@@ -51,6 +51,8 @@ const EVENT_OIDC_HTTP: &str = "oidc_http";
 const OIDC_JWKS_REFRESH_INTERVAL: StdDuration = StdDuration::from_secs(24 * 60 * 60);
 const OIDC_DISCOVERY_TRANSPORT_RETRIES: usize = 3;
 const OIDC_DISCOVERY_TRANSPORT_RETRY_DELAY: StdDuration = StdDuration::from_millis(50);
+const OIDC_HTTP_REQUEST_TIMEOUT: StdDuration = StdDuration::from_secs(10);
+const OIDC_HTTP_CONNECT_TIMEOUT: StdDuration = StdDuration::from_secs(3);
 const OIDC_PLUGIN_AUTHN_WINDOW: StdDuration = StdDuration::from_secs(60);
 
 #[derive(Debug, Clone, Copy, Default)]
@@ -254,6 +256,7 @@ fn oidc_http_error_diagnostics(error: &OidcHttpError) -> (&'static str, String)
         OidcHttpError::Reqwest(_) => ("request", String::new()),
         OidcHttpError::Http(_) => ("http_build", String::new()),
         OidcHttpError::ForbiddenOutbound(_) => ("forbidden_outbound", String::new()),
+        OidcHttpError::ResponseTooLarge(limit) => ("response_too_large", limit.to_string()),
     }
 }
 
@@ -268,6 +271,9 @@ pub enum OidcHttpError {
     /// connection was attempted (invalid URL, loopback/link-local/metadata/private IP,
     /// or a malformed allow-origins configuration).
     ForbiddenOutbound(String),
+    /// The provider response body exceeded [`MAX_OIDC_RESPONSE_SIZE`] and was abandoned
+    /// instead of being buffered in full.
+    ResponseTooLarge(usize),
 }
 
 impl std::fmt::Display for OidcHttpError {
@@ -276,6 +282,7 @@ impl std::fmt::Display for OidcHttpError {
             Self::Reqwest(e) => write!(f, "{e}"),
             Self::Http(e) => write!(f, "{e}"),
             Self::ForbiddenOutbound(reason) => write!(f, "outbound request rejected: {reason}"),
+            Self::ResponseTooLarge(limit) => write!(f, "oidc response body exceeds {limit} bytes"),
         }
     }
 }
@@ -285,7 +292,7 @@ impl std::error::Error for OidcHttpError {
         match self {
             Self::Reqwest(e) => Some(e),
             Self::Http(e) => Some(e),
-            Self::ForbiddenOutbound(_) => None,
+            Self::ForbiddenOutbound(_) | Self::ResponseTooLarge(_) => None,
         }
     }
 }
@@ -309,7 +316,8 @@ pub(crate) struct ReqwestHttpClient {
 /// link-local, metadata, multicast and unauthorized private addresses up front, and the
 /// returned `OutboundDnsResolver` re-resolves and re-classifies the host on every new
 /// connection so DNS rebinding fails closed. Redirects are not followed: a redirect target
-/// would otherwise skip URL-shape re-validation.
+/// would otherwise skip URL-shape re-validation. The timeouts bound how long a slow or
+/// stalled provider can pin the calling task.
 fn build_oidc_http_client(uri: &str, policy_override: Option<&OutboundPolicy>) -> Result<Client, OidcHttpError> {
     let url = Url::parse(uri).map_err(|_| OidcHttpError::ForbiddenOutbound("invalid outbound OIDC URL".to_string()))?;
     let resolver = match policy_override {
@@ -322,13 +330,35 @@ fn build_oidc_http_client(uri: &str, policy_override: Option<&OutboundPolicy>) -
 
     let mut builder = reqwest::Client::builder()
         .dns_resolver(resolver)
-        .redirect(reqwest::redirect::Policy::none());
+        .redirect(reqwest::redirect::Policy::none())
+        .timeout(OIDC_HTTP_REQUEST_TIMEOUT)
+        .connect_timeout(OIDC_HTTP_CONNECT_TIMEOUT);
     if should_bypass_proxy_for_oidc_uri(uri) {
         builder = builder.no_proxy();
     }
     builder.build().map_err(OidcHttpError::Reqwest)
 }
 
+/// Buffer a provider response body, failing closed once `limit` bytes have been seen.
+///
+/// `Response::bytes` would buffer the whole body unconditionally, so a hostile or compromised
+/// provider endpoint could stream an arbitrarily large (or endless) body into memory.
+async fn read_bounded_response_body(response: reqwest::Response, limit: usize) -> Result<Vec<u8>, OidcHttpError> {
+    if response.content_length().is_some_and(|len| len > limit as u64) {
+        return Err(OidcHttpError::ResponseTooLarge(limit));
+    }
+
+    let mut response = response;
+    let mut body = Vec::new();
+    while let Some(chunk) = response.chunk().await.map_err(OidcHttpError::Reqwest)? {
+        if body.len() + chunk.len() > limit {
+            return Err(OidcHttpError::ResponseTooLarge(limit));
+        }
+        body.extend_from_slice(&chunk);
+    }
+    Ok(body)
+}
+
 fn should_bypass_proxy_for_oidc_uri(uri: &str) -> bool {
     let Some(host) = Url::parse(uri).ok().and_then(|url| url.host_str().map(str::to_owned)) else {
         return false;
@@ -407,21 +437,23 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient {
 
             let status = response.status();
             let headers = response.headers().clone();
-            let body_bytes = response.bytes().await.map_err(|err| {
-                error!(
-                    event = EVENT_OIDC_HTTP,
-                    component = LOG_COMPONENT_IAM,
-                    subsystem = LOG_SUBSYSTEM_OIDC,
-                    result = "response_body_failed",
-                    method = %method,
-                    uri = %uri,
-                    status = status.as_u16(),
-                    elapsed_ms,
-                    error = %err,
-                    "oidc outbound http"
-                );
-                OidcHttpError::Reqwest(err)
-            })?;
+            let body_bytes = read_bounded_response_body(response, MAX_OIDC_RESPONSE_SIZE)
+                .await
+                .map_err(|err| {
+                    error!(
+                        event = EVENT_OIDC_HTTP,
+                        component = LOG_COMPONENT_IAM,
+                        subsystem = LOG_SUBSYSTEM_OIDC,
+                        result = "response_body_failed",
+                        method = %method,
+                        uri = %uri,
+                        status = status.as_u16(),
+                        elapsed_ms,
+                        error = %err,
+                        "oidc outbound http"
+                    );
+                    err
+                })?;
             if tracing::enabled!(tracing::Level::DEBUG) {
                 let response_headers = format_http_headers(&headers);
                 debug!(
@@ -442,7 +474,7 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient {
 
             let mut http_response = http::Response::builder()
                 .status(status)
-                .body(body_bytes.to_vec())
+                .body(body_bytes)
                 .map_err(OidcHttpError::Http)?;
             *http_response.headers_mut() = headers;
 
@@ -2837,6 +2869,93 @@ mod tests {
         );
     }
 
+    /// Serve exactly `body_len` bytes with no `Content-Length`, so the body ends only at EOF
+    /// and the size guard cannot rely on an advertised length.
+    fn start_unbounded_body_server(body_len: usize) -> Option<(String, std::thread::JoinHandle<()>)> {
+        use std::io::{Read, Write};
+        use std::net::TcpListener;
+        use std::sync::mpsc;
+        use std::time::Duration;
+
+        let listener = match TcpListener::bind("127.0.0.1:0") {
+            Ok(listener) => listener,
+            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
+            Err(err) => panic!("test listener should bind: {err}"),
+        };
+        let base = format!("http://{}", listener.local_addr().expect("listener local address should be available"));
+        let (ready_tx, ready_rx) = mpsc::channel();
+
+        let handle = std::thread::spawn(move || {
+            let _ = ready_tx.send(());
+            let Ok((mut stream, _)) = listener.accept() else {
+                return;
+            };
+            let _ = stream.set_read_timeout(Some(Duration::from_secs(1)));
+            let mut buffer = [0u8; 4096];
+            let _ = stream.read(&mut buffer);
+            let _ = stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nConnection: close\r\n\r\n");
+
+            let chunk = vec![b'a'; 64 * 1024];
+            let mut written = 0usize;
+            while written < body_len {
+                let take = chunk.len().min(body_len - written);
+                if stream.write_all(&chunk[..take]).is_err() {
+                    break;
+                }
+                written += take;
+            }
+            let _ = stream.flush();
+        });
+        ready_rx
+            .recv_timeout(Duration::from_millis(100))
+            .expect("mock body server should become ready");
+
+        Some((base, handle))
+    }
+
+    async fn fetch_oidc_mock_body(base: &str) -> Result<Vec<u8>, OidcHttpError> {
+        let policy = OutboundPolicy::from_allowed_origins(base).expect("origin should parse");
+        let client = ReqwestHttpClient::with_policy(policy);
+        let request = http::Request::builder()
+            .method(http::Method::GET)
+            .uri(base)
+            .body(Vec::new())
+            .expect("request should build");
+        client.call(request).await.map(http::Response::into_body)
+    }
+
+    #[tokio::test]
+    async fn oidc_response_body_at_the_limit_is_accepted() {
+        let Some((base, handle)) = start_unbounded_body_server(MAX_OIDC_RESPONSE_SIZE) else {
+            return;
+        };
+
+        let body = fetch_oidc_mock_body(&base)
+            .await
+            .expect("a body at the limit must be accepted");
+
+        assert_eq!(body.len(), MAX_OIDC_RESPONSE_SIZE);
+        handle.join().expect("mock body server thread should exit");
+    }
+
+    #[tokio::test]
+    async fn oidc_response_body_past_the_limit_is_rejected() {
+        let Some((base, handle)) = start_unbounded_body_server(MAX_OIDC_RESPONSE_SIZE + 1) else {
+            return;
+        };
+
+        let err = fetch_oidc_mock_body(&base)
+            .await
+            .map(|body| body.len())
+            .expect_err("an oversized provider response must fail closed instead of being buffered");
+
+        assert!(
+            matches!(err, OidcHttpError::ResponseTooLarge(MAX_OIDC_RESPONSE_SIZE)),
+            "unexpected error: {err}"
+        );
+        handle.join().expect("mock body server thread should exit");
+    }
+
     /// Helper to create an OidcSys with configs only (no provider states needed).
     fn make_test_sys(configs: Vec<OidcProviderConfig>) -> OidcSys {
         let mut config_map = HashMap::new();
diff --git a/crates/kms/src/backends/local.rs b/crates/kms/src/backends/local.rs
index 480b4e989..755f23dcd 100644
--- a/crates/kms/src/backends/local.rs
+++ b/crates/kms/src/backends/local.rs
@@ -32,12 +32,41 @@ use rand::RngExt;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::collections::HashMap;
-use std::path::PathBuf;
+use std::path::{Component, Path, PathBuf};
 use std::time::Duration;
 use tokio::fs;
 use tokio::sync::RwLock;
 use tracing::{debug, warn};
 
+/// Reject key identifiers that would not name a single file directly inside the key
+/// directory.
+///
+/// The rule is containment, not a character allowlist: anything that stays inside
+/// `key_dir` is accepted, so identifiers already in use by existing deployments keep
+/// resolving. Only separators, traversal and the degenerate cases are refused, which is
+/// what stops `key_dir.join(...)` from escaping.
+fn validate_key_id(key_id: &str) -> Result<()> {
+    if key_id.is_empty() {
+        return Err(KmsError::invalid_key("key identifier must not be empty"));
+    }
+    if key_id.contains('/') || key_id.contains('\\') || key_id.contains('\0') {
+        return Err(KmsError::invalid_key(format!(
+            "key identifier must not contain path separators or NUL: {key_id:?}"
+        )));
+    }
+
+    // Catches `.`, `..`, absolute paths, and platform-specific forms such as Windows
+    // drive prefixes, all of which would move the join outside key_dir.
+    let file_name = format!("{key_id}.key");
+    let mut components = Path::new(&file_name).components();
+    match (components.next(), components.next()) {
+        (Some(Component::Normal(_)), None) => Ok(()),
+        _ => Err(KmsError::invalid_key(format!(
+            "key identifier must name a single file inside the key directory: {key_id:?}"
+        ))),
+    }
+}
+
 const LOCAL_KMS_MASTER_KEY_SALT_FILE: &str = ".master-key.salt";
 const LOCAL_KMS_MASTER_KEY_SALT_LEN: usize = 16;
 const LOCAL_KMS_MASTER_KEY_LEN: usize = 32;
@@ -187,14 +216,26 @@ impl LocalKmsClient {
         Ok(())
     }
 
-    /// Get the file path for a master key
-    fn master_key_path(&self, key_id: &str) -> PathBuf {
-        self.config.key_dir.join(format!("{key_id}.key"))
+    /// Get the file path for a master key.
+    ///
+    /// Key identifiers reach this from request input (the `name` tag on CreateKey, the
+    /// `keyId` body field or query parameter on DeleteKey), so they are joined onto
+    /// `key_dir` only after being confirmed to name a single file inside it. Without that
+    /// check an identifier such as `../../tmp/evil` escapes the configured key directory,
+    /// turning key creation into a constrained arbitrary-file write and key deletion into
+    /// a cross-directory delete.
+    ///
+    /// Every filesystem path in this backend is derived here, so validating at this one
+    /// point covers `decode_stored_key`, `load_master_key`, `save_master_key`, `create_key`
+    /// and `delete_key`.
+    fn master_key_path(&self, key_id: &str) -> Result<PathBuf> {
+        validate_key_id(key_id)?;
+        Ok(self.config.key_dir.join(format!("{key_id}.key")))
     }
 
     /// Decode and decrypt a stored key file, returning both the metadata and decrypted key material
     async fn decode_stored_key(&self, key_id: &str) -> Result<(StoredMasterKey, Vec<u8>)> {
-        let key_path = self.master_key_path(key_id);
+        let key_path = self.master_key_path(key_id)?;
         if !fs::try_exists(&key_path).await? {
             return Err(KmsError::key_not_found(key_id));
         }
@@ -284,7 +325,7 @@ impl LocalKmsClient {
 
     /// Save a master key to disk
     async fn save_master_key(&self, master_key: &MasterKeyInfo, key_material: &[u8]) -> Result<()> {
-        let key_path = self.master_key_path(&master_key.key_id);
+        let key_path = self.master_key_path(&master_key.key_id)?;
 
         // Encrypt key material if master cipher is available
         let (encrypted_key_material, nonce, at_rest_protection) = if let Some(ref cipher) = self.master_cipher {
@@ -453,7 +494,7 @@ impl KmsClient for LocalKmsClient {
         debug!("Creating master key: {}", key_id);
 
         // Check if key already exists
-        if self.master_key_path(key_id).exists() {
+        if self.master_key_path(key_id)?.exists() {
             return Err(KmsError::key_already_exists(key_id));
         }
 
@@ -704,6 +745,15 @@ impl KmsBackend for LocalKmsBackend {
     async fn create_key(&self, request: CreateKeyRequest) -> Result<CreateKeyResponse> {
         let key_id = request.key_name.unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
 
+        // `save_master_key` writes through a temp file and renames over the destination, so
+        // creating a key under an existing name would replace its material and silently
+        // destroy the ability to decrypt everything wrapped under it. The sibling
+        // `KmsClient::create_key` has always refused this; the backend path did not, and
+        // this is the path the admin API uses.
+        if self.client.master_key_path(&key_id)?.exists() {
+            return Err(KmsError::key_already_exists(&key_id));
+        }
+
         // Create master key with description directly
         let _master_key = {
             let algorithm = "AES_256";
@@ -833,7 +883,7 @@ impl KmsBackend for LocalKmsBackend {
 
         let (deletion_date_str, deletion_date_dt) = if request.force_immediate.unwrap_or(false) {
             // For immediate deletion, actually delete the key from filesystem
-            let key_path = self.client.master_key_path(key_id);
+            let key_path = self.client.master_key_path(key_id)?;
             tokio::fs::remove_file(&key_path)
                 .await
                 .map_err(|e| KmsError::internal_error(format!("Failed to delete key file: {e}")))?;
@@ -1132,7 +1182,7 @@ mod tests {
         assert_eq!(salt.len(), LOCAL_KMS_MASTER_KEY_SALT_LEN);
 
         let stored: StoredMasterKey = serde_json::from_slice(
-            &fs::read(client.master_key_path("encrypted-key"))
+            &fs::read(client.master_key_path("encrypted-key").expect("valid key id"))
                 .await
                 .expect("stored key should exist"),
         )
@@ -1150,7 +1200,7 @@ mod tests {
             .expect("Failed to create plaintext-dev-only key");
 
         let stored: StoredMasterKey = serde_json::from_slice(
-            &fs::read(client.master_key_path("plaintext-key"))
+            &fs::read(client.master_key_path("plaintext-key").expect("valid key id"))
                 .await
                 .expect("stored key should exist"),
         )
@@ -1208,7 +1258,7 @@ mod tests {
             "nonce": Vec::<u8>::new()
         });
 
-        let key_path = client.master_key_path("legacy-key");
+        let key_path = client.master_key_path("legacy-key").expect("valid key id");
         fs::write(&key_path, serde_json::to_vec_pretty(&stored_key).expect("serialize test key"))
             .await
             .expect("write legacy key");
@@ -1226,7 +1276,7 @@ mod tests {
             .await
             .expect("Failed to create encrypted key");
 
-        let key_path = client.master_key_path("legacy-encrypted-key");
+        let key_path = client.master_key_path("legacy-encrypted-key").expect("valid key id");
         let mut stored_json: serde_json::Value =
             serde_json::from_slice(&fs::read(&key_path).await.expect("stored key should exist"))
                 .expect("stored key should deserialize");
@@ -1322,4 +1372,101 @@ mod tests {
             .expect_err("wrong beta.5 master key must not decrypt the fixture");
         assert!(matches!(error, KmsError::CryptographicError { .. }));
     }
+
+    /// R03-CAN-072 / R03-CAN-073: key identifiers arrive from request input, so every path
+    /// derived from one must stay inside the configured key directory. Traversal here would
+    /// turn CreateKey into a constrained arbitrary-file write and DeleteKey into a
+    /// cross-directory delete.
+    #[tokio::test]
+    async fn master_key_path_confines_key_ids_to_the_key_directory() {
+        let (client, temp_dir) = create_test_client().await;
+
+        // The invariant is containment, so assert that directly: whatever the input, the
+        // result is either refused or a path whose parent is exactly the key directory.
+        // Note `.` and `..` are contained rather than refused — the `.key` suffix turns
+        // them into the ordinary filenames `..key` and `...key`.
+        for candidate in [
+            "../escape",
+            "../../etc/rustfs",
+            "sub/dir",
+            "..",
+            ".",
+            "",
+            "/absolute",
+            "back\\slash",
+            "nul\0byte",
+            "....//....//escape",
+        ] {
+            match client.master_key_path(candidate) {
+                Err(KmsError::InvalidKey { .. }) => {}
+                Err(other) => panic!("unexpected error kind for {candidate:?}: {other:?}"),
+                Ok(path) => assert_eq!(
+                    path.parent(),
+                    Some(temp_dir.path()),
+                    "{candidate:?} was accepted but escapes the key directory: {path:?}"
+                ),
+            }
+        }
+
+        // The traversal forms specifically must be refused, not merely contained.
+        for escaping in ["../escape", "sub/dir", "/absolute", "back\\slash", "nul\0byte", ""] {
+            let err = client.master_key_path(escaping).expect_err("traversal must be refused");
+            assert!(
+                matches!(err, KmsError::InvalidKey { .. }),
+                "expected InvalidKey for {escaping:?}, got {err:?}"
+            );
+        }
+
+        // Ordinary identifiers, including the UUID form used when no name is supplied,
+        // must still resolve — and must land directly in the key directory.
+        for ok in ["test-key", "a.b_c-1", "3f2504e0-4f89-11d3-9a0c-0305e82c3301"] {
+            let path = client.master_key_path(ok).expect("valid key id must be accepted");
+            assert_eq!(
+                path.parent(),
+                Some(temp_dir.path()),
+                "{ok:?} must resolve directly inside the key directory"
+            );
+        }
+    }
+
+    /// R07-CAN-103: `save_master_key` writes a temp file and renames over the destination,
+    /// so creating a key under an existing name would replace its material and silently
+    /// destroy the ability to decrypt anything wrapped under it.
+    #[tokio::test]
+    async fn backend_create_key_refuses_to_replace_existing_key_material() {
+        let temp_dir = TempDir::new().expect("Failed to create temp dir");
+        let client = LocalKmsClient::new(LocalConfig {
+            key_dir: temp_dir.path().to_path_buf(),
+            master_key: Some("test-master-key".to_string()),
+            file_permissions: Some(0o600),
+        })
+        .await
+        .expect("Failed to create client");
+        let backend = LocalKmsBackend { client };
+
+        let request = || CreateKeyRequest {
+            key_name: Some("duplicate-key".to_string()),
+            ..Default::default()
+        };
+
+        backend.create_key(request()).await.expect("first create must succeed");
+        let original = backend
+            .client
+            .get_key_material("duplicate-key")
+            .await
+            .expect("key material must be readable after creation");
+
+        let err = backend
+            .create_key(request())
+            .await
+            .expect_err("creating a key under an existing name must be refused");
+        assert!(matches!(err, KmsError::KeyAlreadyExists { .. }), "expected KeyAlreadyExists, got {err:?}");
+
+        let after = backend
+            .client
+            .get_key_material("duplicate-key")
+            .await
+            .expect("original key material must survive the refused create");
+        assert_eq!(original, after, "existing key material must not be replaced");
+    }
 }
diff --git a/crates/protocols/src/common/dummy_storage.rs b/crates/protocols/src/common/dummy_storage.rs
index 68821bebd..7f9254798 100644
--- a/crates/protocols/src/common/dummy_storage.rs
+++ b/crates/protocols/src/common/dummy_storage.rs
@@ -37,8 +37,8 @@ use s3s::dto::{
     AbortMultipartUploadInput, AbortMultipartUploadOutput, CompleteMultipartUploadInput, CompleteMultipartUploadOutput,
     CopyObjectInput, CopyObjectOutput, CopyPartResult, CreateBucketOutput, CreateMultipartUploadInput,
     CreateMultipartUploadOutput, DeleteBucketOutput, DeleteObjectOutput, ETag, GetObjectOutput, HeadBucketOutput,
-    HeadObjectOutput, ListBucketsOutput, ListObjectsV2Input, ListObjectsV2Output, PutObjectInput, PutObjectOutput, StreamingBlob,
-    Timestamp, UploadPartCopyInput, UploadPartCopyOutput, UploadPartInput, UploadPartOutput,
+    HeadObjectOutput, ListBucketsOutput, ListObjectsV2Input, ListObjectsV2Output, Object, ObjectKey, PutObjectInput,
+    PutObjectOutput, StreamingBlob, Timestamp, UploadPartCopyInput, UploadPartCopyOutput, UploadPartInput, UploadPartOutput,
 };
 use std::collections::{HashMap, VecDeque};
 use std::sync::{Arc, Mutex};
@@ -121,6 +121,14 @@ pub struct HeadObjectCall {
     pub key: String,
 }
 
+/// Recorded invocation of delete_object. Tests assert on these to observe
+/// which objects a recursive delete path actually removed.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DeleteObjectCall {
+    pub bucket: String,
+    pub key: String,
+}
+
 struct Inner {
     // Response queues. Each method pops from its own queue. Empty queue
     // plus no default means a configured-miss error.
@@ -148,6 +156,8 @@ struct Inner {
     upload_part_calls: Vec<UploadPartCall>,
     complete_multipart_calls: Vec<CompleteCall>,
     head_object_calls: Vec<HeadObjectCall>,
+    delete_object_calls: Vec<DeleteObjectCall>,
+    delete_bucket_calls: Vec<String>,
 
     // Cancellation-test support. When stall_upload_part is true every
     // upload_part invocation signals upload_part_entered and then awaits
@@ -197,6 +207,8 @@ impl Inner {
             upload_part_calls: Vec::new(),
             complete_multipart_calls: Vec::new(),
             head_object_calls: Vec::new(),
+            delete_object_calls: Vec::new(),
+            delete_bucket_calls: Vec::new(),
             stall_upload_part: false,
             upload_part_entered: None,
             stall_put_object: false,
@@ -213,8 +225,13 @@ impl Inner {
 /// tested, and keep another clone for observation. Method calls are
 /// fire-and-forget from the driver's perspective and synchronous on the
 /// test side.
+///
+/// Cloning shares the same queues and observation logs, so a test can hand
+/// one clone to a driver that takes its backend by value and keep another
+/// for assertions.
+#[derive(Clone)]
 pub struct DummyBackend {
-    inner: Mutex<Inner>,
+    inner: Arc<Mutex<Inner>>,
 }
 
 // Drivers whose trait bounds require `Debug` (for example FtpsDriver) cannot be
@@ -237,7 +254,7 @@ impl DummyBackend {
     /// configured-miss error until a queue is populated.
     pub fn new() -> Self {
         Self {
-            inner: Mutex::new(Inner::new()),
+            inner: Arc::new(Mutex::new(Inner::new())),
         }
     }
 
@@ -377,6 +394,43 @@ impl DummyBackend {
             .push_back(Ok(ListObjectsV2Output::default()));
     }
 
+    /// Queue a list_objects_v2 Ok response listing the given keys as a
+    /// single, non-truncated page. Recursive-delete tests use this to give
+    /// the delete loop something to iterate over.
+    pub fn queue_list_objects_v2_ok_with_keys(&self, keys: &[&str]) {
+        let contents = keys
+            .iter()
+            .map(|key| Object {
+                key: Some(ObjectKey::from((*key).to_string())),
+                ..Default::default()
+            })
+            .collect();
+        let out = ListObjectsV2Output {
+            contents: Some(contents),
+            is_truncated: Some(false),
+            ..Default::default()
+        };
+        self.inner.lock().expect("lock").list_objects_v2.push_back(Ok(out));
+    }
+
+    /// Queue a delete_object Ok response.
+    pub fn queue_delete_object_ok(&self) {
+        self.inner
+            .lock()
+            .expect("lock")
+            .delete_object
+            .push_back(Ok(DeleteObjectOutput::default()));
+    }
+
+    /// Queue a delete_bucket Ok response.
+    pub fn queue_delete_bucket_ok(&self) {
+        self.inner
+            .lock()
+            .expect("lock")
+            .delete_bucket
+            .push_back(Ok(DeleteBucketOutput::default()));
+    }
+
     /// Queue a list_objects_v2 error. Used to verify that callers do
     /// not fall through to a destructive operation when the empty-check
     /// itself fails.
@@ -497,6 +551,16 @@ impl DummyBackend {
     }
 
     /// Snapshot the head_object call log.
+    /// Snapshot the recorded delete_object invocations.
+    pub fn delete_object_calls(&self) -> Vec<DeleteObjectCall> {
+        self.inner.lock().expect("lock").delete_object_calls.clone()
+    }
+
+    /// Snapshot the buckets passed to delete_bucket.
+    pub fn delete_bucket_calls(&self) -> Vec<String> {
+        self.inner.lock().expect("lock").delete_bucket_calls.clone()
+    }
+
     pub fn head_object_calls(&self) -> Vec<HeadObjectCall> {
         self.inner.lock().expect("lock").head_object_calls.clone()
     }
@@ -565,7 +629,12 @@ impl StorageBackend for DummyBackend {
     }
 
     async fn delete_object(&self, bucket: &str, key: &str, _ak: &str, _sk: &str) -> Result<DeleteObjectOutput, Self::Error> {
-        match self.inner.lock().expect("lock").delete_object.pop_front() {
+        let mut inner = self.inner.lock().expect("lock");
+        inner.delete_object_calls.push(DeleteObjectCall {
+            bucket: bucket.to_string(),
+            key: key.to_string(),
+        });
+        match inner.delete_object.pop_front() {
             Some(r) => r,
             None => Err(DummyError::NoSuchKey(format!("{bucket}/{key}"))),
         }
@@ -636,7 +705,9 @@ impl StorageBackend for DummyBackend {
     }
 
     async fn delete_bucket(&self, bucket: &str, _ak: &str, _sk: &str) -> Result<DeleteBucketOutput, Self::Error> {
-        match self.inner.lock().expect("lock").delete_bucket.pop_front() {
+        let mut inner = self.inner.lock().expect("lock");
+        inner.delete_bucket_calls.push(bucket.to_string());
+        match inner.delete_bucket.pop_front() {
             Some(r) => r,
             None => Err(DummyError::NoSuchBucket(bucket.to_string())),
         }
diff --git a/crates/protocols/src/common/session.rs b/crates/protocols/src/common/session.rs
index 8ba092579..8c684bed2 100644
--- a/crates/protocols/src/common/session.rs
+++ b/crates/protocols/src/common/session.rs
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use rustfs_credentials::Credentials;
 use rustfs_policy::auth::UserIdentity;
 use std::net::IpAddr;
 #[cfg(test)]
@@ -43,6 +44,21 @@ impl ProtocolPrincipal {
     }
 }
 
+/// Returns `true` when `credentials` are short-lived STS/AssumeRole credentials.
+///
+/// SECURITY: password-based protocol logins (FTPS, SFTP, WebDAV Basic) have nowhere to carry the
+/// session token that an STS credential is only valid with, so accepting the access/secret pair
+/// alone would authenticate the holder as the parent user with none of the session-policy
+/// restrictions the token encodes. Such logins must be rejected.
+///
+/// Service accounts also hold a signed token, but their policy is resolved from stored IAM state
+/// rather than from a token the client must present, so they stay eligible for these protocols.
+/// The `is_temp() && !is_service_account()` pairing is the same STS discriminator the IAM manager
+/// uses when routing an identity into the STS account cache.
+pub fn is_temporary_credential(credentials: &Credentials) -> bool {
+    credentials.is_temp() && !credentials.is_service_account()
+}
+
 /// Session context for protocol operations
 #[derive(Debug, Clone)]
 pub struct SessionContext {
@@ -85,6 +101,95 @@ pub fn test_session(protocol: Protocol) -> SessionContext {
 #[cfg(test)]
 mod regression_prevention {
     use super::*;
+    use rustfs_credentials::{IAM_POLICY_CLAIM_NAME_SA, INHERITED_POLICY_TYPE};
+    use serde_json::Value;
+    use std::collections::HashMap;
+    use time::{Duration, OffsetDateTime};
+
+    fn service_account_claims() -> HashMap<String, Value> {
+        let mut claims = HashMap::new();
+        claims.insert(IAM_POLICY_CLAIM_NAME_SA.to_string(), Value::String(INHERITED_POLICY_TYPE.to_string()));
+        claims
+    }
+
+    #[test]
+    fn sts_credentials_are_temporary() {
+        let sts = Credentials {
+            access_key: "VV0V3VYJK2PV6EG45X2Y".to_string(),
+            secret_key: "CS_TEST_SECRET_DO_NOT_LOG".to_string(),
+            session_token: "jwt-session-token".to_string(),
+            parent_user: "alice".to_string(),
+            ..Default::default()
+        };
+
+        assert!(is_temporary_credential(&sts));
+    }
+
+    #[test]
+    fn service_accounts_are_not_temporary() {
+        // Service accounts also carry a signed token; they must keep working over FTPS/SFTP/WebDAV.
+        let service_account = Credentials {
+            access_key: "39KNO04Z34D6T4AGL6E6".to_string(),
+            secret_key: "CS_TEST_SECRET_DO_NOT_LOG".to_string(),
+            session_token: "jwt-session-token".to_string(),
+            parent_user: "alice".to_string(),
+            claims: Some(service_account_claims()),
+            ..Default::default()
+        };
+
+        assert!(service_account.is_service_account());
+        assert!(!is_temporary_credential(&service_account));
+    }
+
+    #[test]
+    fn long_term_credentials_are_not_temporary() {
+        let long_term = Credentials {
+            access_key: "alice".to_string(),
+            secret_key: "CS_TEST_SECRET_DO_NOT_LOG".to_string(),
+            ..Default::default()
+        };
+
+        assert!(!is_temporary_credential(&long_term));
+    }
+
+    #[test]
+    fn expired_sts_credentials_are_rejected_by_the_validity_gate() {
+        // is_temp() goes false once the session expires, so the STS guard alone would let an
+        // expired session in. The `is_valid` check every password path runs first is what covers
+        // this case; assert both halves so neither can be dropped unnoticed.
+        let expired = Credentials {
+            access_key: "VV0V3VYJK2PV6EG45X2Y".to_string(),
+            secret_key: "CS_TEST_SECRET_DO_NOT_LOG".to_string(),
+            session_token: "jwt-session-token".to_string(),
+            parent_user: "alice".to_string(),
+            expiration: Some(OffsetDateTime::now_utc() - Duration::hours(1)),
+            ..Default::default()
+        };
+
+        assert!(!is_temporary_credential(&expired));
+        assert!(!expired.is_valid());
+    }
+
+    #[test]
+    fn password_auth_paths_reject_temporary_credentials() {
+        let guard = concat!("is_temporary_credential(&identity.", "credentials)");
+        for (protocol, source) in [
+            ("ftps", include_str!("../ftps/server.rs")),
+            ("webdav", include_str!("../webdav/server.rs")),
+            ("sftp", include_str!("../sftp/server.rs")),
+        ] {
+            let guard_at = source
+                .find(guard)
+                .unwrap_or_else(|| panic!("{protocol} password authentication must reject temporary STS credentials"));
+            let accept_at = source
+                .find(r#"result = "authenticated""#)
+                .unwrap_or_else(|| panic!("{protocol} has no authenticated log line to anchor the guard against"));
+            assert!(
+                guard_at < accept_at,
+                "{protocol} must reject temporary STS credentials before authentication succeeds"
+            );
+        }
+    }
 
     // Compile-time check that every Protocol variant is acknowledged here.
     // This is intentionally an exhaustive match with no wildcard arm: if a
diff --git a/crates/protocols/src/ftps/driver.rs b/crates/protocols/src/ftps/driver.rs
index 7b1225ffc..2075a4d46 100644
--- a/crates/protocols/src/ftps/driver.rs
+++ b/crates/protocols/src/ftps/driver.rs
@@ -170,6 +170,13 @@ where
         bucket: &str,
         session_context: &crate::common::session::SessionContext,
     ) -> Result<()> {
+        // SECURITY: s3:DeleteBucket does not imply the right to destroy the
+        // bucket contents. Enumerating and deleting each object are separate
+        // authorization boundaries and must be cleared on their own.
+        authorize_operation(session_context, &S3Action::ListBucket, bucket, None)
+            .await
+            .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?;
+
         // First, delete all objects in the bucket (with pagination)
         let mut continuation_token = None;
         loop {
@@ -196,6 +203,10 @@ where
                 if let Some(objects) = output.contents {
                     for obj in objects {
                         if let Some(obj_key) = obj.key {
+                            authorize_operation(session_context, &S3Action::DeleteObject, bucket, Some(&obj_key))
+                                .await
+                                .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?;
+
                             let _ = self
                                 .storage
                                 .delete_object(
@@ -929,6 +940,48 @@ mod tests {
         );
     }
 
+    /// RMD deletes every object in the bucket, so `s3:DeleteBucket` alone must
+    /// not be enough: each object needs its own `s3:DeleteObject` boundary. The
+    /// backend is primed so that the whole recursive delete would succeed if the
+    /// per-object check were removed again.
+    #[tokio::test]
+    async fn ftps_rmd_denied_per_object_does_not_delete_bucket_contents() {
+        use super::FtpsDriver;
+        use crate::common::dummy_storage::DummyBackend;
+        use crate::common::gateway::{S3Action, with_test_auth_override};
+        use crate::common::session::{Protocol, test_session};
+        use unftp_core::storage::StorageBackend as _;
+
+        let backend = DummyBackend::new();
+        backend.queue_list_objects_v2_ok_with_keys(&["secret.txt"]);
+        backend.queue_delete_object_ok();
+        backend.queue_delete_bucket_ok();
+
+        let driver = FtpsDriver::new(backend.clone());
+        let user = super::super::server::FtpsUser {
+            username: "bucket-only-user".to_string(),
+            name: None,
+            session_context: test_session(Protocol::Ftps),
+        };
+
+        let result = with_test_auth_override(
+            |action, _bucket, _object| !matches!(action, S3Action::DeleteObject),
+            driver.rmd(&user, "/victim-bucket"),
+        )
+        .await;
+
+        assert!(result.is_err(), "RMD must fail closed when s3:DeleteObject is denied for a bucket member");
+        assert!(
+            backend.delete_object_calls().is_empty(),
+            "no object may be deleted once s3:DeleteObject is denied, got {:?}",
+            backend.delete_object_calls()
+        );
+        assert!(
+            backend.delete_bucket_calls().is_empty(),
+            "the bucket must survive when its contents could not be authorized for deletion"
+        );
+    }
+
     proptest::proptest! {
         #[test]
         fn parse_s3_path_never_leaks_control_bytes_or_traversal_in_ok_output(
diff --git a/crates/protocols/src/ftps/server.rs b/crates/protocols/src/ftps/server.rs
index 6903c4afb..f76b6edb6 100644
--- a/crates/protocols/src/ftps/server.rs
+++ b/crates/protocols/src/ftps/server.rs
@@ -15,7 +15,7 @@
 use super::config::{FtpsConfig, FtpsInitError};
 use super::driver::FtpsDriver;
 use crate::common::client::s3::StorageBackend;
-use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext};
+use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext, is_temporary_credential};
 use crate::constants::{network::DEFAULT_SOURCE_IP, paths::ROOT_PATH};
 use libunftp::options::FtpsRequired;
 use rustfs_config::{DEFAULT_TLS_RELOAD_ENABLE, DEFAULT_TLS_RELOAD_INTERVAL, ENV_TLS_RELOAD_ENABLE, ENV_TLS_RELOAD_INTERVAL};
@@ -465,6 +465,19 @@ impl Authenticator for FtpsAuthenticator {
             AuthenticationError::BadUser
         })?;
 
+        if is_temporary_credential(&identity.credentials) {
+            warn!(
+                event = EVENT_FTPS_AUTH_STATE,
+                component = LOG_COMPONENT_PROTOCOLS,
+                subsystem = LOG_SUBSYSTEM_FTPS_AUTH,
+                result = "temporary_credential_rejected",
+                phase = "authenticate",
+                username = %masked_username,
+                "FTPS auth rejected temporary credential"
+            );
+            return Err(AuthenticationError::BadUser);
+        }
+
         // Constant-time secret comparison to prevent timing side-channel
         // attacks. Same primitive used by the SFTP handler and rustfs/src/auth.rs.
         let secret_matches: bool = identity
diff --git a/crates/protocols/src/sftp/server.rs b/crates/protocols/src/sftp/server.rs
index 8f3953912..503923a08 100644
--- a/crates/protocols/src/sftp/server.rs
+++ b/crates/protocols/src/sftp/server.rs
@@ -34,7 +34,7 @@ use super::lifecycle::{SessionDiag, SessionRegistry, new_session_registry};
 #[cfg(target_os = "linux")]
 use super::wedge_watchdog;
 use crate::common::client::s3::StorageBackend;
-use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext};
+use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext, is_temporary_credential};
 use russh::keys::{self, PrivateKey, PublicKeyBase64};
 use russh::server::{Auth, ChannelOpenHandle, Msg, Session};
 use russh::{Channel, ChannelId, ChannelOpenFailure, MethodKind, MethodSet, Pty, Sig};
@@ -1101,6 +1101,19 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
                 return Ok(Auth::reject());
             }
 
+            if is_temporary_credential(&identity.credentials) {
+                warn!(
+                    event = EVENT_SFTP_AUTH_STATE,
+                    component = LOG_COMPONENT_PROTOCOLS,
+                    subsystem = LOG_SUBSYSTEM_SFTP_AUTH,
+                    result = "temporary_credential_rejected",
+                    user = %masked_user,
+                    peer = %peer_addr,
+                    "sftp auth state changed"
+                );
+                return Ok(Auth::reject());
+            }
+
             // Constant-time secret comparison to prevent timing side-channel
             // attacks. Same primitive used by rustfs/src/auth.rs.
             use subtle::ConstantTimeEq;
diff --git a/crates/protocols/src/swift/slo.rs b/crates/protocols/src/swift/slo.rs
index 4d31a47da..ddb21d718 100644
--- a/crates/protocols/src/swift/slo.rs
+++ b/crates/protocols/src/swift/slo.rs
@@ -27,6 +27,9 @@ use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::io::Cursor;
 
+/// Maximum accepted size of an SLO manifest document
+const MAX_SLO_MANIFEST_SIZE: usize = 2 * 1024 * 1024;
+
 /// SLO manifest segment descriptor
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct SLOSegment {
@@ -274,14 +277,14 @@ pub async fn handle_slo_put(
         .map_err(|e| SwiftError::BadRequest(format!("Failed to read manifest: {}", e)))?
         .to_bytes();
 
-    // 2. Parse manifest
-    let manifest = SLOManifest::from_json(&manifest_bytes)?;
-
-    // 3. Validate manifest size (2MB limit)
-    if manifest_bytes.len() > 2 * 1024 * 1024 {
+    // 2. Validate manifest size (2MB limit)
+    if manifest_bytes.len() > MAX_SLO_MANIFEST_SIZE {
         return Err(SwiftError::BadRequest("Manifest exceeds 2MB".to_string()));
     }
 
+    // 3. Parse manifest
+    let manifest = SLOManifest::from_json(&manifest_bytes)?;
+
     // 4. Validate segments exist and match ETags/sizes
     manifest.validate(account, creds).await?;
 
@@ -326,6 +329,30 @@ pub async fn handle_slo_put(
         .map_err(|e| SwiftError::InternalServerError(format!("Failed to build response: {}", e)))
 }
 
+/// Read a stored SLO manifest without buffering more than [`MAX_SLO_MANIFEST_SIZE`].
+///
+/// The manifest lives at a caller-predictable `<object>.slo-manifest` key, so its content
+/// is attacker-controlled: the read must be bounded rather than sized by the stored object.
+async fn read_manifest_bytes<R>(reader: R) -> Result<Vec<u8>, SwiftError>
+where
+    R: tokio::io::AsyncRead + Unpin,
+{
+    use tokio::io::AsyncReadExt;
+
+    let mut manifest_bytes = Vec::new();
+    reader
+        .take(MAX_SLO_MANIFEST_SIZE as u64 + 1)
+        .read_to_end(&mut manifest_bytes)
+        .await
+        .map_err(|e| SwiftError::InternalServerError(format!("Failed to read manifest: {}", e)))?;
+
+    if manifest_bytes.len() > MAX_SLO_MANIFEST_SIZE {
+        return Err(SwiftError::BadRequest("Manifest exceeds 2MB".to_string()));
+    }
+
+    Ok(manifest_bytes)
+}
+
 /// Handle GET /v1/{account}/{container}/{object} for SLO
 pub async fn handle_slo_get(
     account: &str,
@@ -341,16 +368,9 @@ pub async fn handle_slo_get(
 
     // 1. Load manifest
     let manifest_key = format!("{}.slo-manifest", object);
-    let mut manifest_reader = object::get_object(account, container, &manifest_key, creds, None).await?;
+    let manifest_reader = object::get_object(account, container, &manifest_key, creds, None).await?;
 
-    // Read manifest bytes
-    let mut manifest_bytes = Vec::new();
-    use tokio::io::AsyncReadExt;
-    manifest_reader
-        .stream
-        .read_to_end(&mut manifest_bytes)
-        .await
-        .map_err(|e| SwiftError::InternalServerError(format!("Failed to read manifest: {}", e)))?;
+    let manifest_bytes = read_manifest_bytes(manifest_reader.stream).await?;
 
     let manifest = SLOManifest::from_json(&manifest_bytes)?;
 
@@ -472,16 +492,9 @@ pub async fn handle_slo_get_manifest(
 
     // Load and return the manifest JSON directly
     let manifest_key = format!("{}.slo-manifest", object);
-    let mut manifest_reader = object::get_object(account, container, &manifest_key, creds, None).await?;
+    let manifest_reader = object::get_object(account, container, &manifest_key, creds, None).await?;
 
-    // Read manifest bytes
-    let mut manifest_bytes = Vec::new();
-    use tokio::io::AsyncReadExt;
-    manifest_reader
-        .stream
-        .read_to_end(&mut manifest_bytes)
-        .await
-        .map_err(|e| SwiftError::InternalServerError(format!("Failed to read manifest: {}", e)))?;
+    let manifest_bytes = read_manifest_bytes(manifest_reader.stream).await?;
 
     let trans_id = generate_trans_id();
     Response::builder()
@@ -508,16 +521,9 @@ pub async fn handle_slo_delete(
 
     // 1. Load manifest
     let manifest_key = format!("{}.slo-manifest", object);
-    let mut manifest_reader = object::get_object(account, container, &manifest_key, creds, None).await?;
+    let manifest_reader = object::get_object(account, container, &manifest_key, creds, None).await?;
 
-    // Read manifest bytes
-    let mut manifest_bytes = Vec::new();
-    use tokio::io::AsyncReadExt;
-    manifest_reader
-        .stream
-        .read_to_end(&mut manifest_bytes)
-        .await
-        .map_err(|e| SwiftError::InternalServerError(format!("Failed to read manifest: {}", e)))?;
+    let manifest_bytes = read_manifest_bytes(manifest_reader.stream).await?;
 
     let manifest = SLOManifest::from_json(&manifest_bytes)?;
 
@@ -971,4 +977,47 @@ mod tests {
         // Empty path
         assert!(parse_segment_path("").is_err());
     }
+
+    #[tokio::test]
+    async fn test_read_manifest_bytes_accepts_manifest_at_size_limit() {
+        let source = std::io::Cursor::new(vec![b'x'; MAX_SLO_MANIFEST_SIZE]);
+        let bytes = read_manifest_bytes(source).await.expect("manifest at the limit is accepted");
+        assert_eq!(bytes.len(), MAX_SLO_MANIFEST_SIZE);
+    }
+
+    #[tokio::test]
+    async fn test_read_manifest_bytes_rejects_oversized_manifest() {
+        let source = std::io::Cursor::new(vec![b'x'; MAX_SLO_MANIFEST_SIZE + 1]);
+        let result = read_manifest_bytes(source).await.map(|bytes| bytes.len());
+        assert!(
+            matches!(result, Err(SwiftError::BadRequest(_))),
+            "manifest above the limit must be rejected, got {:?}",
+            result
+        );
+    }
+
+    /// A stored manifest is attacker-controlled, so the read must stop at the limit
+    /// instead of buffering the whole object.
+    #[tokio::test]
+    async fn test_read_manifest_bytes_stops_reading_oversized_manifest() {
+        use tokio::io::AsyncReadExt;
+
+        let total = 32 * 1024 * 1024u64;
+        let mut source = tokio::io::repeat(b'x').take(total);
+
+        let result = read_manifest_bytes(&mut source).await.map(|bytes| bytes.len());
+        assert!(
+            matches!(result, Err(SwiftError::BadRequest(_))),
+            "oversized manifest must be rejected, got {:?}",
+            result
+        );
+
+        let consumed = total - source.limit();
+        assert!(
+            consumed <= MAX_SLO_MANIFEST_SIZE as u64 + 1,
+            "read {} bytes, expected at most {}",
+            consumed,
+            MAX_SLO_MANIFEST_SIZE + 1
+        );
+    }
 }
diff --git a/crates/protocols/src/webdav/README.md b/crates/protocols/src/webdav/README.md
index 5c9ef9517..1976edb15 100644
--- a/crates/protocols/src/webdav/README.md
+++ b/crates/protocols/src/webdav/README.md
@@ -59,6 +59,7 @@ Configure WebDAV via environment variables:
 | `RUSTFS_WEBDAV_CA_FILE` | CA file for client verification | - |
 | `RUSTFS_WEBDAV_MAX_BODY_SIZE` | Max upload size (bytes) | 5GB |
 | `RUSTFS_WEBDAV_REQUEST_TIMEOUT` | Request timeout (seconds) | 300 |
+| `RUSTFS_WEBDAV_MAX_CONNECTIONS` | Max concurrently served connections | 1024 |
 
 ## Quick Start
 
diff --git a/crates/protocols/src/webdav/config.rs b/crates/protocols/src/webdav/config.rs
index c71922f50..c9b2e8eb0 100644
--- a/crates/protocols/src/webdav/config.rs
+++ b/crates/protocols/src/webdav/config.rs
@@ -44,6 +44,8 @@ pub struct WebDavConfig {
     pub max_body_size: u64,
     /// Request timeout in seconds (default: 300)
     pub request_timeout_secs: u64,
+    /// Maximum number of connections served concurrently (default: 1024)
+    pub max_connections: usize,
 }
 
 impl WebDavConfig {
@@ -51,6 +53,8 @@ impl WebDavConfig {
     pub const DEFAULT_MAX_BODY_SIZE: u64 = 5 * 1024 * 1024 * 1024;
     /// Default request timeout (300 seconds)
     pub const DEFAULT_REQUEST_TIMEOUT_SECS: u64 = 300;
+    /// Default concurrent connection cap
+    pub const DEFAULT_MAX_CONNECTIONS: usize = 1024;
 
     /// Validates the configuration
     pub async fn validate(&self) -> Result<(), WebDavInitError> {
@@ -84,6 +88,13 @@ impl WebDavConfig {
             return Err(WebDavInitError::InvalidConfig("request_timeout_secs cannot be zero".to_string()));
         }
 
+        // Validate connection cap. Zero is rejected rather than treated as
+        // "unlimited": an unbounded accept loop is the resource-exhaustion
+        // hole this cap exists to close.
+        if self.max_connections == 0 {
+            return Err(WebDavInitError::InvalidConfig("max_connections cannot be zero".to_string()));
+        }
+
         Ok(())
     }
 }
@@ -98,6 +109,7 @@ impl Default for WebDavConfig {
             ca_file: None,
             max_body_size: Self::DEFAULT_MAX_BODY_SIZE,
             request_timeout_secs: Self::DEFAULT_REQUEST_TIMEOUT_SECS,
+            max_connections: Self::DEFAULT_MAX_CONNECTIONS,
         }
     }
 }
diff --git a/crates/protocols/src/webdav/driver.rs b/crates/protocols/src/webdav/driver.rs
index 99237c452..11c91d193 100644
--- a/crates/protocols/src/webdav/driver.rs
+++ b/crates/protocols/src/webdav/driver.rs
@@ -1000,6 +1000,13 @@ where
 
     /// Recursively delete all objects in a bucket, then delete the bucket itself
     async fn delete_bucket_recursively(&self, bucket: &str) -> FsResult<()> {
+        // SECURITY: s3:DeleteBucket does not imply the right to destroy the
+        // bucket contents. Enumerating and deleting each object are separate
+        // authorization boundaries and must be cleared on their own.
+        authorize_operation(&self.session_context, &S3Action::ListBucket, bucket, None)
+            .await
+            .map_err(|_| FsError::Forbidden)?;
+
         // First, delete all objects in the bucket (with pagination)
         let mut continuation_token = None;
         loop {
@@ -1024,6 +1031,10 @@ where
                 if let Some(objects) = output.contents {
                     for obj in objects {
                         if let Some(obj_key) = obj.key {
+                            authorize_operation(&self.session_context, &S3Action::DeleteObject, bucket, Some(&obj_key))
+                                .await
+                                .map_err(|_| FsError::Forbidden)?;
+
                             let _ = self
                                 .storage
                                 .delete_object(
@@ -1370,6 +1381,14 @@ where
                     .await
                     .map_err(|_| FsError::Forbidden)?;
 
+                // SECURITY: clearing s3:DeleteObject on the directory marker key
+                // says nothing about the children stored under it. Enumerating the
+                // prefix and deleting each child are separate authorization
+                // boundaries and must be cleared on their own.
+                authorize_operation(&self.session_context, &S3Action::ListBucket, &bucket, Some(&prefix_with_slash))
+                    .await
+                    .map_err(|_| FsError::Forbidden)?;
+
                 // List and delete all objects with this prefix
                 let mut continuation_token = None;
                 loop {
@@ -1395,6 +1414,10 @@ where
                         if let Some(objects) = output.contents {
                             for obj in objects {
                                 if let Some(obj_key) = obj.key {
+                                    authorize_operation(&self.session_context, &S3Action::DeleteObject, &bucket, Some(&obj_key))
+                                        .await
+                                        .map_err(|_| FsError::Forbidden)?;
+
                                     let _ = self
                                         .storage
                                         .delete_object(
@@ -1692,11 +1715,12 @@ where
 mod tests {
     use super::WebDavDriver;
     use crate::common::client::s3::StorageBackend as S3StorageBackend;
+    use crate::common::gateway::{S3Action, with_test_auth_override};
     use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext};
     use async_trait::async_trait;
     use bytes::Bytes;
     use dav_server::davpath::DavPath;
-    use dav_server::fs::FsError;
+    use dav_server::fs::{DavFileSystem, FsError};
     use futures_util::StreamExt;
     use percent_encoding::{NON_ALPHANUMERIC, utf8_percent_encode};
     use rustfs_credentials::Credentials;
@@ -1887,6 +1911,7 @@ mod tests {
         objects: HashMap<(String, String), Vec<u8>>,
         put_keys: Vec<String>,
         delete_keys: Vec<String>,
+        deleted_buckets: Vec<String>,
         fail_delete_keys: HashSet<String>,
     }
 
@@ -2006,11 +2031,34 @@ mod tests {
 
         async fn list_objects_v2(
             &self,
-            _input: ListObjectsV2Input,
+            input: ListObjectsV2Input,
             _access_key: &str,
             _secret_key: &str,
         ) -> Result<ListObjectsV2Output, Self::Error> {
-            unreachable!("list_objects_v2 is not used in rename regression tests")
+            let prefix = input.prefix.map(|p| p.to_string()).unwrap_or_default();
+            let mut keys: Vec<String> = self
+                .state
+                .lock()
+                .expect("recording storage lock poisoned")
+                .objects
+                .keys()
+                .filter(|(bucket, key)| *bucket == input.bucket && key.starts_with(&prefix))
+                .map(|(_, key)| key.clone())
+                .collect();
+            keys.sort();
+
+            Ok(ListObjectsV2Output {
+                contents: Some(
+                    keys.into_iter()
+                        .map(|key| Object {
+                            key: Some(ObjectKey::from(key)),
+                            ..Default::default()
+                        })
+                        .collect(),
+                ),
+                is_truncated: Some(false),
+                ..Default::default()
+            })
         }
 
         async fn list_buckets(&self, _access_key: &str, _secret_key: &str) -> Result<ListBucketsOutput, Self::Error> {
@@ -2028,11 +2076,16 @@ mod tests {
 
         async fn delete_bucket(
             &self,
-            _bucket: &str,
+            bucket: &str,
             _access_key: &str,
             _secret_key: &str,
         ) -> Result<DeleteBucketOutput, Self::Error> {
-            unreachable!("delete_bucket is not used in rename regression tests")
+            self.state
+                .lock()
+                .expect("recording storage lock poisoned")
+                .deleted_buckets
+                .push(bucket.to_string());
+            Ok(DeleteBucketOutput::default())
         }
 
         async fn copy_object(
@@ -2248,6 +2301,63 @@ mod tests {
         }
     }
 
+    /// A bucket DELETE wipes every object in the bucket, so `s3:DeleteBucket`
+    /// alone must not be enough: each object needs its own `s3:DeleteObject`
+    /// boundary. The backend would happily complete the whole recursive delete
+    /// if the per-object check were removed again.
+    #[tokio::test]
+    async fn bucket_delete_denied_per_object_leaves_contents_and_bucket_intact() {
+        let (driver, storage) = recording_driver(&[("bucket", "secret.txt", b"secret")], &[]);
+        let path = DavPath::new("/bucket/").expect("path should parse");
+
+        let err = with_test_auth_override(
+            |action, _bucket, _object| !matches!(action, S3Action::DeleteObject),
+            driver.remove_dir(&path),
+        )
+        .await
+        .expect_err("bucket DELETE must fail closed when s3:DeleteObject is denied for a bucket member");
+
+        assert_eq!(err, FsError::Forbidden);
+
+        let state = storage.state.lock().expect("recording storage lock poisoned");
+        assert!(state.delete_keys.is_empty(), "no object may be deleted once the deny lands");
+        assert!(
+            state.deleted_buckets.is_empty(),
+            "the bucket must survive when its contents could not be authorized for deletion"
+        );
+        assert!(state.objects.contains_key(&("bucket".to_string(), "secret.txt".to_string())));
+    }
+
+    /// A directory DELETE authorizes the `dir/` marker key, but the children
+    /// stored under that prefix are separate resources and each needs its own
+    /// `s3:DeleteObject` boundary.
+    #[tokio::test]
+    async fn directory_delete_denied_for_child_leaves_child_intact() {
+        let (driver, storage) = recording_driver(&[("bucket", "dir/child.txt", b"child")], &[]);
+        let path = DavPath::new("/bucket/dir/").expect("path should parse");
+
+        let err = with_test_auth_override(
+            |action, _bucket, object| !matches!((action, object), (S3Action::DeleteObject, Some("dir/child.txt"))),
+            driver.remove_dir(&path),
+        )
+        .await
+        .expect_err("directory DELETE must fail closed when a child object denies s3:DeleteObject");
+
+        assert_eq!(err, FsError::Forbidden);
+
+        let state = storage.state.lock().expect("recording storage lock poisoned");
+        assert!(
+            state.delete_keys.is_empty(),
+            "the denied child must not be deleted, got {:?}",
+            state.delete_keys
+        );
+        assert!(
+            state
+                .objects
+                .contains_key(&("bucket".to_string(), "dir/child.txt".to_string()))
+        );
+    }
+
     #[tokio::test]
     async fn directory_rename_returns_error_when_delete_fails_after_successful_copy() {
         let (driver, storage) = recording_driver(
diff --git a/crates/protocols/src/webdav/server.rs b/crates/protocols/src/webdav/server.rs
index 2023207db..975eb3cb8 100644
--- a/crates/protocols/src/webdav/server.rs
+++ b/crates/protocols/src/webdav/server.rs
@@ -15,26 +15,30 @@
 use super::config::{WebDavConfig, WebDavInitError};
 use super::driver::WebDavDriver;
 use crate::common::client::s3::StorageBackend;
-use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext};
+use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext, is_temporary_credential};
 use bytes::Bytes;
 use dav_server::DavHandler;
 use dav_server::fakels::FakeLs;
-use http_body_util::{BodyExt, Full};
+use http_body_util::{BodyExt, Full, LengthLimitError, Limited};
+use hyper::body::Body as HttpBody;
 use hyper::server::conn::http1;
 use hyper::service::service_fn;
 use hyper::{Request, Response, StatusCode};
-use hyper_util::rt::TokioIo;
+use hyper_util::rt::{TokioIo, TokioTimer};
 use rustfs_config::{DEFAULT_TLS_RELOAD_ENABLE, DEFAULT_TLS_RELOAD_INTERVAL, ENV_TLS_RELOAD_ENABLE, ENV_TLS_RELOAD_INTERVAL};
 use rustfs_tls_runtime::{ReloadableServerCertResolver, TlsReloadOptions, spawn_server_cert_reload_loop};
 use rustfs_utils::MaskedAccessKey;
 use rustls::ServerConfig;
 use std::convert::Infallible;
+use std::io;
 use std::net::IpAddr;
+use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
 use subtle::ConstantTimeEq;
 use tokio::net::TcpListener;
-use tokio::sync::{broadcast, watch};
+use tokio::sync::{Semaphore, broadcast, watch};
+use tokio::time::timeout;
 use tokio_rustls::TlsAcceptor;
 use tracing::{Instrument, debug, error, info, info_span, warn};
 
@@ -47,6 +51,13 @@ const EVENT_WEBDAV_CONNECTION_STATE: &str = "webdav_connection_state";
 const EVENT_WEBDAV_REQUEST_VALIDATION_FAILED: &str = "webdav_request_validation_failed";
 const EVENT_WEBDAV_REQUEST_BODY_FAILED: &str = "webdav_request_body_failed";
 const EVENT_WEBDAV_AUTH_STATE: &str = "webdav_auth_state";
+const EVENT_WEBDAV_CONNECTION_CAP_STATE: &str = "webdav_connection_cap_state";
+
+/// Response body handed back to Hyper.
+///
+/// Boxed instead of collected: buffering the dav-server body would
+/// materialise a whole object in memory for every GET.
+type WebDavBody = Pin<Box<dyn HttpBody<Data = Bytes, Error = io::Error> + Send>>;
 
 /// WebDAV server implementation
 pub struct WebDavServer<S>
@@ -78,7 +89,7 @@ where
     }
 
     /// Start the WebDAV server
-    pub async fn start(&self, mut shutdown_rx: broadcast::Receiver<()>) -> Result<(), WebDavInitError> {
+    pub async fn start(&self, shutdown_rx: broadcast::Receiver<()>) -> Result<(), WebDavInitError> {
         info!(
             event = EVENT_WEBDAV_SERVER_STATE,
             component = LOG_COMPONENT_PROTOCOLS,
@@ -87,10 +98,16 @@ where
             bind_addr = %self.config.bind_addr,
             tls_enabled = self.config.tls_enabled,
             max_body_size = self.config.max_body_size,
+            max_connections = self.config.max_connections,
             "WebDAV server starting"
         );
 
         let listener = TcpListener::bind(self.config.bind_addr).await?;
+        self.serve(listener, shutdown_rx).await
+    }
+
+    /// Serve connections from an already bound listener
+    async fn serve(&self, listener: TcpListener, mut shutdown_rx: broadcast::Receiver<()>) -> Result<(), WebDavInitError> {
         info!(
             event = EVENT_WEBDAV_SERVER_STATE,
             component = LOG_COMPONENT_PROTOCOLS,
@@ -135,8 +152,47 @@ where
         };
 
         let storage = self.storage.clone();
+        let request_timeout = Duration::from_secs(self.config.request_timeout_secs);
+        let connection_limiter = Arc::new(Semaphore::new(self.config.max_connections.min(Semaphore::MAX_PERMITS)));
 
         loop {
+            // Admission control: hold a permit before accepting, so at
+            // saturation the kernel backlog absorbs the burst instead of the
+            // process spawning an unbounded number of connection tasks. The
+            // permit travels into the task and is released when it ends.
+            let permit = match connection_limiter.clone().try_acquire_owned() {
+                Ok(permit) => permit,
+                Err(_) => {
+                    debug!(
+                        event = EVENT_WEBDAV_CONNECTION_CAP_STATE,
+                        component = LOG_COMPONENT_PROTOCOLS,
+                        subsystem = LOG_SUBSYSTEM_WEBDAV_SERVER,
+                        state = "saturated",
+                        max_connections = self.config.max_connections,
+                        "WebDAV connection cap saturated"
+                    );
+                    tokio::select! {
+                        permit = connection_limiter.clone().acquire_owned() => match permit {
+                            Ok(permit) => permit,
+                            // The semaphore is never closed; fail safe by
+                            // stopping the accept loop if it ever is.
+                            Err(_) => break,
+                        },
+                        _ = shutdown_rx.recv() => {
+                            info!(
+                                event = EVENT_WEBDAV_SERVER_STATE,
+                                component = LOG_COMPONENT_PROTOCOLS,
+                                subsystem = LOG_SUBSYSTEM_WEBDAV_SERVER,
+                                state = "shutdown_requested",
+                                "WebDAV shutdown requested"
+                            );
+                            let _ = reload_shutdown_tx.send(true);
+                            break;
+                        }
+                    }
+                }
+            };
+
             tokio::select! {
                 accept_result = listener.accept() => {
                     match accept_result {
@@ -153,11 +209,14 @@ where
                             );
                             tokio::spawn(
                                 async move {
+                                    let _permit = permit;
                                     if let Some(acceptor) = tls_acceptor {
-                                        match acceptor.accept(stream).await {
-                                            Ok(tls_stream) => {
+                                        // A handshake that never completes would otherwise
+                                        // hold its connection permit forever.
+                                        match timeout(request_timeout, acceptor.accept(stream)).await {
+                                            Ok(Ok(tls_stream)) => {
                                                 let io = TokioIo::new(tls_stream);
-                                                if let Err(e) = Self::handle_connection_impl(io, storage, source_ip, max_body_size).await {
+                                                if let Err(e) = Self::handle_connection_impl(io, storage, source_ip, max_body_size, request_timeout).await {
                                                     debug!(
                                                         event = EVENT_WEBDAV_CONNECTION_STATE,
                                                         component = LOG_COMPONENT_PROTOCOLS,
@@ -170,7 +229,7 @@ where
                                                     );
                                                 }
                                             }
-                                            Err(e) => {
+                                            Ok(Err(e)) => {
                                                 debug!(
                                                     event = EVENT_WEBDAV_CONNECTION_STATE,
                                                     component = LOG_COMPONENT_PROTOCOLS,
@@ -181,10 +240,21 @@ where
                                                     "webdav connection ended with error"
                                                 );
                                             }
+                                            Err(_) => {
+                                                debug!(
+                                                    event = EVENT_WEBDAV_CONNECTION_STATE,
+                                                    component = LOG_COMPONENT_PROTOCOLS,
+                                                    subsystem = LOG_SUBSYSTEM_WEBDAV_SERVER,
+                                                    result = "tls_handshake_timeout",
+                                                    peer = %source_ip,
+                                                    timeout_secs = request_timeout.as_secs(),
+                                                    "webdav connection ended with error"
+                                                );
+                                            }
                                         }
                                     } else {
                                         let io = TokioIo::new(stream);
-                                        if let Err(e) = Self::handle_connection_impl(io, storage, source_ip, max_body_size).await {
+                                        if let Err(e) = Self::handle_connection_impl(io, storage, source_ip, max_body_size, request_timeout).await {
                                             debug!(
                                                 event = EVENT_WEBDAV_CONNECTION_STATE,
                                                 component = LOG_COMPONENT_PROTOCOLS,
@@ -244,16 +314,24 @@ where
         storage: S,
         source_ip: IpAddr,
         max_body_size: u64,
+        request_timeout: Duration,
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync>>
     where
         I: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
     {
         let service = service_fn(move |req: Request<hyper::body::Incoming>| {
             let storage = storage.clone();
-            async move { Self::handle_request(req, storage, source_ip, max_body_size).await }
+            async move { Self::handle_request(req, storage, source_ip, max_body_size, request_timeout).await }
         });
 
-        http1::Builder::new().serve_connection(io, service).await?;
+        // A peer that opens a connection and dribbles (or never finishes)
+        // request headers is disconnected once the configured request
+        // timeout elapses. The timer is required for the deadline to apply.
+        http1::Builder::new()
+            .timer(TokioTimer::new())
+            .header_read_timeout(request_timeout)
+            .serve_connection(io, service)
+            .await?;
 
         Ok(())
     }
@@ -264,8 +342,12 @@ where
         storage: S,
         source_ip: IpAddr,
         max_body_size: u64,
-    ) -> Result<Response<Full<Bytes>>, Infallible> {
-        // Check Content-Length against max_body_size before reading body
+        request_timeout: Duration,
+    ) -> Result<Response<WebDavBody>, Infallible> {
+        // Advisory fast path only: a declared Content-Length lets an
+        // oversized request be rejected before any body is read. The
+        // authoritative limit is enforced in `dispatch_dav` against the
+        // bytes actually received, which a chunked request cannot dodge.
         if let Some(content_length) = req.headers().get("content-length")
             && let Ok(length_str) = content_length.to_str()
             && let Ok(length) = length_str.parse::<u64>()
@@ -324,49 +406,7 @@ where
             .locksystem(FakeLs::new())
             .build_handler();
 
-        // Convert request body
-        let (parts, body) = req.into_parts();
-        let body_bytes = match body.collect().await {
-            Ok(collected) => collected.to_bytes(),
-            Err(e) => {
-                error!(
-                    event = EVENT_WEBDAV_REQUEST_BODY_FAILED,
-                    component = LOG_COMPONENT_PROTOCOLS,
-                    subsystem = LOG_SUBSYSTEM_WEBDAV_SERVER,
-                    result = "request_body_read_failed",
-                    source_ip = %source_ip,
-                    error = %e,
-                    "webdav request body failed"
-                );
-                return Ok(error_response(StatusCode::BAD_REQUEST, "Failed to read request body"));
-            }
-        };
-
-        // Create request for dav-server using Bytes
-        let dav_req = Request::from_parts(parts, dav_server::body::Body::from(body_bytes));
-
-        // Handle the request
-        let dav_resp = dav_handler.handle(dav_req).await;
-
-        // Convert response
-        let (parts, body) = dav_resp.into_parts();
-        let body_bytes = match body.collect().await {
-            Ok(collected) => collected.to_bytes(),
-            Err(e) => {
-                error!(
-                    event = EVENT_WEBDAV_REQUEST_BODY_FAILED,
-                    component = LOG_COMPONENT_PROTOCOLS,
-                    subsystem = LOG_SUBSYSTEM_WEBDAV_SERVER,
-                    result = "response_body_read_failed",
-                    source_ip = %source_ip,
-                    error = %e,
-                    "webdav request body failed"
-                );
-                return Ok(error_response(StatusCode::INTERNAL_SERVER_ERROR, "Internal server error"));
-            }
-        };
-
-        Ok(Response::from_parts(parts, Full::new(body_bytes)))
+        Ok(dispatch_dav(req, dav_handler, source_ip, max_body_size, request_timeout).await)
     }
 
     /// Authenticate user against IAM system
@@ -442,6 +482,19 @@ where
             WebDavInitError::Server("User not found".to_string())
         })?;
 
+        if is_temporary_credential(&identity.credentials) {
+            warn!(
+                event = EVENT_WEBDAV_AUTH_STATE,
+                component = LOG_COMPONENT_PROTOCOLS,
+                subsystem = LOG_SUBSYSTEM_WEBDAV_AUTH,
+                result = "temporary_credential_rejected",
+                source_ip = %source_ip,
+                access_key = %masked_access_key,
+                "WebDAV auth rejected temporary credential"
+            );
+            return Err(WebDavInitError::Server("Invalid credentials".to_string()));
+        }
+
         // Constant-time secret comparison to prevent timing side-channel
         // attacks. Same primitive used by the SFTP handler and rustfs/src/auth.rs.
         let secret_matches: bool = identity
@@ -491,21 +544,114 @@ where
     }
 }
 
+/// Read the request body and run it through the dav handler.
+///
+/// Both limits configured for the listener are enforced here: the body is
+/// truncated at `max_body_size` bytes actually received (a chunked upload
+/// declares no length, so a header check cannot bound it), and neither the
+/// body read nor the dav handler may run past `request_timeout`.
+async fn dispatch_dav<B>(
+    req: Request<B>,
+    dav_handler: DavHandler,
+    source_ip: IpAddr,
+    max_body_size: u64,
+    request_timeout: Duration,
+) -> Response<WebDavBody>
+where
+    B: HttpBody<Data = Bytes> + Send + 'static,
+    B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
+{
+    let (parts, body) = req.into_parts();
+    let limit = usize::try_from(max_body_size).unwrap_or(usize::MAX);
+
+    let body_bytes = match timeout(request_timeout, Limited::new(body, limit).collect()).await {
+        Ok(Ok(collected)) => collected.to_bytes(),
+        Ok(Err(e)) if e.downcast_ref::<LengthLimitError>().is_some() => {
+            warn!(
+                event = EVENT_WEBDAV_REQUEST_VALIDATION_FAILED,
+                component = LOG_COMPONENT_PROTOCOLS,
+                subsystem = LOG_SUBSYSTEM_WEBDAV_SERVER,
+                result = "payload_too_large",
+                max_body_size,
+                source_ip = %source_ip,
+                "webdav request validation failed"
+            );
+            return error_response(
+                StatusCode::PAYLOAD_TOO_LARGE,
+                &format!("Request body too large. Maximum size is {} bytes", max_body_size),
+            );
+        }
+        Ok(Err(e)) => {
+            error!(
+                event = EVENT_WEBDAV_REQUEST_BODY_FAILED,
+                component = LOG_COMPONENT_PROTOCOLS,
+                subsystem = LOG_SUBSYSTEM_WEBDAV_SERVER,
+                result = "request_body_read_failed",
+                source_ip = %source_ip,
+                error = %e,
+                "webdav request body failed"
+            );
+            return error_response(StatusCode::BAD_REQUEST, "Failed to read request body");
+        }
+        Err(_) => {
+            warn!(
+                event = EVENT_WEBDAV_REQUEST_VALIDATION_FAILED,
+                component = LOG_COMPONENT_PROTOCOLS,
+                subsystem = LOG_SUBSYSTEM_WEBDAV_SERVER,
+                result = "request_body_read_timeout",
+                timeout_secs = request_timeout.as_secs(),
+                source_ip = %source_ip,
+                "webdav request validation failed"
+            );
+            return error_response(StatusCode::REQUEST_TIMEOUT, "Request timed out");
+        }
+    };
+
+    // Create request for dav-server using Bytes
+    let dav_req = Request::from_parts(parts, dav_server::body::Body::from(body_bytes));
+
+    let dav_resp = match timeout(request_timeout, dav_handler.handle(dav_req)).await {
+        Ok(resp) => resp,
+        Err(_) => {
+            warn!(
+                event = EVENT_WEBDAV_REQUEST_VALIDATION_FAILED,
+                component = LOG_COMPONENT_PROTOCOLS,
+                subsystem = LOG_SUBSYSTEM_WEBDAV_SERVER,
+                result = "request_handling_timeout",
+                timeout_secs = request_timeout.as_secs(),
+                source_ip = %source_ip,
+                "webdav request validation failed"
+            );
+            return error_response(StatusCode::REQUEST_TIMEOUT, "Request timed out");
+        }
+    };
+
+    // Streamed straight to Hyper: collecting here would hold the whole
+    // object in memory for the duration of a GET.
+    let (parts, body) = dav_resp.into_parts();
+    Response::from_parts(parts, Box::pin(body) as WebDavBody)
+}
+
 /// Create unauthorized response with WWW-Authenticate header
-fn unauthorized_response() -> Response<Full<Bytes>> {
+fn unauthorized_response() -> Response<WebDavBody> {
     Response::builder()
         .status(StatusCode::UNAUTHORIZED)
         .header("WWW-Authenticate", "Basic realm=\"RustFS WebDAV\"")
-        .body(Full::new(Bytes::from("Unauthorized")))
-        .unwrap_or_else(|_| Response::new(Full::new(Bytes::from("Unauthorized"))))
+        .body(fixed_body("Unauthorized"))
+        .unwrap_or_else(|_| Response::new(fixed_body("Unauthorized")))
 }
 
 /// Create error response
-fn error_response(status: StatusCode, message: &str) -> Response<Full<Bytes>> {
+fn error_response(status: StatusCode, message: &str) -> Response<WebDavBody> {
     Response::builder()
         .status(status)
-        .body(Full::new(Bytes::from(message.to_string())))
-        .unwrap_or_else(|_| Response::new(Full::new(Bytes::from("Internal Server Error"))))
+        .body(fixed_body(message.to_string()))
+        .unwrap_or_else(|_| Response::new(fixed_body("Internal Server Error")))
+}
+
+/// Wrap a fixed byte payload in the streaming response body type
+fn fixed_body(message: impl Into<Bytes>) -> WebDavBody {
+    Box::pin(Full::new(message.into()).map_err(|never| match never {}))
 }
 
 /// Decode base64 string
@@ -513,3 +659,380 @@ fn base64_decode(encoded: &str) -> Result<Vec<u8>, ()> {
     use base64::Engine;
     base64::engine::general_purpose::STANDARD.decode(encoded).map_err(|_| ())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::common::client::s3::StorageBackend;
+    use async_trait::async_trait;
+    use dav_server::memfs::MemFs;
+    use futures_util::stream;
+    use http_body_util::StreamBody;
+    use hyper::body::Frame;
+    use s3s::dto::*;
+    use std::fmt::{Debug, Formatter};
+    use std::net::{Ipv4Addr, SocketAddr};
+    use std::task::{Context, Poll};
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpStream;
+
+    const TEST_IP: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST);
+
+    /// Storage double for the connection-level tests. Those requests are
+    /// rejected before authentication succeeds, so storage is never reached.
+    #[derive(Clone)]
+    struct StubStorage;
+
+    impl Debug for StubStorage {
+        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+            f.write_str("StubStorage")
+        }
+    }
+
+    #[async_trait]
+    impl StorageBackend for StubStorage {
+        type Error = std::io::Error;
+
+        async fn get_object(
+            &self,
+            _bucket: &str,
+            _key: &str,
+            _access_key: &str,
+            _secret_key: &str,
+            _start_pos: Option<u64>,
+        ) -> Result<GetObjectOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn get_object_range(
+            &self,
+            _bucket: &str,
+            _key: &str,
+            _access_key: &str,
+            _secret_key: &str,
+            _start_pos: u64,
+            _length: u64,
+        ) -> Result<GetObjectOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn put_object(
+            &self,
+            _input: PutObjectInput,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<PutObjectOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn delete_object(
+            &self,
+            _bucket: &str,
+            _key: &str,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<DeleteObjectOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn head_object(
+            &self,
+            _bucket: &str,
+            _key: &str,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<HeadObjectOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn head_bucket(
+            &self,
+            _bucket: &str,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<HeadBucketOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn list_objects_v2(
+            &self,
+            _input: ListObjectsV2Input,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<ListObjectsV2Output, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn list_buckets(&self, _access_key: &str, _secret_key: &str) -> Result<ListBucketsOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn create_bucket(
+            &self,
+            _bucket: &str,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<CreateBucketOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn delete_bucket(
+            &self,
+            _bucket: &str,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<DeleteBucketOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn copy_object(
+            &self,
+            _input: CopyObjectInput,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<CopyObjectOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn create_multipart_upload(
+            &self,
+            _input: CreateMultipartUploadInput,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<CreateMultipartUploadOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn upload_part(
+            &self,
+            _input: UploadPartInput,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<UploadPartOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn complete_multipart_upload(
+            &self,
+            _input: CompleteMultipartUploadInput,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<CompleteMultipartUploadOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn abort_multipart_upload(
+            &self,
+            _input: AbortMultipartUploadInput,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<AbortMultipartUploadOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+
+        async fn upload_part_copy(
+            &self,
+            _input: UploadPartCopyInput,
+            _access_key: &str,
+            _secret_key: &str,
+        ) -> Result<UploadPartCopyOutput, Self::Error> {
+            unreachable!("connection tests should not hit storage")
+        }
+    }
+
+    /// Request body that never produces a frame, standing in for a peer that
+    /// opens a chunked upload and then stalls.
+    struct StalledBody;
+
+    impl HttpBody for StalledBody {
+        type Data = Bytes;
+        type Error = io::Error;
+
+        fn poll_frame(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Result<Frame<Bytes>, io::Error>>> {
+            Poll::Pending
+        }
+    }
+
+    fn memfs_handler() -> DavHandler {
+        DavHandler::builder()
+            .filesystem(MemFs::new())
+            .locksystem(FakeLs::new())
+            .build_handler()
+    }
+
+    /// Chunked upload: no Content-Length header, `chunks` frames of `chunk_len` bytes.
+    fn chunked_request(
+        chunks: usize,
+        chunk_len: usize,
+    ) -> Request<StreamBody<impl stream::Stream<Item = io::Result<Frame<Bytes>>>>> {
+        let frames: Vec<io::Result<Frame<Bytes>>> = (0..chunks)
+            .map(|_| Ok(Frame::data(Bytes::from(vec![b'a'; chunk_len]))))
+            .collect();
+        Request::builder()
+            .method("PUT")
+            .uri("/upload.bin")
+            .body(StreamBody::new(stream::iter(frames)))
+            .expect("build chunked request")
+    }
+
+    fn get_request(uri: &str) -> Request<Full<Bytes>> {
+        Request::builder()
+            .method("GET")
+            .uri(uri)
+            .body(Full::new(Bytes::new()))
+            .expect("build get request")
+    }
+
+    /// R03-CAN-051 / R03-CAN-067 / R05-CAN-094: a chunked upload declares no
+    /// Content-Length, so the limit has to hold on the bytes actually read.
+    #[tokio::test]
+    async fn chunked_upload_over_max_body_size_is_rejected() {
+        let handler = memfs_handler();
+
+        let resp = dispatch_dav(chunked_request(8, 8), handler.clone(), TEST_IP, 16, Duration::from_secs(30)).await;
+        assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE);
+
+        // The oversized body must not have reached the filesystem.
+        let stored = dispatch_dav(get_request("/upload.bin"), handler, TEST_IP, 16, Duration::from_secs(30)).await;
+        assert_eq!(stored.status(), StatusCode::NOT_FOUND);
+    }
+
+    #[tokio::test]
+    async fn chunked_upload_within_max_body_size_is_accepted() {
+        let handler = memfs_handler();
+
+        let resp = dispatch_dav(chunked_request(2, 4), handler.clone(), TEST_IP, 16, Duration::from_secs(30)).await;
+        assert_eq!(resp.status(), StatusCode::CREATED);
+
+        let stored = dispatch_dav(get_request("/upload.bin"), handler, TEST_IP, 16, Duration::from_secs(30)).await;
+        assert_eq!(stored.status(), StatusCode::OK);
+        let bytes = stored.into_body().collect().await.expect("collect body").to_bytes();
+        assert_eq!(bytes.len(), 8);
+    }
+
+    /// R04-CAN-089 / R05-CAN-094: a request whose body never arrives must be
+    /// cut off at the configured request timeout.
+    #[tokio::test(start_paused = true)]
+    async fn stalled_request_body_hits_request_timeout() {
+        let req = Request::builder()
+            .method("PUT")
+            .uri("/stalled.bin")
+            .body(StalledBody)
+            .expect("build stalled request");
+
+        let resp = timeout(
+            Duration::from_secs(600),
+            dispatch_dav(req, memfs_handler(), TEST_IP, 1024, Duration::from_secs(30)),
+        )
+        .await
+        .expect("stalled body was never cut off by the configured request timeout");
+
+        assert_eq!(resp.status(), StatusCode::REQUEST_TIMEOUT);
+    }
+
+    /// R03-CAN-052: the object body must reach Hyper as a stream. A collected
+    /// body reports an exact size hint; a streamed one does not.
+    #[tokio::test]
+    async fn get_response_body_is_streamed_not_buffered() {
+        let handler = memfs_handler();
+        let put = dispatch_dav(chunked_request(4, 4), handler.clone(), TEST_IP, 1024, Duration::from_secs(30)).await;
+        assert_eq!(put.status(), StatusCode::CREATED);
+
+        let resp = dispatch_dav(get_request("/upload.bin"), handler, TEST_IP, 1024, Duration::from_secs(30)).await;
+        assert_eq!(resp.status(), StatusCode::OK);
+        assert!(
+            resp.body().size_hint().upper().is_none(),
+            "response body was collected into memory instead of streamed"
+        );
+
+        let bytes = resp.into_body().collect().await.expect("collect body").to_bytes();
+        assert_eq!(bytes.len(), 16);
+    }
+
+    /// R04-CAN-089: a peer that opens a connection and never finishes its
+    /// request headers must be disconnected at the configured timeout.
+    #[tokio::test(start_paused = true)]
+    async fn slow_request_headers_hit_request_timeout() {
+        let (mut client, server) = tokio::io::duplex(1024);
+
+        let conn = tokio::spawn(WebDavServer::<StubStorage>::handle_connection_impl(
+            TokioIo::new(server),
+            StubStorage,
+            TEST_IP,
+            1024,
+            Duration::from_secs(30),
+        ));
+
+        client
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n")
+            .await
+            .expect("write partial headers");
+
+        let outcome = timeout(Duration::from_secs(600), conn).await;
+        assert!(outcome.is_ok(), "connection outlived the configured request timeout");
+    }
+
+    /// R05-CAN-097: the accept loop must not serve more connections at once
+    /// than the configured cap.
+    #[tokio::test]
+    async fn accept_loop_is_bounded_by_max_connections() {
+        let config = WebDavConfig {
+            bind_addr: SocketAddr::from((Ipv4Addr::LOCALHOST, 0)),
+            tls_enabled: false,
+            cert_dir: None,
+            ca_file: None,
+            max_body_size: 1024,
+            request_timeout_secs: 30,
+            max_connections: 1,
+        };
+        let server = WebDavServer::new(config, StubStorage).await.expect("build server");
+
+        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.expect("bind listener");
+        let addr = listener.local_addr().expect("listener addr");
+        let (shutdown_tx, shutdown_rx) = broadcast::channel(1);
+        let serving = tokio::spawn(async move { server.serve(listener, shutdown_rx).await });
+
+        // The first client occupies the single permit; keep-alive holds it
+        // after the response.
+        let mut first = TcpStream::connect(addr).await.expect("connect first");
+        first
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .expect("write first");
+        let mut buf = [0u8; 64];
+        let read = first.read(&mut buf).await.expect("read first");
+        assert!(read > 0, "first connection was not served");
+
+        // The second client lands in the backlog and must not be served.
+        let mut second = TcpStream::connect(addr).await.expect("connect second");
+        second
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .expect("write second");
+        let blocked = timeout(Duration::from_millis(500), second.read(&mut buf)).await;
+        assert!(blocked.is_err(), "second connection was served while the cap was saturated");
+
+        // Releasing the first permit admits the queued connection.
+        drop(first);
+        let served = timeout(Duration::from_secs(10), second.read(&mut buf))
+            .await
+            .expect("second connection was never served")
+            .expect("read second");
+        assert!(served > 0, "second connection returned no data");
+
+        let _ = shutdown_tx.send(());
+        let _ = timeout(Duration::from_secs(10), serving).await;
+    }
+
+    #[tokio::test]
+    async fn config_rejects_zero_max_connections() {
+        let config = WebDavConfig {
+            tls_enabled: false,
+            max_connections: 0,
+            ..WebDavConfig::default()
+        };
+
+        let err = config.validate().await.expect_err("zero max_connections must be rejected");
+        assert!(matches!(err, WebDavInitError::InvalidConfig(_)), "unexpected error: {err}");
+    }
+}
diff --git a/crates/signer/Cargo.toml b/crates/signer/Cargo.toml
index ad52a4807..24cb667eb 100644
--- a/crates/signer/Cargo.toml
+++ b/crates/signer/Cargo.toml
@@ -37,6 +37,9 @@ s3s = { workspace = true, features = ["minio"] }
 base64-simd.workspace = true
 thiserror.workspace = true
 
+[dev-dependencies]
+tracing-subscriber = { workspace = true, features = ["fmt"] }
+
 [lints]
 workspace = true
 
diff --git a/crates/signer/src/request_signature_v4.rs b/crates/signer/src/request_signature_v4.rs
index fc88b608b..4e49c7215 100644
--- a/crates/signer/src/request_signature_v4.rs
+++ b/crates/signer/src/request_signature_v4.rs
@@ -20,7 +20,7 @@ use std::collections::{HashMap, HashSet};
 use std::fmt::Write;
 use std::sync::LazyLock;
 use time::{OffsetDateTime, macros::format_description};
-use tracing::{debug, warn};
+use tracing::warn;
 
 use super::constants::UNSIGNED_PAYLOAD;
 use super::request_signature_streaming_unsigned_trailer::streaming_unsigned_v4;
@@ -135,6 +135,8 @@ fn try_get_hashed_payload(req: &request::Request<Body>) -> SignResult<String> {
     Ok(hashed_payload.to_string())
 }
 
+/// The headers signed here carry credential material (`x-amz-security-token`, SSE-C keys),
+/// so neither the names nor the values may be written to any log sink.
 fn try_get_canonical_headers(req: &request::Request<Body>, ignored_headers: &HashSet<&'static str>) -> SignResult<String> {
     let mut headers = Vec::<String>::new();
     let mut vals = HashMap::<String, Vec<String>>::new();
@@ -166,9 +168,6 @@ fn try_get_canonical_headers(req: &request::Request<Body>, ignored_headers: &Has
     }
     headers.sort();
 
-    debug!("get_canonical_headers vals: {:?}", vals);
-    debug!("get_canonical_headers headers: {:?}", headers);
-
     let mut buf = BytesMut::new();
     for k in headers {
         let _ = buf.write_str(&k);
@@ -216,7 +215,6 @@ fn header_exists(key: &str, headers: &[String]) -> bool {
 fn get_signed_headers(req: &request::Request<Body>, ignored_headers: &HashSet<&'static str>) -> String {
     let mut headers = Vec::<String>::new();
     let headers_ref = req.headers();
-    debug!("get_signed_headers headers: {:?}", headers_ref);
     for (k, _) in headers_ref {
         if ignored_headers.contains(k.as_str()) {
             continue;
@@ -1237,4 +1235,70 @@ mod tests {
         let t = datetime!(0001-01-02 03:04:05 UTC);
         assert_eq!(format_yyyymmdd(t), "00010102");
     }
+
+    #[derive(Clone, Default)]
+    struct CapturedLogs {
+        output: std::sync::Arc<std::sync::Mutex<Vec<u8>>>,
+    }
+
+    struct CapturedWriter(CapturedLogs);
+
+    impl std::io::Write for CapturedWriter {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            self.0.output.lock().expect("log buffer lock poisoned").extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturedLogs {
+        type Writer = CapturedWriter;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            CapturedWriter(self.clone())
+        }
+    }
+
+    impl CapturedLogs {
+        fn output(&self) -> String {
+            String::from_utf8(self.output.lock().expect("log buffer lock poisoned").clone()).expect("logs should be UTF-8")
+        }
+    }
+
+    #[test]
+    fn signing_never_logs_signed_header_material() {
+        const MARKER: &str = "CS_TEST_SECRET_DO_NOT_LOG";
+
+        let mut req = request::Request::builder()
+            .method(http::Method::GET)
+            .uri("http://examplebucket.s3.amazonaws.com/object")
+            .body(Body::empty())
+            .expect("test request should build");
+        let headers = req.headers_mut();
+        headers.insert("host", HeaderValue::from_static("examplebucket.s3.amazonaws.com"));
+        headers.insert("x-amz-security-token", HeaderValue::from_static(MARKER));
+        headers.insert("x-amz-server-side-encryption-customer-key", HeaderValue::from_static(MARKER));
+
+        let logs = CapturedLogs::default();
+        let subscriber = tracing_subscriber::fmt()
+            .with_max_level(tracing::Level::TRACE)
+            .with_writer(logs.clone())
+            .finish();
+
+        let (canonical_headers, signed_headers) = tracing::subscriber::with_default(subscriber, || {
+            let canonical_headers = try_get_canonical_headers(&req, &V4_IGNORED_HEADERS).expect("request should canonicalize");
+            let signed_headers = get_signed_headers(&req, &V4_IGNORED_HEADERS);
+            (canonical_headers, signed_headers)
+        });
+
+        assert!(canonical_headers.contains(MARKER), "the header must still take part in signing");
+        assert!(signed_headers.contains("x-amz-security-token"));
+
+        let output = logs.output();
+        assert!(!output.contains(MARKER), "signed header value leaked into logs: {output}");
+        assert!(!output.contains("x-amz-security-token"), "signed header name leaked into logs: {output}");
+    }
 }
diff --git a/helm/README.md b/helm/README.md
index 8cd458917..07df2792d 100644
--- a/helm/README.md
+++ b/helm/README.md
@@ -104,7 +104,7 @@ set both `replicaCount` and `drivesPerNode` explicitly.
 | config.rustfs.kms.type | string | `vault`| The kms type that RustFS supported. |
 | config.rustfs.kms.vault.vault_backend | string | `""`| The vault backend, `vault-kv2` or `vault-transit`. |
 | config.rustfs.kms.vault.vault_address | string | `""`| The vault address. |
-| config.rustfs.kms.vault.vault_token | string | `""`| The vault token. |
+| config.rustfs.kms.vault.vault_token | string | `""`| The vault token. Rendered into a dedicated Secret (`<fullname>-kms-secret`), never into the ConfigMap. |
 | config.rustfs.kms.vault.vault_mount_path | string | `"transit"`| The vault mount path, only works if `vault_backend` equals `vault-transit` . |
 | config.rustfs.kms.vault.default_key | string | `"transit"`| The master key id for RustFS. |
 | extraEnv | map | `[]` |  Extra environment variables for RustFS container. |
diff --git a/helm/rustfs/templates/_helpers.tpl b/helm/rustfs/templates/_helpers.tpl
index 6088ecae3..f9156b834 100644
--- a/helm/rustfs/templates/_helpers.tpl
+++ b/helm/rustfs/templates/_helpers.tpl
@@ -85,6 +85,27 @@ Return the secret name
 {{- end }}
 {{- end }}
 
+{{/*
+Return the name of the Secret holding the Vault KMS token.
+The token is a credential, so it never belongs in the config ConfigMap. It also lives in
+its own Secret rather than in "rustfs.secretName", which may point at an operator-owned
+existingSecret that the chart must not assume contains a KMS key.
+*/}}
+{{- define "rustfs.kmsSecretName" -}}
+{{- printf "%s-kms-secret" (include "rustfs.fullname" .) }}
+{{- end }}
+
+{{/*
+Return the configured Vault KMS token, or the empty string when KMS is disabled, uses a
+different backend type, or no token was supplied. Callers use emptiness to decide whether
+the KMS Secret is rendered and mounted.
+*/}}
+{{- define "rustfs.kmsVaultToken" -}}
+{{- if and .Values.config.rustfs.kms.enabled (eq .Values.config.rustfs.kms.type "vault") -}}
+{{- .Values.config.rustfs.kms.vault.vault_token | default "" -}}
+{{- end -}}
+{{- end }}
+
 {{/*
 Return image pull secret content
 */}}
diff --git a/helm/rustfs/templates/configmap.yaml b/helm/rustfs/templates/configmap.yaml
index cde61dd8e..ea7b50b47 100644
--- a/helm/rustfs/templates/configmap.yaml
+++ b/helm/rustfs/templates/configmap.yaml
@@ -137,7 +137,8 @@ data:
   RUSTFS_KMS_ENABLE: "true"
   RUSTFS_KMS_BACKEND: {{ .vault_backend | quote }}
   RUSTFS_KMS_VAULT_ADDRESS: {{ .vault_address | quote }}
-  RUSTFS_KMS_VAULT_TOKEN: {{ .vault_token | quote }}
+  {{- /* RUSTFS_KMS_VAULT_TOKEN is a credential and is rendered into the KMS Secret
+         (templates/secret.yaml), never into this ConfigMap. */}}
   RUSTFS_KMS_DEFAULT_KEY_ID: {{ .default_key | quote }}
         {{- if eq .vault_backend "vault-transit" }}
   RUSTFS_KMS_VAULT_MOUNT_PATH: {{ .vault_mount_path | quote }}
diff --git a/helm/rustfs/templates/deployment.yaml b/helm/rustfs/templates/deployment.yaml
index 3fc7bdf0d..39effeb24 100644
--- a/helm/rustfs/templates/deployment.yaml
+++ b/helm/rustfs/templates/deployment.yaml
@@ -115,6 +115,10 @@ spec:
                 name: {{ include "rustfs.fullname" . }}-config
             - secretRef:
                 name: {{ include "rustfs.secretName" . }}
+            {{- if include "rustfs.kmsVaultToken" . }}
+            - secretRef:
+                name: {{ include "rustfs.kmsSecretName" . }}
+            {{- end }}
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           {{- include "rustfs.probes" . | nindent 10 }}
diff --git a/helm/rustfs/templates/secret.yaml b/helm/rustfs/templates/secret.yaml
index 773530782..052b62795 100644
--- a/helm/rustfs/templates/secret.yaml
+++ b/helm/rustfs/templates/secret.yaml
@@ -31,6 +31,20 @@ data:
   RUSTFS_SECRET_KEY: {{ $secretKey | b64enc | quote }}
 {{- end }}
 
+---
+{{- if include "rustfs.kmsVaultToken" . }}
+apiVersion: v1
+kind: Secret
+metadata:
+  name: {{ include "rustfs.kmsSecretName" . }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    {{- toYaml .Values.commonLabels | nindent 4 }}
+type: Opaque
+data:
+  RUSTFS_KMS_VAULT_TOKEN: {{ include "rustfs.kmsVaultToken" . | b64enc | quote }}
+{{- end }}
+
 ---
 {{- if .Values.imageRegistryCredentials.enabled }}
 apiVersion: v1
diff --git a/helm/rustfs/templates/statefulset.yaml b/helm/rustfs/templates/statefulset.yaml
index 032395f60..3954ad386 100644
--- a/helm/rustfs/templates/statefulset.yaml
+++ b/helm/rustfs/templates/statefulset.yaml
@@ -219,6 +219,10 @@ spec:
                 name: {{ include "rustfs.fullname" $ }}-config
             - secretRef:
                 name: {{ include "rustfs.secretName" $ }}
+            {{- if include "rustfs.kmsVaultToken" $ }}
+            - secretRef:
+                name: {{ include "rustfs.kmsSecretName" $ }}
+            {{- end }}
           resources:
             {{- toYaml $.Values.resources | nindent 12 }}
           {{- include "rustfs.probes" $ | nindent 10 }}
diff --git a/helm/rustfs/values.yaml b/helm/rustfs/values.yaml
index f5048256c..3ac64b9e3 100644
--- a/helm/rustfs/values.yaml
+++ b/helm/rustfs/values.yaml
@@ -218,7 +218,7 @@ config:
       vault:
         vault_backend: "" # Only support vault kv2 and vault transit.
         vault_address: ""
-        vault_token: ""
+        vault_token: "" # Rendered into a dedicated Secret, never into the config ConfigMap.
         vault_mount_path: ""
         default_key: ""
 
diff --git a/rustfs/src/admin/handlers/user.rs b/rustfs/src/admin/handlers/user.rs
index d6b86e08f..803e9e915 100644
--- a/rustfs/src/admin/handlers/user.rs
+++ b/rustfs/src/admin/handlers/user.rs
@@ -27,7 +27,7 @@ use crate::{
 };
 use http::{HeaderMap, StatusCode};
 use matchit::Params;
-use rustfs_config::{MAX_ADMIN_REQUEST_BODY_SIZE, MAX_IAM_IMPORT_SIZE};
+use rustfs_config::{MAX_ADMIN_REQUEST_BODY_SIZE, MAX_IAM_IMPORT_EXPANDED_SIZE, MAX_IAM_IMPORT_SIZE};
 use rustfs_credentials::Credentials;
 use rustfs_iam::{
     store::{GroupInfo, MappedPolicy, UserType},
@@ -120,6 +120,22 @@ fn should_check_deny_only(target_access_key: &str, requester: &Credentials) -> b
     target_access_key == requester.access_key && requester.parent_user.is_empty()
 }
 
+/// Returns `true` when a derived credential (Console/STS session or service account) targets the
+/// long-term IAM user it was minted from.
+///
+/// SECURITY: [`should_check_deny_only`] relaxes the admin policy check to deny-only for exactly
+/// this target, so without this guard an AddUser call could rewrite the parent's stored secret key
+/// and status, turning a short-lived session into permanent control of the account. The parent is
+/// resolved the same way [`should_check_deny_only`] resolves it, since some stores persist the
+/// parent only inside the session token.
+fn add_user_targets_requester_parent(target_access_key: &str, requester: &Credentials) -> bool {
+    if !requester.is_temp() && !requester.is_service_account() {
+        return false;
+    }
+
+    temp_identity_parent(requester).is_some_and(|parent| parent == target_access_key)
+}
+
 fn should_reject_group_import_name(group_name: &str, group_lookup: &rustfs_iam::error::Error) -> bool {
     has_space_be(group_name) || !matches!(group_lookup, rustfs_iam::error::Error::NoSuchGroup(_))
 }
@@ -251,6 +267,13 @@ impl Operation for AddUser {
             return Err(s3_error!(InvalidArgument, "access key is not valid UTF-8"));
         }
 
+        if add_user_targets_requester_parent(ak, &cred) {
+            return Err(s3_error!(
+                InvalidArgument,
+                "cannot change the credentials of the parent user of this session"
+            ));
+        }
+
         let check_deny_only = should_check_deny_only(ak, &cred);
 
         debug!(
@@ -885,6 +908,31 @@ impl Operation for ExportIam {
     }
 }
 
+/// Read one member of an IAM import archive, drawing from a shared expansion budget.
+///
+/// `MAX_IAM_IMPORT_SIZE` bounds the compressed upload only. Deflate ratios well above
+/// 100:1 are easy to construct, so reading members with `read_to_end` lets a small
+/// archive expand without bound. The budget is shared across every member so the
+/// archive as a whole is capped, not each member independently.
+fn read_import_member(file: &mut impl std::io::Read, budget: &mut u64) -> S3Result<Vec<u8>> {
+    let mut file_content = Vec::new();
+    // Read one byte past the budget: if it arrives, the member exceeded what is left.
+    let read = file
+        .take(budget.saturating_add(1))
+        .read_to_end(&mut file_content)
+        .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))? as u64;
+
+    if read > *budget {
+        return Err(s3_error!(
+            EntityTooLarge,
+            "IAM import archive expands beyond the {MAX_IAM_IMPORT_EXPANDED_SIZE} byte limit"
+        ));
+    }
+
+    *budget -= read;
+    Ok(file_content)
+}
+
 pub struct ImportIam {}
 #[async_trait::async_trait]
 impl Operation for ImportIam {
@@ -926,6 +974,10 @@ impl Operation for ImportIam {
         let mut zip_reader =
             ZipArchive::new(Cursor::new(body)).map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
 
+        // Shared across every member below, so the archive as a whole is bounded rather
+        // than each member being allowed the full limit.
+        let mut expansion_budget = MAX_IAM_IMPORT_EXPANDED_SIZE;
+
         let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
             return Err(s3_error!(InvalidRequest, "iam not init"));
         };
@@ -942,10 +994,7 @@ impl Operation for ImportIam {
                 Err(_) => return Err(s3_error!(InvalidRequest, "get file failed")),
                 Ok(file) => {
                     let mut file = file;
-                    let mut file_content = Vec::new();
-                    file.read_to_end(&mut file_content)
-                        .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
-                    Some(file_content)
+                    Some(read_import_member(&mut file, &mut expansion_budget)?)
                 }
             };
 
@@ -982,10 +1031,7 @@ impl Operation for ImportIam {
                 Err(_) => return Err(s3_error!(InvalidRequest, "get file failed")),
                 Ok(file) => {
                     let mut file = file;
-                    let mut file_content = Vec::new();
-                    file.read_to_end(&mut file_content)
-                        .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
-                    Some(file_content)
+                    Some(read_import_member(&mut file, &mut expansion_budget)?)
                 }
             };
 
@@ -1024,10 +1070,7 @@ impl Operation for ImportIam {
                 Err(_) => return Err(s3_error!(InvalidRequest, "get file failed")),
                 Ok(file) => {
                     let mut file = file;
-                    let mut file_content = Vec::new();
-                    file.read_to_end(&mut file_content)
-                        .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
-                    Some(file_content)
+                    Some(read_import_member(&mut file, &mut expansion_budget)?)
                 }
             };
 
@@ -1068,10 +1111,7 @@ impl Operation for ImportIam {
                 Err(_) => return Err(s3_error!(InvalidRequest, "get file failed")),
                 Ok(file) => {
                     let mut file = file;
-                    let mut file_content = Vec::new();
-                    file.read_to_end(&mut file_content)
-                        .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
-                    Some(file_content)
+                    Some(read_import_member(&mut file, &mut expansion_budget)?)
                 }
             };
 
@@ -1182,10 +1222,7 @@ impl Operation for ImportIam {
                 Err(_) => return Err(s3_error!(InvalidRequest, "get file failed")),
                 Ok(file) => {
                     let mut file = file;
-                    let mut file_content = Vec::new();
-                    file.read_to_end(&mut file_content)
-                        .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
-                    Some(file_content)
+                    Some(read_import_member(&mut file, &mut expansion_budget)?)
                 }
             };
 
@@ -1233,10 +1270,7 @@ impl Operation for ImportIam {
                 Err(_) => return Err(s3_error!(InvalidRequest, "get file failed")),
                 Ok(file) => {
                     let mut file = file;
-                    let mut file_content = Vec::new();
-                    file.read_to_end(&mut file_content)
-                        .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
-                    Some(file_content)
+                    Some(read_import_member(&mut file, &mut expansion_budget)?)
                 }
             };
 
@@ -1274,10 +1308,7 @@ impl Operation for ImportIam {
                 Err(_) => return Err(s3_error!(InvalidRequest, "get file failed")),
                 Ok(file) => {
                     let mut file = file;
-                    let mut file_content = Vec::new();
-                    file.read_to_end(&mut file_content)
-                        .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
-                    Some(file_content)
+                    Some(read_import_member(&mut file, &mut expansion_budget)?)
                 }
             };
 
@@ -1341,10 +1372,11 @@ impl Operation for ImportIam {
 #[cfg(test)]
 mod tests {
     use super::{
-        GROUP_POLICY_MAPPING_USER_TYPE, SERVICE_ACCOUNT_ACCESS_KEY_MISMATCH_ERROR, SERVICE_ACCOUNT_PARENT_SCOPE_ERROR,
+        GROUP_POLICY_MAPPING_USER_TYPE, MAX_IAM_IMPORT_EXPANDED_SIZE, MAX_IAM_IMPORT_SIZE,
+        SERVICE_ACCOUNT_ACCESS_KEY_MISMATCH_ERROR, SERVICE_ACCOUNT_PARENT_SCOPE_ERROR, add_user_targets_requester_parent,
         imported_service_account_access_key_failure, imported_service_account_parent_allowed,
         imported_service_account_parent_scope_failure, imported_service_account_status, map_add_user_create_error,
-        should_check_deny_only, should_reject_group_import_name, should_restore_group_as_disabled,
+        read_import_member, should_check_deny_only, should_reject_group_import_name, should_restore_group_as_disabled,
     };
     use rustfs_credentials::{Credentials, IAM_POLICY_CLAIM_NAME_SA};
     use rustfs_iam::error::Error as IamError;
@@ -1456,6 +1488,78 @@ mod tests {
         assert!(!should_check_deny_only("alice", &cred));
     }
 
+    #[test]
+    fn test_add_user_rejects_sts_session_targeting_its_parent() {
+        let cred = Credentials {
+            access_key: "VV0V3VYJK2PV6EG45X2Y".to_string(),
+            secret_key: "CS_TEST_SECRET_DO_NOT_LOG".to_string(),
+            session_token: "jwt-session-token".to_string(),
+            parent_user: "alice".to_string(),
+            ..Default::default()
+        };
+
+        // The relaxed deny-only admin check is reachable for exactly this target, so a guard on the
+        // write itself is what keeps the session from rewriting alice's long-term secret.
+        assert!(should_check_deny_only("alice", &cred));
+        assert!(add_user_targets_requester_parent("alice", &cred));
+        assert!(!add_user_targets_requester_parent("bob", &cred));
+    }
+
+    #[test]
+    fn test_add_user_rejects_sts_session_with_parent_only_in_jwt_claims() {
+        let mut claims = HashMap::new();
+        claims.insert("parent".to_string(), Value::String("alice".to_string()));
+        let cred = Credentials {
+            access_key: "39KNO04Z34D6T4AGL6E6".to_string(),
+            secret_key: "CS_TEST_SECRET_DO_NOT_LOG".to_string(),
+            session_token: "jwt-session-token".to_string(),
+            claims: Some(claims),
+            ..Default::default()
+        };
+
+        assert!(add_user_targets_requester_parent("alice", &cred));
+    }
+
+    #[test]
+    fn test_add_user_rejects_service_account_targeting_its_parent() {
+        let mut claims = HashMap::new();
+        claims.insert(IAM_POLICY_CLAIM_NAME_SA.to_string(), Value::String("policy".to_string()));
+        let cred = Credentials {
+            access_key: "service-account".to_string(),
+            parent_user: "alice".to_string(),
+            claims: Some(claims),
+            ..Default::default()
+        };
+
+        assert!(add_user_targets_requester_parent("alice", &cred));
+        assert!(!add_user_targets_requester_parent("bob", &cred));
+    }
+
+    #[test]
+    fn test_add_user_allows_long_term_credentials_to_manage_other_users() {
+        let cred = Credentials {
+            access_key: "admin".to_string(),
+            secret_key: "CS_TEST_SECRET_DO_NOT_LOG".to_string(),
+            ..Default::default()
+        };
+
+        assert!(!add_user_targets_requester_parent("alice", &cred));
+        assert!(!add_user_targets_requester_parent("admin", &cred));
+    }
+
+    #[test]
+    fn test_add_user_operation_guards_requester_parent_target() {
+        let guard_call = concat!("add_user_targets_requester_", "parent(ak, &cred)");
+        let source = include_str!("user.rs");
+        let guard_at = source
+            .find(guard_call)
+            .expect("AddUser must reject requests that target the requester's parent user");
+        let sink_at = source
+            .find("iam_store.create_user(ak, &args)")
+            .expect("AddUser create_user sink moved; re-anchor this guard test");
+        assert!(guard_at < sink_at, "the parent-user guard must run before the create_user write");
+    }
+
     #[test]
     fn test_group_import_allows_missing_group_without_spaces() {
         assert!(!should_reject_group_import_name(
@@ -1628,4 +1732,48 @@ mod tests {
     fn test_group_policy_mappings_use_regular_user_type() {
         assert_eq!(GROUP_POLICY_MAPPING_USER_TYPE, rustfs_iam::store::UserType::Reg);
     }
+
+    /// A single member must not be able to exhaust the archive's expansion budget:
+    /// `MAX_IAM_IMPORT_SIZE` bounds the compressed upload only, so an unbounded
+    /// `read_to_end` here would let a small archive expand without limit.
+    #[test]
+    fn import_member_read_is_bounded_by_the_expansion_budget() {
+        let oversized = vec![b'a'; 1024];
+        let mut budget: u64 = 512;
+
+        let err = read_import_member(&mut oversized.as_slice(), &mut budget)
+            .expect_err("a member larger than the remaining budget must be rejected");
+        assert!(
+            matches!(*err.code(), S3ErrorCode::EntityTooLarge),
+            "expected EntityTooLarge, got {:?}",
+            err.code()
+        );
+    }
+
+    /// The budget is shared across members, so successive reads draw it down and the
+    /// archive as a whole is capped rather than each member independently.
+    #[test]
+    fn import_member_budget_is_shared_across_members() {
+        let member = vec![b'a'; 400];
+        let mut budget: u64 = 1000;
+
+        let first = read_import_member(&mut member.as_slice(), &mut budget).expect("first member fits");
+        assert_eq!(first.len(), 400);
+        assert_eq!(budget, 600, "budget must be drawn down by what was read");
+
+        read_import_member(&mut member.as_slice(), &mut budget).expect("second member still fits");
+        assert_eq!(budget, 200);
+
+        read_import_member(&mut member.as_slice(), &mut budget)
+            .expect_err("third member must be rejected once the shared budget is exhausted");
+    }
+
+    /// Guards the bound from being set so tight that ordinary imports break.
+    #[test]
+    fn import_expansion_budget_leaves_headroom_over_the_compressed_cap() {
+        assert!(
+            MAX_IAM_IMPORT_EXPANDED_SIZE >= MAX_IAM_IMPORT_SIZE as u64,
+            "expanded budget must not be smaller than the compressed upload cap"
+        );
+    }
 }
diff --git a/rustfs/src/config/info.rs b/rustfs/src/config/info.rs
index 14a6fd4eb..95289581c 100644
--- a/rustfs/src/config/info.rs
+++ b/rustfs/src/config/info.rs
@@ -787,6 +787,7 @@ fn format_protocol_config_info() -> String {
     const DEFAULT_FTPS_PASSIVE_PORTS: &str = "40000-50000";
     const DEFAULT_WEBDAV_MAX_BODY_SIZE: u64 = 5 * 1024 * 1024 * 1024;
     const DEFAULT_WEBDAV_REQUEST_TIMEOUT_SECS: u64 = 300;
+    const DEFAULT_WEBDAV_MAX_CONNECTIONS: usize = 1024;
 
     let ftps_enable = rustfs_utils::get_env_bool(rustfs_config::ENV_FTPS_ENABLE, false);
     let ftps_address = rustfs_utils::get_env_str(rustfs_config::ENV_FTPS_ADDRESS, rustfs_config::DEFAULT_FTPS_ADDRESS);
@@ -809,6 +810,8 @@ fn format_protocol_config_info() -> String {
     let webdav_max_body_size = rustfs_utils::get_env_u64(rustfs_config::ENV_WEBDAV_MAX_BODY_SIZE, DEFAULT_WEBDAV_MAX_BODY_SIZE);
     let webdav_request_timeout =
         rustfs_utils::get_env_u64(rustfs_config::ENV_WEBDAV_REQUEST_TIMEOUT, DEFAULT_WEBDAV_REQUEST_TIMEOUT_SECS);
+    let webdav_max_connections =
+        rustfs_utils::get_env_usize(rustfs_config::ENV_WEBDAV_MAX_CONNECTIONS, DEFAULT_WEBDAV_MAX_CONNECTIONS);
 
     format!(
         "| FTPS | --- |\n\
@@ -828,7 +831,8 @@ fn format_protocol_config_info() -> String {
          | WebDAV > Certs Dir (`{}`) | {} |\n\
          | WebDAV > CA File (`{}`) | {} |\n\
          | WebDAV > Max Body Size (`{}`) | {} bytes |\n\
-         | WebDAV > Request Timeout (`{}`) | {} seconds |",
+         | WebDAV > Request Timeout (`{}`) | {} seconds |\n\
+         | WebDAV > Max Connections (`{}`) | {} |",
         if cfg!(feature = "ftps") { "enabled" } else { "disabled" },
         rustfs_config::ENV_FTPS_ENABLE,
         ftps_enable,
@@ -858,7 +862,9 @@ fn format_protocol_config_info() -> String {
         rustfs_config::ENV_WEBDAV_MAX_BODY_SIZE,
         webdav_max_body_size,
         rustfs_config::ENV_WEBDAV_REQUEST_TIMEOUT,
-        webdav_request_timeout
+        webdav_request_timeout,
+        rustfs_config::ENV_WEBDAV_MAX_CONNECTIONS,
+        webdav_max_connections
     )
 }
 
diff --git a/rustfs/src/init.rs b/rustfs/src/init.rs
index ab5b5317a..f94523c00 100644
--- a/rustfs/src/init.rs
+++ b/rustfs/src/init.rs
@@ -1110,7 +1110,7 @@ pub async fn init_webdav_system() -> Result<Option<ShutdownHandle>, Box<dyn std:
         use crate::protocols::ProtocolStorageClient;
         use rustfs_config::{
             DEFAULT_WEBDAV_ADDRESS, ENV_WEBDAV_ADDRESS, ENV_WEBDAV_CA_FILE, ENV_WEBDAV_CERTS_DIR, ENV_WEBDAV_ENABLE,
-            ENV_WEBDAV_MAX_BODY_SIZE, ENV_WEBDAV_REQUEST_TIMEOUT, ENV_WEBDAV_TLS_ENABLED,
+            ENV_WEBDAV_MAX_BODY_SIZE, ENV_WEBDAV_MAX_CONNECTIONS, ENV_WEBDAV_REQUEST_TIMEOUT, ENV_WEBDAV_TLS_ENABLED,
         };
         use rustfs_protocols::{WebDavConfig, WebDavServer};
 
@@ -1141,6 +1141,7 @@ pub async fn init_webdav_system() -> Result<Option<ShutdownHandle>, Box<dyn std:
         let max_body_size = rustfs_utils::get_env_u64(ENV_WEBDAV_MAX_BODY_SIZE, WebDavConfig::DEFAULT_MAX_BODY_SIZE);
         let request_timeout_secs =
             rustfs_utils::get_env_u64(ENV_WEBDAV_REQUEST_TIMEOUT, WebDavConfig::DEFAULT_REQUEST_TIMEOUT_SECS);
+        let max_connections = rustfs_utils::get_env_usize(ENV_WEBDAV_MAX_CONNECTIONS, WebDavConfig::DEFAULT_MAX_CONNECTIONS);
 
         // Create WebDAV configuration
         let config = WebDavConfig {
@@ -1150,6 +1151,7 @@ pub async fn init_webdav_system() -> Result<Option<ShutdownHandle>, Box<dyn std:
             ca_file,
             max_body_size,
             request_timeout_secs,
+            max_connections,
         };
 
         // Create WebDAV server with protocol storage client
@@ -1160,6 +1162,7 @@ pub async fn init_webdav_system() -> Result<Option<ShutdownHandle>, Box<dyn std:
         let tls_enabled = server.config().tls_enabled;
         let max_body_size = server.config().max_body_size;
         let request_timeout_secs = server.config().request_timeout_secs;
+        let max_connections = server.config().max_connections;
 
         // Log server configuration
         debug!(
@@ -1173,6 +1176,7 @@ pub async fn init_webdav_system() -> Result<Option<ShutdownHandle>, Box<dyn std:
             tls_enabled,
             max_body_size,
             request_timeout_secs,
+            max_connections,
             "Protocol runtime configured"
         );
 
diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs
index 895f43378..b88a4e53a 100644
--- a/rustfs/src/server/http.rs
+++ b/rustfs/src/server/http.rs
@@ -29,7 +29,8 @@ use crate::server::{
     },
     rate_limit::{RateLimitLayer, api_rate_limit_layer_from_env},
     tls_material::{
-        TlsAcceptorHolder, TlsHandshakeFailureKind, build_acceptor_from_loaded, load_tls_material, spawn_reload_loop,
+        TlsAcceptFailure, TlsAcceptorHolder, TlsHandshakeFailureKind, accept_tls_with_deadline, build_acceptor_from_loaded,
+        load_tls_material, spawn_reload_loop,
     },
 };
 use crate::storage_api::server::http as storage;
@@ -233,6 +234,18 @@ fn log_tls_handshake_failure(peer_addr: &str, kind: TlsHandshakeFailureKind, err
                 "TLS handshake failed"
             );
         }
+        TlsHandshakeFailureKind::Timeout => {
+            warn!(
+                event = EVENT_TLS_HANDSHAKE_FAILED,
+                component = LOG_COMPONENT_SERVER,
+                subsystem = LOG_SUBSYSTEM_TLS,
+                peer_addr = %peer_addr,
+                failure_type = kind.as_str(),
+                error = %err,
+                result = "client_timeout",
+                "TLS handshake failed"
+            );
+        }
         TlsHandshakeFailureKind::Unknown => {
             error!(
                 event = EVENT_TLS_HANDSHAKE_FAILED,
@@ -1049,6 +1062,7 @@ pub async fn start_http_server(
                 trusted_proxy_layer: rustfs_trusted_proxies::is_enabled().then(|| rustfs_trusted_proxies::layer().clone()),
                 rate_limit_layer: api_rate_limit_layer.clone(),
                 server_ctx: Arc::clone(&server_ctx),
+                tls_handshake_timeout: Duration::from_secs(http1_header_read_timeout),
             };
 
             process_connection(socket, tls_acceptor.clone(), connection_ctx, graceful.watcher(), connection_permit);
@@ -1109,6 +1123,10 @@ struct ConnectionContext {
     /// All clones share one limiter, keeping budgets global across connections.
     rate_limit_layer: Option<RateLimitLayer>,
     server_ctx: Arc<ServerContextSlot>,
+    /// Deadline for the TLS handshake of this connection. Reuses the HTTP/1 header-read budget,
+    /// the existing slow-client bound for the pre-request phase, and is pre-computed with the
+    /// other transport parameters to avoid a per-connection env read.
+    tls_handshake_timeout: Duration,
 }
 
 #[derive(Clone)]
@@ -1299,6 +1317,7 @@ fn process_connection(
             trusted_proxy_layer,
             rate_limit_layer,
             server_ctx,
+            tls_handshake_timeout,
         } = context;
 
         // Build the hybrid service per-connection.
@@ -1752,7 +1771,7 @@ fn process_connection(
                 .ok()
                 .map_or_else(|| "unknown".to_string(), |addr| addr.to_string());
             let acceptor = holder.get();
-            match acceptor.accept(socket).await {
+            match accept_tls_with_deadline(&acceptor, socket, tls_handshake_timeout).await {
                 Ok(tls_socket) => {
                     trace!("TLS handshake successful");
                     let stream = TokioIo::new(tls_socket);
@@ -1761,7 +1780,15 @@ fn process_connection(
                         handle_connection_error(Some(peer_addr.as_str()), &*err);
                     }
                 }
-                Err(err) => {
+                Err(TlsAcceptFailure::Timeout) => {
+                    let kind = TlsHandshakeFailureKind::Timeout;
+                    let err = format!("TLS handshake did not complete within {}s", tls_handshake_timeout.as_secs());
+                    log_tls_handshake_failure(&peer_addr, kind, &err);
+                    counter!("rustfs_tls_handshake_failures", &[("failure_type", kind.as_str())]).increment(1);
+
+                    return;
+                }
+                Err(TlsAcceptFailure::Handshake(err)) => {
                     let err_str = err.to_string();
                     let kind = TlsHandshakeFailureKind::classify(&err_str);
                     log_tls_handshake_failure(&peer_addr, kind, &err);
diff --git a/rustfs/src/server/tls_material.rs b/rustfs/src/server/tls_material.rs
index 7a4cb4584..a211f7392 100644
--- a/rustfs/src/server/tls_material.rs
+++ b/rustfs/src/server/tls_material.rs
@@ -438,6 +438,7 @@ pub(crate) enum TlsHandshakeFailureKind {
     ProtocolVersion,
     Certificate,
     Alert,
+    Timeout,
     Unknown,
 }
 
@@ -462,11 +463,38 @@ impl TlsHandshakeFailureKind {
             Self::ProtocolVersion => "PROTOCOL_VERSION",
             Self::Certificate => "CERTIFICATE",
             Self::Alert => "ALERT",
+            Self::Timeout => "TIMEOUT",
             Self::Unknown => "UNKNOWN",
         }
     }
 }
 
+/// Why a TLS handshake did not produce a usable stream.
+pub(crate) enum TlsAcceptFailure {
+    Handshake(std::io::Error),
+    Timeout,
+}
+
+/// Run the server-side TLS handshake under `handshake_timeout`.
+///
+/// Security invariant: the peer is unauthenticated at this point and the connection cap is
+/// unlimited by default, so this deadline is the only control that sheds a client which opens a
+/// socket and then stalls the handshake forever.
+pub(crate) async fn accept_tls_with_deadline<IO>(
+    acceptor: &TlsAcceptor,
+    socket: IO,
+    handshake_timeout: Duration,
+) -> Result<tokio_rustls::server::TlsStream<IO>, TlsAcceptFailure>
+where
+    IO: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+    match tokio::time::timeout(handshake_timeout, acceptor.accept(socket)).await {
+        Ok(Ok(tls_socket)) => Ok(tls_socket),
+        Ok(Err(err)) => Err(TlsAcceptFailure::Handshake(err)),
+        Err(_) => Err(TlsAcceptFailure::Timeout),
+    }
+}
+
 // ── TLS Acceptor Holder (for hot reload) ──
 
 pub(crate) struct TlsAcceptorHolder {
@@ -669,6 +697,35 @@ mod tests {
         assert!(acceptor.is_some());
     }
 
+    #[tokio::test]
+    async fn tls_handshake_deadline_sheds_a_peer_that_never_sends_client_hello() {
+        ensure_rustls_crypto_provider();
+        let temp_dir = TempDir::new().expect("temp dir should create");
+        write_test_cert_pair(temp_dir.path(), "localhost");
+
+        let snapshot = load_tls_material(temp_dir.path().to_str().expect("temp dir path should be utf-8"))
+            .await
+            .expect("TLS material load should succeed");
+        let holder = build_acceptor_from_loaded(snapshot.server, temp_dir.path())
+            .await
+            .expect("TLS acceptor should build")
+            .expect("TLS acceptor should be present");
+
+        // The client half stays open and silent, so no ClientHello ever arrives.
+        let (_client, server) = tokio::io::duplex(1024);
+        let outcome = tokio::time::timeout(
+            Duration::from_secs(5),
+            accept_tls_with_deadline(&holder.get(), server, Duration::from_millis(100)),
+        )
+        .await
+        .expect("a stalled TLS handshake must be shed by the handshake deadline");
+
+        assert!(
+            matches!(outcome, Err(TlsAcceptFailure::Timeout)),
+            "a silent peer must fail the handshake with a timeout"
+        );
+    }
+
     #[tokio::test]
     async fn build_acceptor_accepts_symlinked_root_single_cert_directory() {
         ensure_rustls_crypto_provider();
diff --git a/scripts/test_helm_templates.sh b/scripts/test_helm_templates.sh
index f9ac3e334..4bdccceb4 100755
--- a/scripts/test_helm_templates.sh
+++ b/scripts/test_helm_templates.sh
@@ -195,6 +195,65 @@ if [[ $partial_empty_status -eq 0 ]]; then
   exit 1
 fi
 
+# The Vault KMS token is a credential: it must be rendered into a Secret and must never
+# reach the config ConfigMap, which is readable by anyone allowed to get ConfigMaps.
+kms_token="CS_TEST_SECRET_DO_NOT_LOG"
+kms_values=(
+  --set config.rustfs.kms.enabled=true
+  --set config.rustfs.kms.type=vault
+  --set config.rustfs.kms.vault.vault_backend=vault-kv2
+  --set config.rustfs.kms.vault.vault_address=http://vault.rustfs.svc:8200
+  --set "config.rustfs.kms.vault.vault_token=$kms_token"
+  --set config.rustfs.kms.vault.default_key=test-key
+)
+
+kms_configmap=$(render_distributed_configmap "${kms_values[@]}")
+if grep -q 'RUSTFS_KMS_VAULT_TOKEN' <<<"$kms_configmap"; then
+  echo "The Vault KMS token must not be rendered into the ConfigMap" >&2
+  exit 1
+fi
+if ! grep -q 'RUSTFS_KMS_VAULT_ADDRESS' <<<"$kms_configmap"; then
+  echo "Non-secret KMS settings must stay in the ConfigMap" >&2
+  exit 1
+fi
+
+kms_full=$(helm template rustfs "$CHART_DIR" \
+  --namespace rustfs \
+  --set secret.rustfs.access_key=test-access-key \
+  --set secret.rustfs.secret_key=test-secret-key \
+  "${kms_values[@]}")
+if grep -q "$kms_token" <<<"$kms_full"; then
+  echo "The Vault KMS token must never appear in plaintext in any rendered manifest" >&2
+  exit 1
+fi
+expected_kms_token_b64=$(printf '%s' "$kms_token" | base64)
+if ! grep -q "RUSTFS_KMS_VAULT_TOKEN: \"$expected_kms_token_b64\"" <<<"$kms_full"; then
+  echo "The Vault KMS token must be rendered into a Secret" >&2
+  exit 1
+fi
+
+kms_statefulset=$(render_distributed_statefulset "${kms_values[@]}")
+if ! grep -q 'name: rustfs-kms-secret' <<<"$kms_statefulset"; then
+  echo "The distributed StatefulSet must consume the KMS Secret via envFrom" >&2
+  exit 1
+fi
+
+kms_deployment=$(render_standalone_deployment "${kms_values[@]}")
+if ! grep -q 'name: rustfs-kms-secret' <<<"$kms_deployment"; then
+  echo "The standalone Deployment must consume the KMS Secret via envFrom" >&2
+  exit 1
+fi
+
+# Without a configured token no KMS Secret is rendered and nothing references it.
+no_kms_output=$(helm template rustfs "$CHART_DIR" \
+  --namespace rustfs \
+  --set secret.rustfs.access_key=test-access-key \
+  --set secret.rustfs.secret_key=test-secret-key)
+if grep -q 'kms-secret' <<<"$no_kms_output"; then
+  echo "No KMS Secret must be rendered when no Vault token is configured" >&2
+  exit 1
+fi
+
 command -v yq >/dev/null 2>&1 || { echo "yq is required for extra-volumes structural tests" >&2; exit 1; }
 
 # Structural helpers: verify wiring at the right YAML paths, not just string presence.