refactor(app): resolve the bucket default SSE in one place

PUT, COPY, and extract each walked the bucket encryption configuration themselves to fill in an unrequested SSE algorithm and KMS key. That duplication is what let the unknown-algorithm fallback diverge in the first place (#6022 fixed the divergence; this removes the room for it).

`resolve_bucket_default_sse` now returns the `(sse, kms_key_id)` pair for all three. A request-level value still wins, the unknown-algorithm fallback stays in `bucket_default_write_sse`, and PUT keeps its `ciphertext_passthrough` override at the call site because that clears both values after resolution rather than participating in it.

`has_explicit_ssec` is passed per caller — `true` from COPY, `false` from PUT and extract — because that is what each does today. The resulting divergence is real: PUT hands the resolved default to `validate_sse_headers_for_write`, which rejects SSE-C and managed headers together, so an SSE-C PUT into a bucket with default encryption fails where S3 would let the request win. That is a behaviour change on a P0 encryption path, so it is recorded on backlog#1826 for its own PR rather than folded into this refactor.

Refs backlog#1826

crates/e2e_test/src/multipart_auth_test.rs

@@ -17,7 +17,6 @@
 use crate::common::{RustFSTestEnvironment, init_logging, local_http_client};
 use async_compression::tokio::write::{BzEncoder, XzEncoder};
 use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
-use aws_sdk_s3::operation::head_object::HeadObjectOutput;
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::{
     ServerSideEncryption, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule,
@@ -349,71 +348,6 @@ async fn run_post_object_policy_case(
     Ok(())
 }
 
-/// One accepted POST Object upload driven end-to-end (backlog#1838): starts a
-/// fresh server, allows anonymous PutObject on `bucket`, posts an anonymous
-/// POST Object form whose policy carries `policy_conditions` and whose form
-/// carries `form_field` on top of the mandatory key+policy fields, then asserts
-/// 204 with an empty body, that `read_stored` observes the submitted value on
-/// the stored object, and that the object body round-tripped unchanged.
-/// `case` prefixes every assertion message so a failing table row is
-/// identifiable at a glance.
-#[allow(clippy::too_many_arguments)]
-async fn run_post_object_accept_case(
-    bucket: &str,
-    object_key: &str,
-    policy_conditions: Vec<serde_json::Value>,
-    form_field: (&str, &str),
-    file_mime: &str,
-    file_body: &[u8],
-    read_stored: fn(&HeadObjectOutput) -> Option<&str>,
-    case: &str,
-) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(policy_conditions);
-
-    let (field_name, field_value) = form_field;
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text(field_name.to_string(), field_value.to_string())
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(file_body.to_vec())
-                .file_name("upload.txt")
-                .mime_str(file_mime)?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT, "[{case}] unexpected status");
-    assert!(
-        response_body.is_empty(),
-        "[{case}] 204 response should not contain a body, got: {response_body}"
-    );
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(read_stored(&head), Some(field_value), "[{case}] stored {field_name} mismatch");
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), file_body, "[{case}] uploaded body mismatch");
-
-    Ok(())
-}
-
 /// Table-driven fold of the nine `*_missing_from_policy_conditions` POST
 /// Object tests (backlog#1838 PR1). Every row keeps its original test's exact
 /// bucket, key, form field, file body, and expected error strings; the shared
@@ -1600,6 +1534,59 @@ async fn test_anonymous_post_object_accepts_sse_s3_missing_from_policy_condition
     Ok(())
 }
 
+#[tokio::test]
+async fn test_anonymous_post_object_accepts_storage_class_exact_policy_match()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let bucket = "anon-post-storage-class";
+    let object_key = "post-storage-class-object.txt";
+    let expected_body = b"post-storage-class-body".to_vec();
+    let storage_class = "REDUCED_REDUNDANCY";
+
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(vec![
+        serde_json::json!({ "bucket": bucket }),
+        serde_json::json!({ "key": object_key }),
+        serde_json::json!({ "x-amz-storage-class": storage_class }),
+        serde_json::json!(["content-length-range", 0, 1024]),
+    ]);
+
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text("x-amz-storage-class", storage_class)
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(expected_body.clone())
+                .file_name("upload.txt")
+                .mime_str("text/plain")?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
+        .await?;
+
+    assert_eq!(post_resp.status(), reqwest::StatusCode::NO_CONTENT);
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    assert_eq!(head.storage_class().map(|value| value.as_str()), Some(storage_class));
+
+    let uploaded = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
+    let uploaded = uploaded.body.collect().await?.into_bytes();
+    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_anonymous_post_object_rejects_storage_class_missing_from_policy_conditions()
 -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -2597,182 +2584,512 @@ async fn test_anonymous_post_object_rejects_success_action_redirect_missing_from
     Ok(())
 }
 
-/// Table-driven fold of the eleven accepted POST Object form-field tests
-/// (backlog#1838 PR4). Every row keeps its original test's exact bucket, key,
-/// form field, submitted value, policy condition, file MIME type, and file
-/// body; the shared shape is: the policy covers the field (exact condition or
-/// `starts-with` prefix), the form submits it, the upload returns 204 with an
-/// empty body, and the stored object echoes the submitted value back.
 #[tokio::test]
-async fn test_anonymous_post_object_accepts_fields_covered_by_policy_conditions()
+async fn test_anonymous_post_object_accepts_metadata_field_covered_by_starts_with()
 -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
 
-    // (case, bucket, object_key, field, submitted value, `starts-with` prefix
-    // (`None` pins the field to an exact policy condition), file part MIME type,
-    // file body, stored-value accessor)
-    type Case = (
-        &'static str,
-        &'static str,
-        &'static str,
-        &'static str,
-        &'static str,
-        Option<&'static str>,
-        &'static str,
-        &'static [u8],
-        fn(&HeadObjectOutput) -> Option<&str>,
-    );
-    let cases: &[Case] = &[
-        (
-            "storage-class",
-            "anon-post-storage-class",
-            "post-storage-class-object.txt",
-            "x-amz-storage-class",
-            "REDUCED_REDUNDANCY",
-            None,
-            "text/plain",
-            b"post-storage-class-body",
-            |head: &HeadObjectOutput| head.storage_class().map(|value| value.as_str()),
-        ),
-        (
-            "metadata-starts-with",
-            "anon-post-policy-meta-accept",
-            "uploads/meta-object.txt",
-            "x-amz-meta-project",
-            "alpha-demo",
-            Some("alpha-"),
-            "text/plain",
-            b"post-policy-meta-body",
-            |head: &HeadObjectOutput| head.metadata().and_then(|meta| meta.get("project")).map(String::as_str),
-        ),
-        (
-            "content-type",
-            "anon-post-policy-content-type-accept",
-            "uploads/content-type-accept.txt",
-            "Content-Type",
-            "text/plain",
-            None,
-            "text/plain",
-            b"post-policy-content-type-accept",
-            |head: &HeadObjectOutput| head.content_type(),
-        ),
-        (
-            "content-type-starts-with",
-            "anon-post-policy-content-type-accept",
-            "uploads/content-type-object.txt",
-            "Content-Type",
-            "image/png",
-            Some("image/"),
-            "image/png",
-            b"post-policy-content-type-body",
-            |head: &HeadObjectOutput| head.content_type(),
-        ),
-        (
-            "content-disposition",
-            "anon-post-policy-disposition-accept",
-            "uploads/disposition-object.txt",
-            "Content-Disposition",
-            "attachment; filename=\"upload.txt\"",
-            None,
-            "text/plain",
-            b"post-policy-disposition-body",
-            |head: &HeadObjectOutput| head.content_disposition(),
-        ),
-        (
-            "cache-control",
-            "anon-post-policy-cache-control-accept",
-            "uploads/cache-control-object.txt",
-            "Cache-Control",
-            "max-age=60",
-            None,
-            "text/plain",
-            b"post-policy-cache-control-body",
-            |head: &HeadObjectOutput| head.cache_control(),
-        ),
-        (
-            "content-language",
-            "anon-post-policy-content-language-accept",
-            "uploads/content-language-object.txt",
-            "Content-Language",
-            "en-US",
-            None,
-            "text/plain",
-            b"post-policy-content-language-body",
-            |head: &HeadObjectOutput| head.content_language(),
-        ),
-        (
-            "content-encoding",
-            "anon-post-policy-content-encoding-accept",
-            "uploads/content-encoding-object.txt",
-            "Content-Encoding",
-            "gzip",
-            None,
-            "text/plain",
-            b"post-policy-content-encoding-body",
-            |head: &HeadObjectOutput| head.content_encoding(),
-        ),
-        (
-            "website-redirect-location",
-            "anon-post-policy-website-redirect-accept",
-            "uploads/website-redirect-object.txt",
-            "x-amz-website-redirect-location",
-            "/docs/landing.html",
-            None,
-            "text/plain",
-            b"post-policy-website-redirect-body",
-            |head: &HeadObjectOutput| head.website_redirect_location(),
-        ),
-        (
-            "expires",
-            "anon-post-policy-expires-accept",
-            "uploads/expires-object.txt",
-            "Expires",
-            "Wed, 21 Oct 2037 07:28:00 GMT",
-            None,
-            "text/plain",
-            b"post-policy-expires-body",
-            |head: &HeadObjectOutput| head.expires_string(),
-        ),
-        (
-            "metadata-exact",
-            "anon-post-policy-meta-exact-accept",
-            "uploads/meta-exact-accept-object.txt",
-            "x-amz-meta-project",
-            "alpha-demo",
-            None,
-            "text/plain",
-            b"post-policy-meta-exact-body",
-            |head: &HeadObjectOutput| head.metadata().and_then(|meta| meta.get("project")).map(String::as_str),
-        ),
-    ];
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
 
