Merge commit from fork

* fix(admin): bound IAM import archive expansion

MAX_IAM_IMPORT_SIZE caps the compressed upload at 10 MB, but every member of the
archive was then read with read_to_end into an unbounded Vec. Deflate ratios well
above 100:1 are easy to construct, so a small authorized upload could expand
without limit across the seven members ImportIam reads.

Add a shared expansion budget (MAX_IAM_IMPORT_EXPANDED_SIZE, 10x the compressed
cap) drawn down by every member, and route all seven reads through one helper
that reads a byte past the remaining budget to detect overrun. Sharing the budget
bounds the archive as a whole rather than letting each member spend the full
limit independently.

Covers R03-CAN-024 through R03-CAN-030 plus R04-CAN-077 (backlog #1471) — one
fix rather than seven, since all seven call sites were byte-identical.

* fix(kms): confine local key paths and refuse silent key replacement

Local KMS key identifiers arrive from request input — the `name` tag on CreateKey,
the `keyId` body field or query parameter on DeleteKey — and were joined onto
`key_dir` with no validation. An identifier such as `../../tmp/evil` escaped the
configured directory, making key creation a constrained arbitrary-file write and
`DeleteKey` with `force_immediate` a cross-directory delete.

Validate in `master_key_path` and make it fallible, so every filesystem path in
this backend inherits the guard: decode_stored_key, load_master_key,
save_master_key, create_key and delete_key all derive their paths there. The rule
is containment rather than a character allowlist, so identifiers already in use
keep resolving; only separators, NUL, absolute paths and non-single-component
forms are refused. Note `.` and `..` are contained rather than refused — the
`.key` suffix turns them into the ordinary filenames `..key` and `...key`.

Separately, `LocalKmsBackend::create_key` had no existence check, while the
sibling `KmsClient::create_key` has always had one. Since `save_master_key`
renames over its destination, creating a key under an existing name silently
replaced its material and destroyed the ability to decrypt everything wrapped
under it — and the backend path is the one the admin API uses. It now returns
KeyAlreadyExists, matching StaticKmsBackend.

Covers R03-CAN-072, R03-CAN-073 and R07-CAN-103 (backlog #1475). R03-CAN-073
needed no separate change: delete_key routes both its load and its remove_file
through master_key_path.

* fix(swift): bound SLO manifest reads to the 2 MiB manifest limit

The three Swift SLO handlers that load a stored manifest (handle_slo_get,
handle_slo_get_manifest, handle_slo_delete) read the `<object>.slo-manifest`
object to EOF with AsyncReadExt::read_to_end. That key is predictable and
writable through the ordinary object PUT path, so a tenant can replace the
manifest with an arbitrarily large object and then make the server allocate
its full size on every SLO GET, multipart-manifest=get, or
multipart-manifest=delete request - a memory amplification bounded only by
the stored object size (CWE-400 / CWE-770). The 2 MiB manifest limit that
handle_slo_put enforces was not applied on the read side.

Introduce MAX_SLO_MANIFEST_SIZE (the existing 2 MiB PUT limit, now a named
constant) and a shared read_manifest_bytes helper that reads through a
`take(limit + 1)` and rejects anything larger, so an oversized manifest is
refused instead of being buffered first. All three call sites go through the
helper. handle_slo_put now checks the size before parsing the JSON.

Regression tests: test_read_manifest_bytes_rejects_oversized_manifest and
test_read_manifest_bytes_stops_reading_oversized_manifest (which asserts the
reader is not consumed past the limit), plus a boundary test that a manifest
at exactly 2 MiB is still accepted.

* fix(protocols): authorize every object in FTPS/WebDAV recursive deletes

The FTPS and WebDAV gateways authorized only the container before a
recursive delete and then destroyed everything inside it without a
further check:

- FTPS RMD (and DELE on a bucket path ending in '/') cleared
  s3:DeleteBucket, then delete_bucket_recursively listed the bucket and
  deleted every object.
- WebDAV DELETE on a bucket did the same via its own
  delete_bucket_recursively.
- WebDAV DELETE on a directory cleared s3:DeleteObject for the directory
  marker key ("dir/") only, then listed that prefix and deleted every
  child under it.

A principal holding s3:DeleteBucket (or s3:DeleteObject on a single
marker key) could therefore erase objects it had no s3:DeleteObject
permission for, and the operation reported success.

Deletion stays recursive - that is the expected behaviour for these
protocols - but each object now clears s3:DeleteObject on its own key
before it is removed, and the enumeration clears s3:ListBucket. A denial
aborts the whole operation with access denied rather than being skipped,
so the caller can never be told the delete succeeded while objects were
left behind or removed without authorization.

The test double gained shared-state cloning, delete_object/delete_bucket
call logs, and list/delete queue helpers so the regression tests can
observe that nothing is deleted once a deny lands.

* fix(server,ecstore): bound TLS handshakes and remote volume RPC waits

Three call sites let an unauthenticated client or a misbehaving peer hold
server resources with no deadline.

TLS listener (R03-CAN-035): process_connection awaited
`acceptor.accept(socket)` with no bound. A client that opens a TCP
connection and never finishes the handshake parks a Tokio task and a socket
forever, and the connection cap (RUSTFS_API_MAX_CONNECTIONS) is unlimited by
default, so nothing else sheds it. The handshake now runs under
accept_tls_with_deadline(), reusing the existing HTTP/1 header-read budget —
the established slow-client bound for the pre-request phase — and the
expiry is recorded through the same log/metric path as a handshake error,
under a new TIMEOUT failure kind.

Remote disk RPCs (R03-CAN-049, R03-CAN-050): list_volumes and delete_volume
passed Duration::ZERO, which execute_with_timeout treats as "no deadline",
so a peer that accepts the request and never answers stalls the coordinator
(and, for delete_volume, the bucket-deletion workflow). Both now pass
get_max_timeout_duration(), matching every sibling method in the file.

Regression tests: a silent TLS peer must be shed by the handshake deadline;
list_volumes/delete_volume against a peer that completes the TCP connect and
then goes silent must fail with DiskError::Timeout instead of hanging.

* fix(security): stop leaking signed headers and bound OIDC/KMS credentials

Three independent hygiene fixes found by the security review.

R03-CAN-018 (crates/signer): try_get_canonical_headers and get_signed_headers
logged the complete header map at DEBUG before signing. Runtime callers pass
session credentials and SSE-C key material through these headers, so anyone
able to raise the log level (or read DEBUG logs) recovered
X-Amz-Security-Token and SSE-C keys verbatim. The statements were debugging
leftovers with no operational value and are deleted rather than redacted.

R03-CAN-014 (crates/iam): the OIDC HTTP adapter buffered provider responses
with an unbounded Response::bytes(), so a configured, compromised or
attacker-pointed IdP endpoint could stream an arbitrarily large or endless
body into memory (the ValidateOidcConfig admin handler lets a ServerInfo
caller choose the endpoint). Responses are now read incrementally and fail
closed past MAX_OIDC_RESPONSE_SIZE, and the already SSRF-hardened client
builder gains request and connect timeouts so a stalled provider cannot pin
the calling task indefinitely.

R07-CAN-105 (helm): the Vault KMS token was serialized into the chart
ConfigMap, exposing it to every subject allowed to get ConfigMaps in the
namespace. It now renders into a dedicated Secret that the Deployment and
StatefulSet consume via envFrom; the Secret is separate from the main
credentials Secret so it also works when secret.existingSecret is set.

Regression tests:
- rustfs-signer: signing_never_logs_signed_header_material
- rustfs-iam: oidc_response_body_past_the_limit_is_rejected,
  oidc_response_body_at_the_limit_is_accepted
- scripts/test_helm_templates.sh: KMS token must never render in plaintext

* fix(webdav): enforce body limit, request timeout and connection cap

The configured WebDAV maximum body size was enforced from Content-Length, so a
chunked request declared no length and bypassed it entirely. The configured
request timeout was never applied to the connection at all, and the accept loop
spawned a task per connection with no bound, so an unauthenticated client could
hold resources indefinitely and in unbounded number.

Enforce the limit on bytes actually read rather than the declared length, apply
the configured timeout to the request, and bound accepted connections with a new
RUSTFS_WEBDAV_MAX_CONNECTIONS (default 1024) surfaced in the config report.

Covers R03-CAN-051, R03-CAN-052, R03-CAN-067, R04-CAN-089, R05-CAN-094 and
R05-CAN-097 (backlog #1471, #1474).

* fix(security): stop STS credentials from crossing the parent trust boundary

Two related credential-boundary holes let a short-lived STS credential act
with the full, unrestricted authority of the long-term user it was minted
from.

AddUser (R03-CAN-021, CWE-269/863): should_check_deny_only relaxes the admin
policy check to deny-only when a Console/STS session targets the IAM user it
represents. Nothing then stopped that session from calling AddUser with its
own parent's access key, so the handler wrote an attacker-chosen secret key
and status over the parent's stored Credentials via create_user ->
save_user_identity. A session that expires in minutes became permanent
control of the account. AddUser now rejects any temp or service-account
requester whose resolved parent equals the target access key, resolving the
parent the same way should_check_deny_only does (parent_user field, else the
JWT `parent` claim, since some stores persist the parent only in the token).

FTPS/SFTP/WebDAV password auth (R04-CAN-086, CWE-287/862): these protocols
looked the access key up with check_key, which falls back to the STS account
cache, and then compared only the stored secret. An STS access key plus
secret therefore authenticated with no session token presented and no
session-policy claims applied - the holder got the parent's full permissions.
Password authentication now rejects temporary credentials before the secret
comparison. The discriminator is is_temp() && !is_service_account(), the same
one IamCache::update_user_with_claims uses to route an identity into the STS
cache, so service accounts - which resolve policy from stored IAM state
rather than a client-presented token - keep working over these protocols.

Regression tests cover both predicates and pin the guards to their call
sites so neither can be dropped without a test failure.

Cargo.lock

@@ -9972,6 +9972,7 @@ dependencies = [
  "thiserror 2.0.19",
  "time",
  "tracing",
+ "tracing-subscriber",
 ]
 
 [[package]]

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

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";

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;
 

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();
+    }
 }

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();

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");
+    }
 }

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())),
         }

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

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(

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

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;

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
+        );
+    }
 }

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
 

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,
         }
     }
 }

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(

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}");
+    }
+}

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
 

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}");
+    }
 }

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. |

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
 */}}

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 }}

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 }}

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

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 }}

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: ""
 

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"
+        );
+    }
 }

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
     )
 }
 

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"
         );
 

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);

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();

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.