-    for (case, bucket, object_key, field, value, starts_with_prefix, file_mime, file_body, read_stored) in cases {
-        let condition = match starts_with_prefix {
-            Some(prefix) => serde_json::json!(["starts-with", format!("${field}"), prefix]),
-            None => {
-                let mut exact = serde_json::Map::new();
-                exact.insert((*field).to_string(), serde_json::Value::String((*value).to_string()));
-                serde_json::Value::Object(exact)
-            }
-        };
+    let bucket = "anon-post-policy-meta-accept";
+    let object_key = "uploads/meta-object.txt";
+    let metadata_value = "alpha-demo";
+    let expected_body = b"post-policy-meta-body".to_vec();
 
-        run_post_object_accept_case(
-            bucket,
-            object_key,
-            vec![
-                serde_json::json!({ "bucket": bucket }),
-                serde_json::json!({ "key": object_key }),
-                condition,
-                serde_json::json!(["content-length-range", 0, 1024]),
-            ],
-            (field, value),
-            file_mime,
-            file_body,
-            *read_stored,
-            case,
-        )
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(vec![
+        serde_json::json!({ "bucket": bucket }),
+        serde_json::json!({ "key": object_key }),
+        serde_json::json!(["starts-with", "$x-amz-meta-project", "alpha-"]),
+        serde_json::json!(["content-length-range", 0, 1024]),
+    ]);
+
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text("x-amz-meta-project", metadata_value)
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(expected_body.clone())
+                .file_name("upload.txt")
+                .mime_str("text/plain")?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
         .await?;
-    }
+
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
+    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    let metadata = head.metadata().expect("head_object should expose uploaded metadata");
+    assert_eq!(metadata.get("project").map(String::as_str), Some(metadata_value));
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_anonymous_post_object_accepts_content_type_field_exact_policy_match()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let bucket = "anon-post-policy-content-type-accept";
+    let object_key = "uploads/content-type-accept.txt";
+    let content_type = "text/plain";
+    let expected_body = b"post-policy-content-type-accept".to_vec();
+
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(vec![
+        serde_json::json!({ "bucket": bucket }),
+        serde_json::json!({ "key": object_key }),
+        serde_json::json!({ "Content-Type": content_type }),
+        serde_json::json!(["content-length-range", 0, 1024]),
+    ]);
+
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text("Content-Type", content_type)
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(expected_body.clone())
+                .file_name("upload.txt")
+                .mime_str(content_type)?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
+        .await?;
+
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
+    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    assert_eq!(head.content_type(), Some(content_type));
+
+    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
+    let uploaded = get_out.body.collect().await?.into_bytes();
+    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_anonymous_post_object_accepts_content_type_field_covered_by_starts_with()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let bucket = "anon-post-policy-content-type-accept";
+    let object_key = "uploads/content-type-object.txt";
+    let content_type = "image/png";
+    let expected_body = b"post-policy-content-type-body".to_vec();
+
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(vec![
+        serde_json::json!({ "bucket": bucket }),
+        serde_json::json!({ "key": object_key }),
+        serde_json::json!(["starts-with", "$Content-Type", "image/"]),
+        serde_json::json!(["content-length-range", 0, 1024]),
+    ]);
+
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text("Content-Type", content_type)
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(expected_body.clone())
+                .file_name("upload.txt")
+                .mime_str(content_type)?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
+        .await?;
+
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
+    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    assert_eq!(head.content_type(), Some(content_type));
+
+    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
+    let uploaded = get_out.body.collect().await?.into_bytes();
+    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_anonymous_post_object_accepts_content_disposition_field_exact_policy_match()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let bucket = "anon-post-policy-disposition-accept";
+    let object_key = "uploads/disposition-object.txt";
+    let content_disposition = "attachment; filename=\"upload.txt\"";
+    let expected_body = b"post-policy-disposition-body".to_vec();
+
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(vec![
+        serde_json::json!({ "bucket": bucket }),
+        serde_json::json!({ "key": object_key }),
+        serde_json::json!({ "Content-Disposition": content_disposition }),
+        serde_json::json!(["content-length-range", 0, 1024]),
+    ]);
+
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text("Content-Disposition", content_disposition)
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(expected_body.clone())
+                .file_name("upload.txt")
+                .mime_str("text/plain")?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
+        .await?;
+
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
+    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    assert_eq!(head.content_disposition(), Some(content_disposition));
+
+    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
+    let uploaded = get_out.body.collect().await?.into_bytes();
+    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_anonymous_post_object_accepts_cache_control_field_exact_policy_match()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let bucket = "anon-post-policy-cache-control-accept";
+    let object_key = "uploads/cache-control-object.txt";
+    let cache_control = "max-age=60";
+    let expected_body = b"post-policy-cache-control-body".to_vec();
+
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(vec![
+        serde_json::json!({ "bucket": bucket }),
+        serde_json::json!({ "key": object_key }),
+        serde_json::json!({ "Cache-Control": cache_control }),
+        serde_json::json!(["content-length-range", 0, 1024]),
+    ]);
+
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text("Cache-Control", cache_control)
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(expected_body.clone())
+                .file_name("upload.txt")
+                .mime_str("text/plain")?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
+        .await?;
+
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
+    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    assert_eq!(head.cache_control(), Some(cache_control));
+
+    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
+    let uploaded = get_out.body.collect().await?.into_bytes();
+    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_anonymous_post_object_accepts_content_language_field_exact_policy_match()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let bucket = "anon-post-policy-content-language-accept";
+    let object_key = "uploads/content-language-object.txt";
+    let content_language = "en-US";
+    let expected_body = b"post-policy-content-language-body".to_vec();
+
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(vec![
+        serde_json::json!({ "bucket": bucket }),
+        serde_json::json!({ "key": object_key }),
+        serde_json::json!({ "Content-Language": content_language }),
+        serde_json::json!(["content-length-range", 0, 1024]),
+    ]);
+
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text("Content-Language", content_language)
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(expected_body.clone())
+                .file_name("upload.txt")
+                .mime_str("text/plain")?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
+        .await?;
+
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
+    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    assert_eq!(head.content_language(), Some(content_language));
+
+    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
+    let uploaded = get_out.body.collect().await?.into_bytes();
+    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_anonymous_post_object_accepts_content_encoding_field_exact_policy_match()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let bucket = "anon-post-policy-content-encoding-accept";
+    let object_key = "uploads/content-encoding-object.txt";
+    let content_encoding = "gzip";
+    let expected_body = b"post-policy-content-encoding-body".to_vec();
+
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(vec![
+        serde_json::json!({ "bucket": bucket }),
+        serde_json::json!({ "key": object_key }),
+        serde_json::json!({ "Content-Encoding": content_encoding }),
+        serde_json::json!(["content-length-range", 0, 1024]),
+    ]);
+
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text("Content-Encoding", content_encoding)
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(expected_body.clone())
+                .file_name("upload.txt")
+                .mime_str("text/plain")?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
+        .await?;
+
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
+    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    assert_eq!(head.content_encoding(), Some(content_encoding));
+
+    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
+    let uploaded = get_out.body.collect().await?.into_bytes();
+    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_anonymous_post_object_accepts_website_redirect_location_exact_policy_match()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let bucket = "anon-post-policy-website-redirect-accept";
+    let object_key = "uploads/website-redirect-object.txt";
+    let website_redirect_location = "/docs/landing.html";
+    let expected_body = b"post-policy-website-redirect-body".to_vec();
+
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(vec![
+        serde_json::json!({ "bucket": bucket }),
+        serde_json::json!({ "key": object_key }),
+        serde_json::json!({ "x-amz-website-redirect-location": website_redirect_location }),
+        serde_json::json!(["content-length-range", 0, 1024]),
+    ]);
+
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text("x-amz-website-redirect-location", website_redirect_location)
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(expected_body.clone())
+                .file_name("upload.txt")
+                .mime_str("text/plain")?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
+        .await?;
+
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
+    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    assert_eq!(head.website_redirect_location(), Some(website_redirect_location));
+
+    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
+    let uploaded = get_out.body.collect().await?.into_bytes();
+    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_anonymous_post_object_accepts_expires_field_exact_policy_match()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let bucket = "anon-post-policy-expires-accept";
+    let object_key = "uploads/expires-object.txt";
+    let expires = "Wed, 21 Oct 2037 07:28:00 GMT";
+    let expected_body = b"post-policy-expires-body".to_vec();
+
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(vec![
+        serde_json::json!({ "bucket": bucket }),
+        serde_json::json!({ "key": object_key }),
+        serde_json::json!({ "Expires": expires }),
+        serde_json::json!(["content-length-range", 0, 1024]),
+    ]);
+
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text("Expires", expires)
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(expected_body.clone())
+                .file_name("upload.txt")
+                .mime_str("text/plain")?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
+        .await?;
+
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
+    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    assert_eq!(head.expires_string(), Some(expires));
+
+    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
+    let uploaded = get_out.body.collect().await?.into_bytes();
+    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
 
     Ok(())
 }
@@ -3123,6 +3440,64 @@ async fn test_anonymous_post_object_accepts_tagging_field_exact_policy_match()
     Ok(())
 }
 
+#[tokio::test]
+async fn test_anonymous_post_object_accepts_metadata_field_exact_policy_match()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let bucket = "anon-post-policy-meta-exact-accept";
+    let object_key = "uploads/meta-exact-accept-object.txt";
+    let metadata_value = "alpha-demo";
+    let expected_body = b"post-policy-meta-exact-body".to_vec();
+
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(vec![
+        serde_json::json!({ "bucket": bucket }),
+        serde_json::json!({ "key": object_key }),
+        serde_json::json!({ "x-amz-meta-project": metadata_value }),
+        serde_json::json!(["content-length-range", 0, 1024]),
+    ]);
+
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text("x-amz-meta-project", metadata_value)
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(expected_body.clone())
+                .file_name("upload.txt")
+                .mime_str("text/plain")?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
+        .await?;
+
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
+    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    let metadata = head.metadata().expect("head_object should expose uploaded metadata");
+    assert_eq!(metadata.get("project").map(String::as_str), Some(metadata_value));
+
+    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
+    let uploaded = get_out.body.collect().await?.into_bytes();
+    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_anonymous_post_object_allows_x_ignore_fields_outside_policy_conditions()
 -> Result<(), Box<dyn std::error::Error + Send + Sync>> {

crates/ecstore/src/data_usage/mod.rs

@@ -19,7 +19,7 @@ pub mod local_snapshot;
 use crate::storage_api_contracts::{
     bucket::{BucketOperations as _, BucketOptions},
     list::{ListOperations as _, StorageListObjectVersionsInfo},
-    object::{EcstoreObjectIO, HTTPPreconditions, ObjectOperations as _},
+    object::{EcstoreObjectIO, HTTPPreconditions, ObjectIO as _, ObjectOperations as _},
 };
 use crate::{
     bucket::{metadata_sys::get_replication_config, versioning::VersioningApi as _, versioning_sys::BucketVersioningSys},
@@ -2009,93 +2009,80 @@ pub async fn apply_bucket_usage_memory_overlay(data_usage_info: &mut DataUsageIn
 }
 
 // Helper functions for DataUsageCache operations
-
-/// How many times `load_data_usage_cache` tries a read that failed for a
-/// transient reason before giving up.
-const DATA_USAGE_CACHE_LOAD_ATTEMPTS: usize = 5;
-const DATA_USAGE_CACHE_LOAD_BASE_DELAY: std::time::Duration = std::time::Duration::from_millis(100);
-const DATA_USAGE_CACHE_LOAD_MAX_DELAY: std::time::Duration = std::time::Duration::from_millis(1_000);
-
-/// Result of one attempt at reading a data-usage cache object.
-enum DataUsageCacheRead {
-    Loaded(DataUsageCache),
-    /// The object is definitively not there, so retrying cannot turn the read
-    /// into a hit.
-    Absent,
-}
-
-/// True when the error means the cache object does not exist, as opposed to a
-/// transient failure that is worth another attempt.
-fn is_data_usage_cache_absent(err: &Error) -> bool {
-    matches!(err, Error::FileNotFound | Error::VolumeNotFound)
-}
-
-async fn read_data_usage_cache_object<S>(store: &S, key: &str) -> crate::error::Result<DataUsageCacheRead>
-where
-    S: EcstoreObjectIO,
-{
-    use crate::disk::RUSTFS_META_BUCKET;
+pub async fn load_data_usage_cache(store: &crate::set_disk::SetDisks, name: &str) -> crate::error::Result<DataUsageCache> {
+    use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
     use crate::object_api::ObjectOptions;
     use http::HeaderMap;
-
-    match store
-        .get_object_reader(
-            RUSTFS_META_BUCKET,
-            key,
-            None,
-            HeaderMap::new(),
-            &ObjectOptions {
-                no_lock: true,
-                ..Default::default()
-            },
-        )
-        .await
-    {
-        // A cache object that fails to decode is treated as absent rather than
-        // as an error: a corrupt cache should not stall the caller, and the
-        // next scanner pass rewrites it.
-        Ok(mut reader) => Ok(DataUsageCache::unmarshal(&reader.read_all().await?)
-            .map(DataUsageCacheRead::Loaded)
-            .unwrap_or(DataUsageCacheRead::Absent)),
-        Err(err) if is_data_usage_cache_absent(&err) => Ok(DataUsageCacheRead::Absent),
-        Err(err) => Err(err),
-    }
-}
-
-/// Load a data-usage cache, preferring the prefixed key and falling back to the
-/// legacy unprefixed one.
-///
-/// A cache that is absent under both keys yields an empty cache; a transient
-/// read failure is retried with capped, jittered backoff and surfaces as an
-/// error once the attempts are exhausted.
-pub(crate) async fn load_data_usage_cache<S>(store: &S, name: &str) -> crate::error::Result<DataUsageCache>
-where
-    S: EcstoreObjectIO,
-{
-    use crate::disk::BUCKET_META_PREFIX;
+    use rand::RngExt;
     use std::path::Path;
+    use std::time::Duration;
+    use tokio::time::sleep;
 
-    let prefixed = Path::new(BUCKET_META_PREFIX).join(name);
-    let prefixed = prefixed
-        .to_str()
-        .ok_or_else(|| Error::other("data usage cache path is not valid UTF-8"))?
-        .to_owned();
-
-    rustfs_utils::retry::retry_with_backoff(
-        || async {
-            match read_data_usage_cache_object(store, &prefixed).await? {
-                DataUsageCacheRead::Loaded(cache) => Ok(cache),
-                DataUsageCacheRead::Absent => match read_data_usage_cache_object(store, name).await? {
-                    DataUsageCacheRead::Loaded(cache) => Ok(cache),
-                    DataUsageCacheRead::Absent => Ok(DataUsageCache::default()),
+    let mut d = DataUsageCache::default();
+    let mut retries = 0;
+    while retries < 5 {
+        let path = Path::new(BUCKET_META_PREFIX).join(name);
+        match store
+            .get_object_reader(
+                RUSTFS_META_BUCKET,
+                path.to_str().unwrap(),
+                None,
+                HeaderMap::new(),
+                &ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
                 },
+            )
+            .await
+        {
+            Ok(mut reader) => {
+                if let Ok(info) = DataUsageCache::unmarshal(&reader.read_all().await?) {
+                    d = info
+                }
+                break;
             }
-        },
-        DATA_USAGE_CACHE_LOAD_ATTEMPTS,
-        DATA_USAGE_CACHE_LOAD_BASE_DELAY,
-        DATA_USAGE_CACHE_LOAD_MAX_DELAY,
-    )
-    .await
+            Err(err) => match err {
+                Error::FileNotFound | Error::VolumeNotFound => {
+                    match store
+                        .get_object_reader(
+                            RUSTFS_META_BUCKET,
+                            name,
+                            None,
+                            HeaderMap::new(),
+                            &ObjectOptions {
+                                no_lock: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                    {
+                        Ok(mut reader) => {
+                            if let Ok(info) = DataUsageCache::unmarshal(&reader.read_all().await?) {
+                                d = info
+                            }
+                            break;
+                        }
+                        Err(_) => match err {
+                            Error::FileNotFound | Error::VolumeNotFound => {
+                                break;
+                            }
+                            _ => {}
+                        },
+                    }
+                }
+                _ => {
+                    break;
+                }
+            },
+        }
+        retries += 1;
+        let dur = {
+            let mut rng = rand::rng();
+            rng.random_range(0..1_000)
+        };
+        sleep(Duration::from_millis(dur)).await;
+    }
+    Ok(d)
 }
 
 /// Persist the current in-memory compression total to the backend.
@@ -2233,7 +2220,6 @@ pub async fn init_compression_total_memory_from_backend(store: Arc<ECStore>) {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::storage_api_contracts::object::ObjectIO as _;
     use rustfs_data_usage::BucketUsageInfo;
     use rustfs_lock::{LocalClient, LockRequest, LockType, NamespaceLock, ObjectKey};
     use serial_test::serial;
@@ -2466,128 +2452,6 @@ mod tests {
         }
     }
 
-    /// Minimal ObjectIO backing `load_data_usage_cache` tests: records the keys
-    /// read and fails the first N reads with a transient (non-absence) error.
-    #[derive(Debug, Default)]
-    struct UsageCacheReadStore {
-        transient_failures: Mutex<usize>,
-        reads: Mutex<Vec<String>>,
-    }
-
-    impl UsageCacheReadStore {
-        fn failing_first(n: usize) -> Self {
-            Self {
-                transient_failures: Mutex::new(n),
-                reads: Mutex::new(Vec::new()),
-            }
-        }
-
-        async fn read_keys(&self) -> Vec<String> {
-            self.reads.lock().await.clone()
-        }
-    }
-
-    #[async_trait::async_trait]
-    impl crate::storage_api_contracts::object::ObjectIO for UsageCacheReadStore {
-        type Error = Error;
-        type RangeSpec = crate::storage_api_contracts::range::HTTPRangeSpec;
-        type HeaderMap = http::HeaderMap;
-        type ObjectOptions = ObjectOptions;
-        type ObjectInfo = ObjectInfo;
-        type GetObjectReader = crate::object_api::GetObjectReader;
-        type PutObjectReader = PutObjReader;
-
-        async fn get_object_reader(
-            &self,
-            _bucket: &str,
-            object: &str,
-            _range: Option<Self::RangeSpec>,
-            _h: Self::HeaderMap,
-            _opts: &Self::ObjectOptions,
-        ) -> Result<Self::GetObjectReader, Self::Error> {
-            self.reads.lock().await.push(object.to_string());
-            let mut remaining = self.transient_failures.lock().await;
-            if *remaining > 0 {
-                *remaining -= 1;
-                return Err(Error::other("transient read failure"));
-            }
-            Err(Error::FileNotFound)
-        }
-
-        async fn put_object(
-            &self,
-            _bucket: &str,
-            _object: &str,
-            _data: &mut Self::PutObjectReader,
-            _opts: &Self::ObjectOptions,
-        ) -> Result<Self::ObjectInfo, Self::Error> {
-            unimplemented!("load_data_usage_cache never writes")
-        }
-    }
-
-    fn prefixed_usage_key(name: &str) -> String {
-        std::path::Path::new(crate::disk::BUCKET_META_PREFIX)
-            .join(name)
-            .to_str()
-            .expect("utf-8 path")
-            .to_string()
-    }
-
-    #[tokio::test]
-    async fn load_data_usage_cache_treats_absence_as_an_empty_cache_without_retrying() {
-        let name = "usage-cache";
-        let store = UsageCacheReadStore::default();
-
-        let cache = load_data_usage_cache(&store, name).await.expect("absence is not an error");
-
-        assert!(cache.cache.is_empty());
-        assert_eq!(
-            store.read_keys().await,
-            vec![prefixed_usage_key(name), name.to_string()],
-            "the prefixed key is tried first, then the legacy one, and neither absence is retried"
-        );
-    }
-
-    #[tokio::test]
-    async fn load_data_usage_cache_retries_a_transient_failure() {
-        let name = "usage-cache";
-        // Two transient failures, then the object reads as absent.
-        let store = UsageCacheReadStore::failing_first(2);
-
-        let cache = load_data_usage_cache(&store, name)
-            .await
-            .expect("retry should reach the absent read");
-
-        assert!(cache.cache.is_empty());
-        assert_eq!(
-            store.read_keys().await,
-            vec![
-                prefixed_usage_key(name),
-                prefixed_usage_key(name),
-                prefixed_usage_key(name),
-                name.to_string(),
-            ],
-            "a transient failure retries the prefixed read rather than falling through"
-        );
-    }
-
-    #[tokio::test]
-    async fn load_data_usage_cache_surfaces_a_persistent_failure() {
-        let name = "usage-cache";
-        let store = UsageCacheReadStore::failing_first(usize::MAX);
-
-        let err = load_data_usage_cache(&store, name)
-            .await
-            .expect_err("an exhausted retry must not be reported as an empty cache");
-
-        assert!(err.to_string().contains("transient read failure"));
-        assert_eq!(
-            store.read_keys().await.len(),
-            DATA_USAGE_CACHE_LOAD_ATTEMPTS,
-            "every attempt is used before giving up"
-        );
-    }
-
     async fn clear_usage_memory_cache_for_test() {
         memory_cache().write().await.clear();
         *cache_updating().write().await = false;

crates/ecstore/src/diagnostics/admin_server_info.rs

@@ -679,7 +679,7 @@ async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32,
         if erasure_set.id == 0 {
             erasure_set.id = d.set_index;
             match load_data_usage_cache(
-                store.pools[d.pool_index as usize].disk_set[d.set_index as usize].as_ref(),
+                &store.pools[d.pool_index as usize].disk_set[d.set_index as usize].clone(),
                 DATA_USAGE_CACHE_NAME,
             )
             .await

crates/heal/src/heal/replacement_readiness.rs

@@ -157,219 +157,4 @@ mod tests {
         )
         .await;
     }
-
-    #[cfg(target_os = "linux")]
-    mod linux_privileged_tests {
-        use super::*;
-        use std::error::Error;
-        use std::path::Path;
-        use std::process::Command;
-
-        const ENABLE_ENV: &str = "RUSTFS_PRIVILEGED_MOUNT_READINESS_TESTS";
-        const NAMESPACE_ENV: &str = "RUSTFS_PRIVILEGED_MOUNT_READINESS_TESTS_IN_NAMESPACE";
-        const MOUNT_SIZE: &str = "size=32m,mode=0700";
-
-        struct MountGuard {
-            mounts: Vec<std::path::PathBuf>,
-        }
-
-        impl MountGuard {
-            fn new() -> Result<Self, Box<dyn Error + Send + Sync>> {
-                run_command("mount", &["--make-rprivate", "/"])?;
-                Ok(Self { mounts: Vec::new() })
-            }
-
-            fn mount_tmpfs(&mut self, target: &Path, label: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
-                mount_tmpfs(target, label)?;
-                self.mounts.push(target.to_path_buf());
-                Ok(())
-            }
-
-            fn mount_bind(&mut self, source: &Path, target: &Path) -> Result<(), Box<dyn Error + Send + Sync>> {
-                mount_bind(source, target)?;
-                self.mounts.push(target.to_path_buf());
-                Ok(())
-            }
-        }
-
-        impl Drop for MountGuard {
-            fn drop(&mut self) {
-                for mount in self.mounts.iter().rev() {
-                    let _ = detach_mount(mount);
-                }
-            }
-        }
-
-        fn run_command(program: &str, args: &[&str]) -> Result<(), Box<dyn Error + Send + Sync>> {
-            let output = Command::new(program).args(args).output()?;
-            if output.status.success() {
-                return Ok(());
-            }
-            Err(format!(
-                "{program} {} failed with status {}: stdout={} stderr={}",
-                args.join(" "),
-                output.status,
-                String::from_utf8_lossy(&output.stdout),
-                String::from_utf8_lossy(&output.stderr)
-            )
-            .into())
-        }
-
-        fn path_to_string(path: &Path, label: &str) -> Result<String, Box<dyn Error + Send + Sync>> {
-            path.to_str()
-                .map(str::to_owned)
-                .ok_or_else(|| format!("{label} path is not UTF-8: {path:?}").into())
-        }
-
-        fn mount_tmpfs(target: &Path, label: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
-            let target = path_to_string(target, "tmpfs target")?;
-            run_command("mount", &["-t", "tmpfs", "-o", MOUNT_SIZE, label, &target])
-        }
-
-        fn mount_bind(source: &Path, target: &Path) -> Result<(), Box<dyn Error + Send + Sync>> {
-            let source = path_to_string(source, "bind source")?;
-            let target = path_to_string(target, "bind target")?;
-            run_command("mount", &["--bind", &source, &target])
-        }
-
-        fn detach_mount(target: &Path) -> Result<(), Box<dyn Error + Send + Sync>> {
-            let target = path_to_string(target, "umount target")?;
-            run_command("umount", &[&target])
-        }
-
-        fn privileged_enabled() -> Result<bool, Box<dyn Error + Send + Sync>> {
-            let enabled = std::env::var(ENABLE_ENV)
-                .ok()
-                .is_some_and(|value| matches!(value.as_str(), "1" | "true" | "TRUE" | "yes" | "YES"));
-            if !enabled {
-                return Ok(false);
-            }
-            Ok(true)
-        }
-
-        fn run_current_test_in_mount_namespace() -> Result<(), Box<dyn Error + Send + Sync>> {
-            let test_name = std::thread::current()
-                .name()
-                .ok_or("privileged mount readiness test thread is unnamed")?
-                .to_owned();
-            let test_binary = std::env::current_exe()?;
-            let status = Command::new("unshare")
-                .arg("--mount")
-                .arg("--propagation")
-                .arg("private")
-                .arg("--")
-                .arg(test_binary)
-                .arg("--exact")
-                .arg(test_name)
-                .arg("--ignored")
-                .arg("--nocapture")
-                .env(NAMESPACE_ENV, "1")
-                .status()?;
-            if status.success() {
-                return Ok(());
-            }
-            Err(format!("{ENABLE_ENV}=1 requires Linux root or CAP_SYS_ADMIN; unshare exited with status {status}").into())
-        }
-
-        fn run_privileged_mount_test<F, Fut>(test: F) -> Result<(), Box<dyn Error + Send + Sync>>
-        where
-            F: FnOnce(MountGuard) -> Fut + Send + 'static,
-            Fut: std::future::Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + 'static,
-        {
-            if !privileged_enabled()? {
-                return Ok(());
-            }
-            if std::env::var_os(NAMESPACE_ENV).is_none() {
-                return run_current_test_in_mount_namespace();
-            }
-
-            let guard = MountGuard::new()?;
-            let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
-            runtime.block_on(test(guard))
-        }
-
-        #[test]
-        #[ignore = "requires Linux root/CAP_SYS_ADMIN and RUSTFS_PRIVILEGED_MOUNT_READINESS_TESTS=1"]
-        fn auto_replacement_readiness_accepts_an_independent_mount() -> Result<(), Box<dyn Error + Send + Sync>> {
-            run_privileged_mount_test(|mut mounts| async move {
-                let temp = TempDir::new().expect("temporary replacement roots should be created");
-                let target = temp.path().join("target");
-                let sibling = temp.path().join("sibling");
-                std::fs::create_dir(&target).expect("target mountpoint should be created");
-                std::fs::create_dir(&sibling).expect("sibling mountpoint should be created");
-                mounts.mount_tmpfs(&target, "rustfs-readiness-target")?;
-                mounts.mount_tmpfs(&sibling, "rustfs-readiness-sibling")?;
-
-                let target_endpoint = Endpoint::try_from(target.to_string_lossy().as_ref())?;
-                let sibling_endpoint = Endpoint::try_from(sibling.to_string_lossy().as_ref())?;
-                let target_disk = new_disk(
-                    &target_endpoint,
-                    &DiskOption {
-                        cleanup: false,
-                        health_check: false,
-                    },
-                )
-                .await?;
-                let sibling_disk = new_disk(
-                    &sibling_endpoint,
-                    &DiskOption {
-                        cleanup: false,
-                        health_check: false,
-                    },
-                )
-                .await?;
-
-                let identity = auto_replacement_target_identity(&target_disk, &[target_disk.clone(), sibling_disk.clone()]).await;
-                assert!(
-                    identity.is_some(),
-                    "a separately mounted replacement target with no sibling device overlap must be admitted"
-                );
-                Ok(())
-            })
-        }
-
-        #[test]
-        #[ignore = "requires Linux root/CAP_SYS_ADMIN and RUSTFS_PRIVILEGED_MOUNT_READINESS_TESTS=1"]
-        fn auto_replacement_readiness_rejects_a_same_device_sibling_bind_mount() -> Result<(), Box<dyn Error + Send + Sync>> {
-            run_privileged_mount_test(|mut mounts| async move {
-                let temp = TempDir::new().expect("temporary replacement roots should be created");
-                let source = temp.path().join("source");
-                let target = temp.path().join("target");
-                let sibling = temp.path().join("sibling");
-                std::fs::create_dir(&source).expect("source mountpoint should be created");
-                std::fs::create_dir(&target).expect("target mountpoint should be created");
-                std::fs::create_dir(&sibling).expect("sibling mountpoint should be created");
-                mounts.mount_tmpfs(&source, "rustfs-readiness-shared-source")?;
-                mounts.mount_bind(&source, &target)?;
-                mounts.mount_bind(&source, &sibling)?;
-
-                let target_endpoint = Endpoint::try_from(target.to_string_lossy().as_ref())?;
-                let sibling_endpoint = Endpoint::try_from(sibling.to_string_lossy().as_ref())?;
-                let target_disk = new_disk(
-                    &target_endpoint,
-                    &DiskOption {
-                        cleanup: false,
-                        health_check: false,
-                    },
-                )
-                .await?;
-                let sibling_disk = new_disk(
-                    &sibling_endpoint,
-                    &DiskOption {
-                        cleanup: false,
-                        health_check: false,
-                    },
-                )
-                .await?;
-
-                assert!(
-                    auto_replacement_target_identity(&target_disk, &[target_disk.clone(), sibling_disk.clone()])
-                        .await
-                        .is_none(),
-                    "replacement readiness must reject a target sharing its physical device with a sibling endpoint"
-                );
-                Ok(())
-            })
-        }
-    }
 }

crates/kms/src/lib.rs

@@ -255,10 +255,6 @@ pub use cache::KmsCacheStats;
 pub use config::*;
 pub use deletion_worker::DeletionReferenceChecker;
 pub use encryption::is_data_key_envelope;
-// Re-exported so the object layer binds encryption context exactly the way the
-// KMS backends do. A second canonicalization is how the object layer once
-// serialized a HashMap directly while the Static backend already sorted keys.
-pub use encryption::context_aad;
 pub use error::{KmsError, KmsUnavailableError, Result};
 pub use key_impact::{KeyImpactReport, KeyReference, KeyReferenceKind, ReferenceCompleteness, ReferenceCoverage, ReferenceScope};
 pub use manager::KmsManager;

docs/testing/e2e-suite-inventory.md

@@ -63,7 +63,7 @@
 | list_objects_v2_metadata_extension_test | 1 |  |
 | list_objects_v2_pagination_test | 12 | ✅ |
 | mc_mirror_small_bucket_test | 1 |  |
-| multipart_auth_test | 75 |  |
+| multipart_auth_test | 85 |  |
 | multipart_storage_class_test | 3 | ✅ |
 | namespace_lock_quorum_test | 2 |  |
 | negative_sigv4_test | 6 | ✅ |

rustfs/src/app/object_usecase.rs

@@ -169,8 +169,8 @@ use s3s::dto::{
     ObjectLockLegalHoldStatus, ObjectLockMode, ObjectLockRetention, ObjectLockRetentionMode, ObjectPart, PutObjectInput,
     PutObjectOutput, Range, RequestCharged, RestoreObjectInput, RestoreObjectOutput, RestoreStatus, SSECustomerAlgorithm,
     SSECustomerKeyMD5, SSEKMSKeyId, SelectObjectContentInput, SelectObjectContentOutput, ServerSideEncryption,
-    ServerSideEncryptionByDefault, StorageClass, StreamingBlob, TaggingDirective, TaggingHeader, Timestamp, TimestampFormat,
-    WebsiteRedirectLocation,
+    ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, StorageClass, StreamingBlob, TaggingDirective,
+    TaggingHeader, Timestamp, TimestampFormat, WebsiteRedirectLocation,
 };
 use s3s::header::{X_AMZ_RESTORE, X_AMZ_RESTORE_OUTPUT_PATH};
 use s3s::stream::{ByteStream, DynByteStream, RemainingLength};
@@ -2695,6 +2695,39 @@ fn bucket_default_write_sse(sse: &ServerSideEncryptionByDefault) -> ServerSideEn
     }
 }
 
+/// Resolve the effective server-side encryption for a write against the bucket's
+/// default encryption configuration.
+///
+/// A request-level value always wins; the bucket default only fills a gap, and
+/// the unknown-algorithm fallback lives once in [`bucket_default_write_sse`].
+///
+/// `has_explicit_ssec` suppresses the default entirely. Only COPY passes `true`
+/// today: its destination may carry SSE-C, which must not also be given managed
+/// encryption. PUT and extract pass `false`, matching their current behaviour —
+/// see backlog#1826 for the divergence that leaves.
+///
+/// Callers layering further overrides (PUT's `ciphertext_passthrough`) apply
+/// them to the returned pair.
+fn resolve_bucket_default_sse(
+    bucket_sse_config: Option<&ServerSideEncryptionConfiguration>,
+    requested_sse: Option<ServerSideEncryption>,
+    requested_kms_key_id: Option<SSEKMSKeyId>,
+    has_explicit_ssec: bool,
+) -> (Option<ServerSideEncryption>, Option<SSEKMSKeyId>) {
+    let bucket_default = || {
+        if has_explicit_ssec {
+            return None;
+        }
+        bucket_sse_config
+            .and_then(|config| config.rules.first())
+            .and_then(|rule| rule.apply_server_side_encryption_by_default.as_ref())
+    };
+
+    let effective_sse = requested_sse.or_else(|| bucket_default().map(bucket_default_write_sse));
+    let effective_kms_key_id = requested_kms_key_id.or_else(|| bucket_default().and_then(|sse| sse.kms_master_key_id.clone()));
+    (effective_sse, effective_kms_key_id)
+}
+
 fn should_use_small_eager_put_path(
     size: i64,
     headers: &HeaderMap,
@@ -5823,19 +5856,12 @@ impl DefaultObjectUsecase {
         );
 
         let original_sse = server_side_encryption.clone();
-        let mut effective_sse = server_side_encryption.or_else(|| {
-            bucket_sse_config.as_ref().and_then(|(config, _timestamp)| {
-                config.rules.first().and_then(|rule| {
-                    rule.apply_server_side_encryption_by_default.as_ref().map(|sse| {
-                        match sse.sse_algorithm.as_str() {
-                            "AES256" => ServerSideEncryption::from_static(ServerSideEncryption::AES256),
-                            "aws:kms" => ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS),
-                            _ => ServerSideEncryption::from_static(ServerSideEncryption::AES256), // fallback to AES256
-                        }
-                    })
-                })
-            })
-        });
+        let (mut effective_sse, mut effective_kms_key_id) = resolve_bucket_default_sse(
+            bucket_sse_config.as_ref().map(|(config, _timestamp)| config),
+            server_side_encryption,
+            ssekms_key_id,
+            false,
+        );
         debug!(
             target: "rustfs::app::object_usecase",
             component = "app",
@@ -5847,16 +5873,6 @@ impl DefaultObjectUsecase {
             "Resolved effective SSE configuration"
         );
 
-        let mut effective_kms_key_id = ssekms_key_id.or_else(|| {
-            bucket_sse_config.as_ref().and_then(|(config, _timestamp)| {
-                config.rules.first().and_then(|rule| {
-                    rule.apply_server_side_encryption_by_default
-                        .as_ref()
-                        .and_then(|sse| sse.kms_master_key_id.clone())
-                })
-            })
-        });
-
         if ciphertext_passthrough {
             // The replica keeps the source's SSE-C metadata; the bucket
             // default must not claim managed encryption on it.
@@ -7658,30 +7674,12 @@ impl DefaultObjectUsecase {
             }
         };
 
-        let mut effective_sse = requested_sse.or_else(|| {
-            if has_explicit_ssec {
-                return None;
-            }
-            bucket_sse_config.as_ref().and_then(|(config, _)| {
-                config.rules.first().and_then(|rule| {
-                    rule.apply_server_side_encryption_by_default
-                        .as_ref()
-                        .map(bucket_default_write_sse)
-                })
-            })
-        });
-        let mut effective_kms_key_id = requested_kms_key_id.or_else(|| {
-            if has_explicit_ssec {
-                return None;
-            }
-            bucket_sse_config.as_ref().and_then(|(config, _)| {
-                config.rules.first().and_then(|rule| {
-                    rule.apply_server_side_encryption_by_default
-                        .as_ref()
-                        .and_then(|sse| sse.kms_master_key_id.clone())
-                })
-            })
-        });
+        let (mut effective_sse, mut effective_kms_key_id) = resolve_bucket_default_sse(
+            bucket_sse_config.as_ref().map(|(config, _)| config),
+            requested_sse,
+            requested_kms_key_id,
+            has_explicit_ssec,
+        );
 
         let h = build_ssec_read_headers(
             copy_source_sse_customer_algorithm.as_ref(),
@@ -9587,28 +9585,12 @@ impl DefaultObjectUsecase {
 
         let original_sse = server_side_encryption.or(extract_server_side_encryption_from_headers(&req.headers)?);
         let bucket_sse_config = metadata_sys::get_sse_config(&bucket).await.ok();
-        let mut effective_sse = original_sse.or_else(|| {
-            bucket_sse_config.as_ref().and_then(|(config, _timestamp)| {
-                config.rules.first().and_then(|rule| {
-                    rule.apply_server_side_encryption_by_default
-                        .as_ref()
-                        .map(|sse| match sse.sse_algorithm.as_str() {
-                            "AES256" => ServerSideEncryption::from_static(ServerSideEncryption::AES256),
-                            "aws:kms" => ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS),
-                            _ => ServerSideEncryption::from_static(ServerSideEncryption::AES256),
-                        })
-                })
-            })
-        });
-        let mut effective_kms_key_id = ssekms_key_id.or_else(|| {
-            bucket_sse_config.as_ref().and_then(|(config, _timestamp)| {
-                config.rules.first().and_then(|rule| {
-                    rule.apply_server_side_encryption_by_default
-                        .as_ref()
-                        .and_then(|sse| sse.kms_master_key_id.clone())
-                })
-            })
-        });
+        let (mut effective_sse, mut effective_kms_key_id) = resolve_bucket_default_sse(
+            bucket_sse_config.as_ref().map(|(config, _timestamp)| config),
+            original_sse,
+            ssekms_key_id,
+            false,
+        );
         if effective_sse
             .as_ref()
             .is_some_and(|sse| sse.as_str().eq_ignore_ascii_case(ServerSideEncryption::AWS_KMS))
@@ -10397,6 +10379,74 @@ mod tests {
         }
     }
 
+    fn bucket_sse_config_with(algorithm: &str, kms_key_id: Option<&str>) -> ServerSideEncryptionConfiguration {
+        ServerSideEncryptionConfiguration {
+            rules: vec![ServerSideEncryptionRule {
+                apply_server_side_encryption_by_default: Some(ServerSideEncryptionByDefault {
+                    sse_algorithm: ServerSideEncryption::from(String::from(algorithm)),
+                    kms_master_key_id: kms_key_id.map(|id| SSEKMSKeyId::from(id.to_string())),
+                }),
+                bucket_key_enabled: None,
+            }],
+        }
+    }
+
+    #[test]
+    fn resolve_bucket_default_sse_prefers_the_request_over_the_bucket_default() {
+        let config = bucket_sse_config_with(ServerSideEncryption::AWS_KMS, Some("bucket-key"));
+
+        let (sse, kms_key_id) = resolve_bucket_default_sse(
+            Some(&config),
+            Some(ServerSideEncryption::from_static(ServerSideEncryption::AES256)),
+            Some(SSEKMSKeyId::from("request-key".to_string())),
+            false,
+        );
+
+        assert_eq!(sse.as_ref().map(|sse| sse.as_str()), Some(ServerSideEncryption::AES256));
+        assert_eq!(kms_key_id.as_deref(), Some("request-key"));
+    }
+
+    #[test]
+    fn resolve_bucket_default_sse_fills_gaps_from_the_bucket_default() {
+        let config = bucket_sse_config_with(ServerSideEncryption::AWS_KMS, Some("bucket-key"));
+
+        let (sse, kms_key_id) = resolve_bucket_default_sse(Some(&config), None, None, false);
+
+        assert_eq!(sse.as_ref().map(|sse| sse.as_str()), Some(ServerSideEncryption::AWS_KMS));
+        assert_eq!(kms_key_id.as_deref(), Some("bucket-key"));
+    }
+
+    #[test]
+    fn resolve_bucket_default_sse_falls_back_to_aes256_for_an_unknown_algorithm() {
+        // Reachable only through corrupt or hand-edited bucket metadata;
+        // PutBucketEncryption rejects unknown algorithms. All three call sites
+        // now share this single decision (backlog#1826).
+        let config = bucket_sse_config_with("garbage", None);
+
+        let (sse, kms_key_id) = resolve_bucket_default_sse(Some(&config), None, None, false);
+
+        assert_eq!(sse.as_ref().map(|sse| sse.as_str()), Some(ServerSideEncryption::AES256));
+        assert!(kms_key_id.is_none());
+    }
+
+    #[test]
+    fn resolve_bucket_default_sse_suppresses_the_default_for_explicit_ssec() {
+        let config = bucket_sse_config_with(ServerSideEncryption::AES256, Some("bucket-key"));
+
+        let (sse, kms_key_id) = resolve_bucket_default_sse(Some(&config), None, None, true);
+
+        assert!(sse.is_none(), "an SSE-C destination must not also get managed encryption");
+        assert!(kms_key_id.is_none());
+    }
+
+    #[test]
+    fn resolve_bucket_default_sse_returns_nothing_without_a_bucket_default() {
+        let (sse, kms_key_id) = resolve_bucket_default_sse(None, None, None, false);
+
+        assert!(sse.is_none());
+        assert!(kms_key_id.is_none());
+    }
+
     #[test]
     fn put_request_user_metadata_cannot_suppress_bucket_default_retention() {
         let mut metadata =

rustfs/src/server/layer.rs

@@ -747,12 +747,11 @@ pub struct S3ErrorMessageCompatService<S> {
     inner: S,
 }
 
-impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for S3ErrorMessageCompatService<S>
+impl<S, RestBody, GrpcBody> Service<HttpRequest<Incoming>> for S3ErrorMessageCompatService<S>
 where
-    S: Service<HttpRequest<ReqBody>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
+    S: Service<HttpRequest<Incoming>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
     S::Future: Send + 'static,
     S::Error: Send + 'static,
-    ReqBody: Send + 'static,
     RestBody: Body<Data = Bytes> + From<Bytes> + Send + 'static,
     RestBody::Error: Into<S::Error> + Send + 'static,
     GrpcBody: Send + 'static,
@@ -765,27 +764,28 @@ where
         self.inner.poll_ready(cx)
     }
 
-    fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
+    fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
         let is_sts_query =
             req.method() == Method::POST && req.uri().path() == "/" && req.extensions().get::<StsQueryRequest>().is_some();
         let mut inner = self.inner.clone();
 
         Box::pin(async move {
             let response = inner.call(req).await?;
-            if is_sts_query || response.status() != StatusCode::FORBIDDEN || !is_xml_response(response.headers()) {
-                return Ok(response);
-            }
-
             let (parts, body) = response.into_parts();
+            let should_fix = !is_sts_query && parts.status == StatusCode::FORBIDDEN && is_xml_response(&parts.headers);
 
             let response = match body {
                 HybridBody::Rest { rest_body } => {
-                    let (rest_body, changed) = fix_s3_error_message_in_xml(rest_body).await.map_err(Into::into)?;
-                    let mut parts = parts;
-                    if changed {
-                        parts.headers.remove(http::header::CONTENT_LENGTH);
+                    if !should_fix {
+                        Response::from_parts(parts, HybridBody::Rest { rest_body })
+                    } else {
+                        let (rest_body, changed) = fix_s3_error_message_in_xml(rest_body).await.map_err(Into::into)?;
+                        let mut parts = parts;
+                        if changed {
+                            parts.headers.remove(http::header::CONTENT_LENGTH);
+                        }
+                        Response::from_parts(parts, HybridBody::Rest { rest_body })
                     }
-                    Response::from_parts(parts, HybridBody::Rest { rest_body })
                 }
                 HybridBody::Grpc { grpc_body } => Response::from_parts(parts, HybridBody::Grpc { grpc_body }),
             };
@@ -886,12 +886,11 @@ pub struct IcebergRestErrorCompatService<S> {
     inner: S,
 }
 
-impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for IcebergRestErrorCompatService<S>
+impl<S, RestBody, GrpcBody> Service<HttpRequest<Incoming>> for IcebergRestErrorCompatService<S>
 where
-    S: Service<HttpRequest<ReqBody>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
+    S: Service<HttpRequest<Incoming>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
     S::Future: Send + 'static,
     S::Error: Send + 'static,
-    ReqBody: Send + 'static,
     RestBody: Body<Data = Bytes> + From<Bytes> + Send + 'static,
     RestBody::Error: Into<S::Error> + Send + 'static,
     GrpcBody: Send + 'static,
@@ -904,21 +903,18 @@ where
         self.inner.poll_ready(cx)
     }
 
-    fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
+    fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
         let catalog_path =
             (req.method() != Method::HEAD && is_table_catalog_path(req.uri().path())).then(|| req.uri().path().to_string());
         let mut inner = self.inner.clone();
 
         Box::pin(async move {
             let response = inner.call(req).await?;
-            if catalog_path.is_none() || response.status().is_success() || !is_xml_response(response.headers()) {
-                return Ok(response);
-            }
-
             let (parts, body) = response.into_parts();
+            let should_convert = catalog_path.is_some() && !parts.status.is_success() && is_xml_response(&parts.headers);
 
             let response = match body {
-                HybridBody::Rest { rest_body } => {
+                HybridBody::Rest { rest_body } if should_convert => {
                     let (rest_body, converted_status) = convert_iceberg_error_in_xml(
                         rest_body,
                         parts.status,
@@ -936,6 +932,7 @@ where
                     }
                     Response::from_parts(parts, HybridBody::Rest { rest_body })
                 }
+                HybridBody::Rest { rest_body } => Response::from_parts(parts, HybridBody::Rest { rest_body }),
                 HybridBody::Grpc { grpc_body } => Response::from_parts(parts, HybridBody::Grpc { grpc_body }),
             };
 
@@ -1048,12 +1045,11 @@ pub struct ObjectAttributesEtagFixService<S> {
     inner: S,
 }
 
-impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for ObjectAttributesEtagFixService<S>
+impl<S, RestBody, GrpcBody> Service<HttpRequest<Incoming>> for ObjectAttributesEtagFixService<S>
 where
-    S: Service<HttpRequest<ReqBody>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
+    S: Service<HttpRequest<Incoming>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
     S::Future: Send + 'static,
     S::Error: Send + 'static,
-    ReqBody: Send + 'static,
     RestBody: Body<Data = Bytes> + From<Bytes> + Send + 'static,
     RestBody::Error: Into<S::Error> + Send + 'static,
     GrpcBody: Send + 'static,
@@ -1066,26 +1062,27 @@ where
         self.inner.poll_ready(cx)
     }
 
-    fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
+    fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
         let is_target = is_object_attributes_request(&req);
         let mut inner = self.inner.clone();
 
         Box::pin(async move {
             let response = inner.call(req).await?;
-            if !is_target || !response.status().is_success() || !is_xml_response(response.headers()) {
-                return Ok(response);
-            }
-
             let (parts, body) = response.into_parts();
+            let should_fix = is_target && parts.status.is_success() && is_xml_response(&parts.headers);
 
             let response = match body {
                 HybridBody::Rest { rest_body } => {
-                    let rest_body = fix_object_attributes_etag_in_xml(rest_body).await.map_err(Into::into)?;
+                    if !should_fix {
+                        Response::from_parts(parts, HybridBody::Rest { rest_body })
+                    } else {
+                        let rest_body = fix_object_attributes_etag_in_xml(rest_body).await.map_err(Into::into)?;
 
-                    let mut parts = parts;
-                    parts.headers.remove(http::header::CONTENT_LENGTH);
+                        let mut parts = parts;
+                        parts.headers.remove(http::header::CONTENT_LENGTH);
 
-                    Response::from_parts(parts, HybridBody::Rest { rest_body })
+                        Response::from_parts(parts, HybridBody::Rest { rest_body })
+                    }
                 }
                 HybridBody::Grpc { grpc_body } => Response::from_parts(parts, HybridBody::Grpc { grpc_body }),
             };
@@ -1147,11 +1144,12 @@ where
 
         Box::pin(async move {
             let response = inner.call(req).await?;
-            if !is_bodyless_status(response.status()) {
-                return Ok(response);
+            let (mut parts, body) = response.into_parts();
+
+            if !is_bodyless_status(parts.status) {
+                return Ok(Response::from_parts(parts, body));
             }
 
-            let (mut parts, body) = response.into_parts();
             let response = match body {
                 HybridBody::Rest { .. } => {
                     parts.headers.remove(http::header::CONTENT_LENGTH);
@@ -1804,7 +1802,7 @@ fn strip_quotes_from_first_etag(xml: String) -> String {
     fixed
 }
 
-fn is_object_attributes_request<B>(req: &HttpRequest<B>) -> bool {
+fn is_object_attributes_request(req: &HttpRequest<Incoming>) -> bool {
     if req.method() != Method::GET {
         return false;
     }
@@ -1969,12 +1967,11 @@ fn apply_bucket_cors_result(response_headers: &mut HeaderMap, bucket_cors_header
     }
 }
 
-impl<S, ReqBody, ResBody> Service<HttpRequest<ReqBody>> for ConditionalCorsService<S>
+impl<S, ResBody> Service<HttpRequest<Incoming>> for ConditionalCorsService<S>
 where
-    S: Service<HttpRequest<ReqBody>, Response = Response<ResBody>> + Clone + Send + 'static,
+    S: Service<HttpRequest<Incoming>, Response = Response<ResBody>> + Clone + Send + 'static,
     S::Future: Send + 'static,
     S::Error: Into<Box<dyn std::error::Error + Send + Sync>> + Send + 'static,
-    ReqBody: Send + 'static,
     ResBody: Default + Send + 'static,
 {
     type Response = Response<ResBody>;
@@ -1985,14 +1982,7 @@ where
         self.inner.poll_ready(cx).map_err(Into::into)
     }
 
-    fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
-        let is_options = req.method() == Method::OPTIONS;
-        let has_origin = req.headers().contains_key(cors::standard::ORIGIN);
-        if !is_options && !has_origin {
-            let mut inner = self.inner.clone();
-            return Box::pin(async move { inner.call(req).await.map_err(Into::into) });
-        }
-
+    fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
         let path = req.uri().path().to_string();
         let method = req.method().clone();
         let request_headers = req.headers().clone();
@@ -2000,7 +1990,7 @@ where
         let is_s3 = ConditionalCorsLayer::is_s3_path(&path);
         let is_root = path == "/";
 
-        if is_options {
+        if method == Method::OPTIONS {
             let has_acrm = request_headers.contains_key(cors::request::ACCESS_CONTROL_REQUEST_METHOD);
 
             if is_root {
@@ -2202,7 +2192,6 @@ mod tests {
     use futures::future::{Ready, ready};
     use http::Request;
     use http_body_util::BodyExt;
-    use http_body_util::Empty;
     use http_body_util::Full;
     use opentelemetry::global;
     use opentelemetry_sdk::propagation::TraceContextPropagator;
@@ -3794,188 +3783,6 @@ mod tests {
         assert_eq!(bytes, input);
     }
 
-    #[derive(Clone)]
-    struct FixedHybridResponse {
-        status: StatusCode,
-        body: Bytes,
-        content_type: &'static str,
-    }
-
-    impl<B: Send + 'static> Service<Request<B>> for FixedHybridResponse {
-        type Response = Response<HybridBody<Full<Bytes>, Empty<Bytes>>>;
-        type Error = Infallible;
-        type Future = Ready<Result<Self::Response, Self::Error>>;
-
-        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-            Poll::Ready(Ok(()))
-        }
-
-        fn call(&mut self, _req: Request<B>) -> Self::Future {
-            let body = self.body.clone();
-            ready(Ok(Response::builder()
-                .status(self.status)
-                .header(http::header::CONTENT_TYPE, self.content_type)
-                .header(http::header::CONTENT_LENGTH, body.len().to_string())
-                .body(HybridBody::Rest {
-                    rest_body: Full::from(body),
-                })
-                .expect("fixed hybrid response")))
-        }
-    }
-
-    async fn collect_hybrid_response(
-        response: Response<HybridBody<Full<Bytes>, Empty<Bytes>>>,
-    ) -> (StatusCode, HeaderMap, String) {
-        let status = response.status();
-        let headers = response.headers().clone();
-        let body = BodyExt::collect(response.into_body())
-            .await
-            .expect("collect hybrid body")
-            .to_bytes();
-        (
-            status,
-            headers,
-            String::from_utf8(body.to_vec()).expect("hybrid response body should be UTF-8"),
-        )
-    }
-
-    #[tokio::test]
-    async fn s3_error_message_compat_fixes_regular_forbidden_xml() {
-        let body = Bytes::from_static(b"<Error><Code>SignatureDoesNotMatch</Code></Error>");
-        let mut service = S3ErrorMessageCompatLayer.layer(FixedHybridResponse {
-            status: StatusCode::FORBIDDEN,
-            body,
-            content_type: "application/xml",
-        });
-        let request = Request::builder()
-            .method(Method::GET)
-            .uri("/bucket/object")
-            .body(())
-            .expect("request");
-
-        let response = service.call(request).await.expect("service response");
-        let (status, headers, body) = collect_hybrid_response(response).await;
-
-        assert_eq!(status, StatusCode::FORBIDDEN);
-        assert!(headers.get(http::header::CONTENT_LENGTH).is_none());
-        assert!(body.contains("<Message>"));
-    }
-
-    #[tokio::test]
-    async fn s3_error_message_compat_leaves_sts_query_response_unchanged() {
-        let input = Bytes::from_static(b"<Error><Code>SignatureDoesNotMatch</Code></Error>");
-        let mut service = S3ErrorMessageCompatLayer.layer(FixedHybridResponse {
-            status: StatusCode::FORBIDDEN,
-            body: input.clone(),
-            content_type: "application/xml",
-        });
-        let mut request = Request::builder().method(Method::POST).uri("/").body(()).expect("request");
-        request.extensions_mut().insert(StsQueryRequest);
-
-        let response = service.call(request).await.expect("service response");
-        let (_status, headers, body) = collect_hybrid_response(response).await;
-
-        let expected_len = input.len().to_string();
-        assert_eq!(
-            headers
-                .get(http::header::CONTENT_LENGTH)
-                .and_then(|value| value.to_str().ok()),
-            Some(expected_len.as_str())
-        );
-        assert_eq!(body.as_bytes(), input.as_ref());
-    }
-
-    #[tokio::test]
-    async fn iceberg_rest_error_compat_converts_catalog_xml_errors() {
-        let mut service = IcebergRestErrorCompatLayer.layer(FixedHybridResponse {
-            status: StatusCode::NOT_FOUND,
-            body: Bytes::from_static(b"<Error><Code>NoSuchTableException</Code><Message>missing</Message></Error>"),
-            content_type: "application/xml",
-        });
-        let request = Request::builder()
-            .method(Method::GET)
-            .uri("/iceberg/v1/warehouse/namespaces/ns/tables/events")
-            .body(())
-            .expect("request");
-
-        let response = service.call(request).await.expect("service response");
-        let (status, headers, body) = collect_hybrid_response(response).await;
-
-        assert_eq!(status, StatusCode::NOT_FOUND);
-        assert_eq!(headers.get(http::header::CONTENT_TYPE).unwrap(), "application/json");
-        assert!(headers.get(http::header::CONTENT_LENGTH).is_none());
-        assert!(body.contains("\"type\":\"NoSuchTableException\""));
-    }
-
-    #[tokio::test]
-    async fn iceberg_rest_error_compat_leaves_non_catalog_errors_unchanged() {
-        let input = Bytes::from_static(b"<Error><Code>NoSuchKey</Code><Message>missing</Message></Error>");
-        let mut service = IcebergRestErrorCompatLayer.layer(FixedHybridResponse {
-            status: StatusCode::NOT_FOUND,
-            body: input.clone(),
-            content_type: "application/xml",
-        });
-        let request = Request::builder()
-            .method(Method::GET)
-            .uri("/bucket/object")
-            .body(())
-            .expect("request");
-
-        let response = service.call(request).await.expect("service response");
-        let (status, headers, body) = collect_hybrid_response(response).await;
-
-        assert_eq!(status, StatusCode::NOT_FOUND);
-        assert_eq!(headers.get(http::header::CONTENT_TYPE).unwrap(), "application/xml");
-        assert_eq!(body.as_bytes(), input.as_ref());
-    }
-
-    #[tokio::test]
-    async fn object_attributes_etag_fix_rewrites_target_response() {
-        let mut service = ObjectAttributesEtagFixLayer.layer(FixedHybridResponse {
-            status: StatusCode::OK,
-            body: Bytes::from_static(b"<GetObjectAttributesOutput><ETag>\"abc\"</ETag></GetObjectAttributesOutput>"),
-            content_type: "application/xml",
-        });
-        let request = Request::builder()
-            .method(Method::GET)
-            .uri("/bucket/object?attributes")
-            .body(())
-            .expect("request");
-
-        let response = service.call(request).await.expect("service response");
-        let (_status, headers, body) = collect_hybrid_response(response).await;
-
-        assert!(headers.get(http::header::CONTENT_LENGTH).is_none());
-        assert!(body.contains("<ETag>abc</ETag>"));
-    }
-
-    #[tokio::test]
-    async fn object_attributes_etag_fix_leaves_regular_get_unchanged() {
-        let input = Bytes::from_static(b"<GetObjectAttributesOutput><ETag>\"abc\"</ETag></GetObjectAttributesOutput>");
-        let mut service = ObjectAttributesEtagFixLayer.layer(FixedHybridResponse {
-            status: StatusCode::OK,
-            body: input.clone(),
-            content_type: "application/xml",
-        });
-        let request = Request::builder()
-            .method(Method::GET)
-            .uri("/bucket/object")
-            .body(())
-            .expect("request");
-
-        let response = service.call(request).await.expect("service response");
-        let (_status, headers, body) = collect_hybrid_response(response).await;
-
-        let expected_len = input.len().to_string();
-        assert_eq!(
-            headers
-                .get(http::header::CONTENT_LENGTH)
-                .and_then(|value| value.to_str().ok()),
-            Some(expected_len.as_str())
-        );
-        assert_eq!(body.as_bytes(), input.as_ref());
-    }
-
     #[derive(Clone)]
     struct FixedStsResponse {
         status: StatusCode,
@@ -4463,63 +4270,6 @@ mod tests {
         });
     }
 
-    #[derive(Clone)]
-    struct CorsOkService;
-
-    impl<B> Service<Request<B>> for CorsOkService {
-        type Response = Response<Empty<Bytes>>;
-        type Error = Infallible;
-        type Future = Ready<Result<Self::Response, Self::Error>>;
-
-        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-            Poll::Ready(Ok(()))
-        }
-
-        fn call(&mut self, _req: Request<B>) -> Self::Future {
-            ready(Ok(Response::builder()
-                .status(StatusCode::OK)
-                .body(Empty::new())
-                .expect("response")))
-        }
-    }
-
-    #[tokio::test]
-    async fn conditional_cors_passthrough_without_origin() {
-        let layer = ConditionalCorsLayer {
-            cors_origins: Some("*".to_string()),
-        };
-        let mut service = layer.layer(CorsOkService);
-        let request = Request::builder()
-            .method(Method::GET)
-            .uri("/bucket/object")
-            .body(())
-            .expect("request");
-
-        let response = service.call(request).await.expect("response");
-
-        assert_eq!(response.status(), StatusCode::OK);
-        assert!(response.headers().get(cors::response::ACCESS_CONTROL_ALLOW_ORIGIN).is_none());
-    }
-
-    #[tokio::test]
-    async fn conditional_cors_applies_origin_headers() {
-        let layer = ConditionalCorsLayer {
-            cors_origins: Some("*".to_string()),
-        };
-        let mut service = layer.layer(CorsOkService);
-        let request = Request::builder()
-            .method(Method::GET)
-            .uri("/bucket/object")
-            .header(cors::standard::ORIGIN, "https://example.com")
-            .body(())
-            .expect("request");
-
-        let response = service.call(request).await.expect("response");
-
-        assert_eq!(response.status(), StatusCode::OK);
-        assert_eq!(response.headers().get(cors::response::ACCESS_CONTROL_ALLOW_ORIGIN).unwrap(), "*");
-    }
-
     #[test]
     fn request_context_layer_populates_context_without_mutating_signed_headers() {
         let mut service = RequestContextLayer.layer(CaptureService);

rustfs/src/storage/concurrency/io_schedule.rs

@@ -1766,9 +1766,11 @@ mod tests {
     };
     use rustfs_io_core::io_profile::{AccessPattern, StorageMedia};
     use rustfs_io_metrics::bandwidth::{BandwidthSnapshot, BandwidthTier};
+    use serial_test::serial;
     use std::time::Duration;
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_queue_basic() {
         let config = IoPriorityQueueConfig::default();
         let queue = IoPriorityQueue::new(config);
@@ -1787,6 +1789,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_queue_dequeue_order() {
         let config = IoPriorityQueueConfig::default();
         let queue = IoPriorityQueue::new(config);
@@ -1814,6 +1817,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_queue_status() {
         let config = IoPriorityQueueConfig::default();
         let queue = IoPriorityQueue::new(config);
@@ -1831,6 +1835,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_queue_starvation_prevention() {
         let config = IoPriorityQueueConfig {
             starvation_threshold_secs: 1,
@@ -1854,6 +1859,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_from_size() {
         // High priority: < 1MB
         assert_eq!(IoPriority::from_size(100 * 1024), IoPriority::High);
@@ -1869,6 +1875,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_load_level_from_wait_duration() {
         use std::time::Duration;
 
@@ -1886,6 +1893,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_scheduler_config_default() {
         let config = IoSchedulerConfig::default();
 
@@ -1899,6 +1907,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_scheduler_config_to_core_config() {
         let config = IoSchedulerConfig::default();
         let core = config.to_core_config();
@@ -1914,6 +1923,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_queue_config_to_core_config() {
         let config = IoPriorityQueueConfig::default();
         let core = config.to_core_config();
@@ -1925,6 +1935,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_io_priority_queue_config_from_scheduler_config() {
         let scheduler_config = IoSchedulerConfig {
             queue_high_capacity: 128,
@@ -1947,6 +1958,7 @@ mod tests {
     // ============================================
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_nvme_sequential_low_load() {
         // NVMe + Sequential + Low load = maximum buffer size
         let context = IoSchedulingContext {
@@ -1973,6 +1985,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_hdd_random_high_load() {
         // HDD + Random + High load = conservative buffer size
         let context = IoSchedulingContext {
@@ -1999,6 +2012,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_ssd_mixed_medium_load() {
         // SSD + Mixed + Medium load = moderate buffer
         let context = IoSchedulingContext {
@@ -2026,6 +2040,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_critical_load_disables_features() {
         // Any media + Critical load = minimal features
         let context = IoSchedulingContext {
@@ -2050,6 +2065,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_buffer_cap_enforcement() {
         // Test that storage media caps are enforced
         let context = IoSchedulingContext {
@@ -2074,6 +2090,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_applies_sequential_hint_when_pattern_unknown() {
         let context = IoSchedulingContext {
             file_size: 2 * 1024 * 1024 * 1024,              // 2GiB
@@ -2098,6 +2115,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_bandwidth_low_reduces_buffer() {
         // Low bandwidth should reduce buffer
         let context = IoSchedulingContext {
@@ -2121,6 +2139,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_high_concurrency_reduction() {
         // High concurrency should reduce buffer
         let context = IoSchedulingContext {
@@ -2143,6 +2162,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_sequential_boost() {
         // Sequential reads should get boost
         let sequential_context = IoSchedulingContext {
@@ -2184,6 +2204,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_unknown_media_conservative() {
         // Unknown media should be conservative
         let context = IoSchedulingContext {
@@ -2209,6 +2230,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_priority_classification() {
         // Test priority classification based on file size
         let small_context = IoSchedulingContext {
@@ -2255,6 +2277,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_readahead_decision_matrix() {
         // Test readahead enable/disable logic
         let configs = vec![
@@ -2340,6 +2363,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_buffer_multiplier_stages() {
         // Test that all multiplier stages are applied
         let context = IoSchedulingContext {
@@ -2374,6 +2398,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial]
     async fn test_multi_factor_strategy_compatibility_path() {
         // Test that compatibility path (from_wait_duration) still works
         let wait_duration = Duration::from_millis(50);

rustfs/src/storage/minio_generated_read_test.rs

@@ -4,7 +4,7 @@ use std::fs;
 use std::io::Cursor;
 use std::path::{Path, PathBuf};
 
-use super::sse::{SseObjectEncryptionResolver, reset_sse_dek_provider};
+use super::sse::SseObjectEncryptionResolver;
 use super::storage_api::ecstore_test_support::{
     DiskAPI as _, DiskOption, Endpoint, Erasure, GetObjectReader, ObjectInfo, ObjectOptions, create_bitrot_reader, new_disk,
 };
@@ -131,13 +131,6 @@ async fn load_fixture_reader_input(case_id: &str) -> (ObjectInfo, Vec<u8>, Strin
 async fn read_fixture_plaintext(encrypted: Vec<u8>, object_info: ObjectInfo, kms_key_b64: String) -> Result<Vec<u8>, String> {
     let object_size = object_info.size;
 
-    // The DEK provider is cached process-wide once built, so without this reset
-    // a case that ran earlier in the same binary keeps serving its master key to
-    // every later case — which silently turned the wrong-key negative below into
-    // a test that could not fail. Reset before each read so the provider is
-    // built from the key this case actually configured.
-    reset_sse_dek_provider();
-
     async_with_vars(
         [
             ("__RUSTFS_SSE_SIMPLE_CMK", Some(kms_key_b64)),

rustfs/src/storage/sse.rs

@@ -1460,16 +1460,6 @@ fn managed_sse_domain(sse_type: SSEType) -> &'static str {
     }
 }
 
-/// The public `x-amz-server-side-encryption` value a managed scheme reports.
-fn managed_sse_public_header(sse_type: SSEType) -> &'static str {
-    match sse_type {
-        SSEType::SseKms => ServerSideEncryption::AWS_KMS,
-        // SSE-C never reaches the managed path; reporting AES256 keeps this
-        // total without inventing a third public value.
-        SSEType::SseS3 | SSEType::SseC => ServerSideEncryption::AES256,
-    }
-}
-
 fn canonical_kms_bucket_path(bucket: &str, key: &str) -> String {
     path_join_buf(&[bucket, key])
 }
@@ -2455,42 +2445,20 @@ async fn apply_managed_decryption_material_inner(
 ) -> Result<Option<DecryptionMaterial>, ApiError> {
     #[cfg(not(feature = "rio-v2"))]
     let _ = (bucket, key);
-    if !contains_managed_encryption_metadata(metadata) {
+    if !contains_managed_encryption_metadata(metadata) || !metadata.contains_key("x-amz-server-side-encryption") {
         return Ok(None);
     }
 
-    let encryption_type = match metadata.get("x-amz-server-side-encryption").map(String::as_str) {
-        Some(ServerSideEncryption::AWS_KMS) => SSEType::SseKms,
-        Some(_) => SSEType::SseS3,
-        // MinIO never persists the public scheme header: `crypto.S3.CreateMetadata`
-        // writes only the `X-Minio-Internal-*` family and the public header is
-        // synthesized onto the response by `DecryptObjectInfo`. Requiring it here
-        // is what made every MinIO-encrypted object unreadable (backlog#1638).
-        //
-        // Inferring from the sealed-key slot is self-consistent by construction:
-        // the slot decides which header the unseal reads AND which domain string
-        // the sealing key is derived under, so a scheme that disagrees with the
-        // slot cannot silently derive a wrong key — it finds no key at all.
-        // Inferring from the KMS key id would NOT be safe: MinIO writes
-        // `-S3-Kms-Key-Id` on SSE-S3 objects too.
-        #[cfg(feature = "rio-v2")]
-        None => match infer_minio_managed_sse_type(metadata) {
-            Some(sse_type) => sse_type,
-            // Still fail-closed, and deliberately not an error raised here: the
-            // read plan independently classifies the object as encrypted from
-            // its markers and refuses to serve it without material, so an
-            // object whose scheme cannot be established never degrades into a
-            // plaintext read.
-            None => return Ok(None),
-        },
-        // Without the rio-v2 reader there is no MinIO-format read path to serve
-        // such an object with, so it stays on the fail-closed branch.
-        #[cfg(not(feature = "rio-v2"))]
-        None => return Ok(None),
-    };
-
+    // Safe: presence is guaranteed by the contains_key check above.
+    let server_side_encryption = metadata.get("x-amz-server-side-encryption").cloned().unwrap_or_default();
     let normalized_metadata = normalize_managed_metadata(metadata, Some(recode_minio_kms_context));
 
+    let encryption_type = match server_side_encryption.as_str() {
+        ServerSideEncryption::AES256 => SSEType::SseS3,
+        ServerSideEncryption::AWS_KMS => SSEType::SseKms,
+        _ => SSEType::SseS3,
+    };
+
     // Extract KMS key ID from metadata (optional, used for provider context)
     let kms_key_id = normalized_metadata
         .get(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
@@ -2588,19 +2556,8 @@ async fn apply_managed_decryption_material_inner(
     } else {
         get_local_sse_dek_provider().await?
     };
-    // A MinIO sealed key alone does not mean MinIO wrote the object: RustFS's own
-    // writer fills MinIO's metadata slots too, while still storing a RustFS
-    // envelope in them, so neither the slot nor the header name distinguishes the
-    // two. The data key's own shape does. RustFS envelopes are strictly-parsed
-    // JSON; MinIO's builtin-KMS ciphertext is opaque bytes that match neither, so
-    // recognizing RustFS positively — and treating only the remainder as MinIO —
-    // keeps a RustFS envelope from ever reaching MinIO's decoder.
     #[cfg(feature = "rio-v2")]
-    let decrypted_data_key = if minio_sealed_key.is_some() && !is_rustfs_managed_data_key(&encrypted_data_key) {
-        provider
-            .decrypt_minio_sse_dek(&encrypted_data_key, &kms_key_id, &object_context)
-            .await
-    } else if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
+    let decrypted_data_key = if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
         provider
             .decrypt_legacy_sse_dek(&encrypted_data_key, &kms_key_id, &object_context)
             .await
@@ -2635,11 +2592,7 @@ async fn apply_managed_decryption_material_inner(
 
     Ok(Some(DecryptionMaterial {
         sse_type: encryption_type,
-        // Synthesized from the resolved scheme rather than read back from
-        // metadata: a MinIO-written object has no stored scheme header, which is
-        // exactly why the gate above had to infer it. MinIO synthesizes the same
-        // header onto its own responses.
-        server_side_encryption: ServerSideEncryption::from(managed_sse_public_header(encryption_type).to_string()),
+        server_side_encryption: ServerSideEncryption::from(server_side_encryption),
         kms_key_id: Some(SSEKMSKeyId::from(kms_key_id)),
         algorithm,
         customer_key_md5: None,
@@ -2706,30 +2659,6 @@ pub trait SseDekProvider: Send + Sync {
     ) -> Result<[u8; 32], ApiError> {
         self.decrypt_sse_dek(encrypted_dek, kms_key_id, context).await
     }
-
-    /// Unwrap a data key that MinIO's builtin KMS sealed.
-    ///
-    /// A separate entry point rather than a shape sniff inside
-    /// [`Self::decrypt_sse_dek`]: the caller already knows the object carries a
-    /// MinIO sealed key, and MinIO's raw ciphertext is unstructured bytes that
-    /// no parser can reliably tell apart from anything else. Routing on the
-    /// caller's knowledge keeps a RustFS envelope from ever reaching MinIO's
-    /// decoder, and vice versa.
-    ///
-    /// Defaults to refusing: only a provider holding the MinIO master secret
-    /// can serve these, and a provider that cannot must fail rather than fall
-    /// back to a decoder that would misread the bytes.
-    #[cfg(feature = "rio-v2")]
-    async fn decrypt_minio_sse_dek(
-        &self,
-        _encrypted_dek: &[u8],
-        _kms_key_id: &str,
-        _context: &ObjectEncryptionContext,
-    ) -> Result<[u8; 32], ApiError> {
-        Err(ApiError::from(StorageError::other(
-            "This KMS provider cannot unwrap a data key sealed by MinIO's builtin KMS",
-        )))
-    }
 }
 
 // ============================================================================
@@ -2868,163 +2797,6 @@ pub(crate) struct LocalSseDekProvider {
 
 const LOCAL_SSE_DEK_FORMAT_VERSION: u8 = 1;
 
-#[cfg(feature = "rio-v2")]
-/// Returns true when a managed-SSE data key is one RustFS itself wrote.
-///
-/// Both RustFS envelope shapes are strict JSON — the KMS envelope
-/// ([`rustfs_kms::is_data_key_envelope`]) and the local provider's
-/// [`LocalSseDekEnvelope`], whose `deny_unknown_fields` keeps it from accepting
-/// anything else. Recognition is deliberately positive: an unrecognized payload
-/// is left to MinIO's decoder rather than guessed at, and neither decoder is
-/// ever handed the other's format.
-fn is_rustfs_managed_data_key(encrypted_dek: &[u8]) -> bool {
-    if rustfs_kms::is_data_key_envelope(encrypted_dek) {
-        return true;
-    }
-    std::str::from_utf8(encrypted_dek)
-        .ok()
-        .is_some_and(|text| serde_json::from_str::<LocalSseDekEnvelope<'_>>(text).is_ok())
-}
-
-#[cfg(feature = "rio-v2")]
-/// Associated data MinIO binds when sealing a data key.
-///
-/// MinIO passes the object's encryption context as the AEAD's associated data,
-/// serialized as canonical JSON with sorted keys — the same canonicalization
-/// [`rustfs_kms::context_aad`] performs, which is why the context RustFS
-/// already rebuilds for the read can be reused verbatim. For SSE-S3 that
-/// context is `{bucket: "bucket/object"}`; for SSE-KMS it is whatever the
-/// request supplied, recovered from the stored MinIO context header.
-fn minio_kms_associated_data(context: &ObjectEncryptionContext) -> Result<Vec<u8>, ApiError> {
-    let mut ctx = context.encryption_context.clone();
-    ctx.entry(context.bucket.clone())
-        .or_insert_with(|| canonical_kms_bucket_path(&context.bucket, &context.object_key));
-    rustfs_kms::context_aad(&ctx)
-        .map_err(|e| ApiError::from(StorageError::other(format!("Failed to canonicalize MinIO KMS context: {e}"))))
-}
-
-#[cfg(feature = "rio-v2")]
-/// MinIO's builtin-KMS ciphertext in its JSON encoding.
-///
-/// Deliberately its own type rather than a relaxation of
-/// [`LocalSseDekEnvelope`]: widening that envelope's `deny_unknown_fields`
-/// to admit this shape would also admit malformed RustFS envelopes, which
-/// backlog#1567 requires to keep failing closed.
-#[derive(Debug, Deserialize)]
-#[serde(deny_unknown_fields)]
-struct MinioKmsCiphertextJson {
-    aead: String,
-    #[allow(
-        dead_code,
-        reason = "present in MinIO's encoding; the key is identified by metadata instead"
-    )]
-    #[serde(default)]
-    id: String,
-    iv: String,
-    nonce: String,
-    bytes: String,
-}
-
-/// Bytes of trailing randomness every MinIO builtin-KMS ciphertext carries:
-/// a 16-byte IV followed by a 12-byte nonce, *after* the sealed bytes.
-#[cfg(feature = "rio-v2")]
-const MINIO_KMS_RANDOM_LEN: usize = 28;
-#[cfg(feature = "rio-v2")]
-const MINIO_KMS_IV_LEN: usize = 16;
-
-#[cfg(feature = "rio-v2")]
-const MINIO_KMS_AEAD_AES_GCM: &str = "AES-256-GCM-HMAC-SHA-256";
-#[cfg(feature = "rio-v2")]
-const MINIO_KMS_AEAD_CHACHA20: &str = "ChaCha20Poly1305";
-
-#[cfg(feature = "rio-v2")]
-/// Unwrap a data key sealed by MinIO's builtin (static-secret) KMS.
-///
-/// The wire format is `sealed_bytes || iv[16] || nonce[12]` — the randomness
-/// trails the ciphertext rather than leading it, and MinIO's own decoder
-/// normalizes its legacy JSON encoding into exactly that byte order before
-/// opening it (`internal/kms/secret-key.go`, `parseCiphertext`). A raw
-/// (non-JSON) ciphertext is AES-256-GCM by definition there; the JSON form
-/// names its algorithm.
-///
-/// The sealing key is derived per ciphertext rather than being the master key:
-/// `HMAC-SHA256(master, iv)` for AES-256-GCM, `HChaCha20(master, iv)` for
-/// ChaCha20-Poly1305. The encryption context is bound as associated data.
-fn decrypt_minio_kms_data_key(encrypted_dek: &[u8], master_key: &[u8; 32], aad: &[u8]) -> Result<[u8; 32], ApiError> {
-    let (body, algorithm) = match std::str::from_utf8(encrypted_dek) {
-        // MinIO only treats a payload as JSON when it both starts and ends like
-        // an object, and falls back to the raw layout when it does not parse —
-        // mirrored here so a ciphertext that merely looks like JSON is not
-        // rejected outright.
-        Ok(text)
-            if text.starts_with('{')
-                && text.ends_with('}')
-                && let Ok(json) = serde_json::from_str::<MinioKmsCiphertextJson>(text) =>
-        {
-            let decode = |what: &str, value: &str| -> Result<Vec<u8>, ApiError> {
-                BASE64_STANDARD
-                    .decode(value)
-                    .map_err(|e| ApiError::from(StorageError::other(format!("Invalid MinIO KMS {what}: {e}"))))
-            };
-            let mut body = decode("ciphertext", &json.bytes)?;
-            body.extend_from_slice(&decode("iv", &json.iv)?);
-            body.extend_from_slice(&decode("nonce", &json.nonce)?);
-            (body, json.aead)
-        }
-        _ => (encrypted_dek.to_vec(), MINIO_KMS_AEAD_AES_GCM.to_string()),
-    };
-
-    if body.len() <= MINIO_KMS_RANDOM_LEN {
-        return Err(ApiError::from(StorageError::other(
-            "MinIO KMS ciphertext is too short to carry its IV and nonce",
-        )));
-    }
-    let (sealed, random) = body.split_at(body.len() - MINIO_KMS_RANDOM_LEN);
-    let (iv, nonce) = random.split_at(MINIO_KMS_IV_LEN);
-
-    let plaintext = match algorithm.as_str() {
-        MINIO_KMS_AEAD_AES_GCM => {
-            use aes_gcm::{Aes256Gcm, KeyInit, aead::Aead};
-            let mut mac = HmacSha256::new_from_slice(master_key)
-                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS sealing key derivation failed")))?;
-            mac.update(iv);
-            let sealing_key: [u8; 32] = mac.finalize().into_bytes().into();
-            let cipher = Aes256Gcm::new_from_slice(&sealing_key)
-                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS sealing key is not a valid AES-256 key")))?;
-            let nonce = aes_gcm::Nonce::try_from(nonce)
-                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS nonce is not 12 bytes")))?;
-            cipher.decrypt(&nonce, aes_gcm::aead::Payload { msg: sealed, aad })
-        }
-        MINIO_KMS_AEAD_CHACHA20 => {
-            use chacha20poly1305::{KeyInit, XChaCha20Poly1305, aead::Aead};
-            // MinIO derives this branch's key with HChaCha20 over the 16-byte
-            // IV, which is exactly XChaCha20-Poly1305's own construction, so the
-            // extended-nonce cipher does the derivation rather than hand-rolling it.
-            let mut extended = Vec::with_capacity(MINIO_KMS_IV_LEN + nonce.len());
-            extended.extend_from_slice(iv);
-            extended.extend_from_slice(nonce);
-            let cipher = XChaCha20Poly1305::new_from_slice(master_key)
-                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS master key is not a valid ChaCha20 key")))?;
-            let nonce = chacha20poly1305::XNonce::try_from(extended.as_slice())
-                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS extended nonce is not 24 bytes")))?;
-            cipher.decrypt(&nonce, chacha20poly1305::aead::Payload { msg: sealed, aad })
-        }
-        other => {
-            return Err(ApiError::from(StorageError::other(format!(
-                "Unsupported MinIO KMS AEAD algorithm: {other}"
-            ))));
-        }
-    }
-    // An AEAD failure here is authentication, not a decode slip: a wrong master
-    // key, a tampered ciphertext, and an encryption context that does not match
-    // what sealed it all land here and must all fail closed.
-    .map_err(|_| ApiError::from(StorageError::other("MinIO KMS data key failed authentication")))?;
-
-    plaintext.try_into().map_err(|value: Vec<u8>| {
-        ApiError::from(StorageError::other(format!("MinIO KMS data key must be 32 bytes, got {}", value.len())))
-    })
-}
-
 #[derive(Debug, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 struct LocalSseDekEnvelope<'a> {
@@ -3241,17 +3013,6 @@ impl SseDekProvider for LocalSseDekProvider {
         let dek = Self::decrypt_dek(encrypted_dek_str, self.master_key)?;
         Ok(dek)
     }
-
-    #[cfg(feature = "rio-v2")]
-    async fn decrypt_minio_sse_dek(
-        &self,
-        encrypted_dek: &[u8],
-        _kms_key_id: &str,
-        context: &ObjectEncryptionContext,
-    ) -> Result<[u8; 32], ApiError> {
-        let aad = minio_kms_associated_data(context)?;
-        decrypt_minio_kms_data_key(encrypted_dek, &self.master_key, &aad)
-    }
 }
 
 // ============================================================================
@@ -3440,23 +3201,6 @@ fn is_legacy_rustfs_managed_metadata(metadata: &HashMap<String, String>) -> bool
         && !metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER)
 }
 
-#[cfg(feature = "rio-v2")]
-#[cfg(feature = "rio-v2")]
-/// Infer the managed SSE scheme from the MinIO sealed-key slot that is present.
-///
-/// Returns `None` when no managed MinIO slot is present, which keeps callers on
-/// their fail-closed path. SSE-C is not a managed scheme and is handled by the
-/// SSE-C read path, so its slot is not considered here.
-fn infer_minio_managed_sse_type(metadata: &HashMap<String, String>) -> Option<SSEType> {
-    if metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER) {
-        Some(SSEType::SseS3)
-    } else if metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER) {
-        Some(SSEType::SseKms)
-    } else {
-        None
-    }
-}
-
 #[cfg(feature = "rio-v2")]
 fn parse_minio_managed_sealed_key(
     metadata: &HashMap<String, String>,