docs(sse): record SSE-C as measured-readable in the migration matrix

The SSE-C row moves from unverified to supported now that a MinIO-generated SSE-C object reads end to end, with the test that establishes it. Single-part stays unverified for all three schemes — the fixture harness still cannot load objects MinIO inlined into xl.meta.

Also notes the one behavioural difference an operator will observe: a MinIO SSE-C object stores no customer-key MD5, so a wrong key is refused by the decryption rather than by the earlier parameter-mismatch check. Different error, same outcome.

Refs rustfs/backlog#1638.

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/ecstore/src/disk/local.rs

@@ -9745,8 +9745,7 @@ impl DiskAPI for LocalDisk {
                 {
                     let fsync_started = rustfs_io_metrics::put_stage_timer();
                     if let Err(err) =
-                        os::fsync_dst_dir_group_commit_or_namespace_file_sync_limit(dst_parent, mutation_lease.clone(), admission)
-                            .await
+                        os::fsync_dir_with_namespace_file_sync_limit(dst_parent, mutation_lease.clone(), admission).await
                     {
                         rustfs_io_metrics::record_put_object_stage_duration_from(
                             rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC,
@@ -13093,80 +13092,6 @@ mod test {
         );
     }
 
-    #[tokio::test]
-    #[serial_test::serial(dst_dir_fsync_group_commit)]
-    async fn rename_data_inline_uses_dst_dir_fsync_group_commit_when_enabled() {
-        use tempfile::tempdir;
-
-        let _group_commit = os::set_dst_dir_fsync_group_commit_for_test(true);
-        let _mode = durability_mode_override::set(DurabilityMode::Strict);
-        let dir = tempdir().expect("temp dir should be created");
-        let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
-        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
-        let bucket = "grouped-inline-dst-fsync-bucket";
-        let object = "dir/inline-object";
-        let tmp_object = "tmp-grouped-inline-dst-fsync";
-        ensure_test_volume(&disk, bucket).await;
-        ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
-
-        let version_id = Uuid::parse_str("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb").expect("version id should parse");
-        let new_fi = test_file_info(object, version_id, None, Some(Bytes::from_static(b"inline-payload")));
-        disk.rename_data(RUSTFS_META_TMP_BUCKET, tmp_object, new_fi, bucket, object)
-            .await
-            .expect("inline rename_data should commit");
-
-        let dst_meta_parent = disk
-            .get_object_path(bucket, &format!("{object}/{STORAGE_FORMAT_FILE}"))
-            .expect("dst meta path should resolve")
-            .parent()
-            .expect("dst meta should have a parent")
-            .to_path_buf();
-        assert_eq!(
-            os::fsync_dir_recorder::grouped_batch_sizes(&dst_meta_parent),
-            vec![1],
-            "enabled inline rename_data must route the dst parent fsync through the group commit coordinator"
-        );
-        assert!(
-            !os::fsync_dir_recorder::was_limited(&dst_meta_parent),
-            "enabled grouped dst parent fsync must not also run the direct file-sync limited path"
-        );
-    }
-
-    #[tokio::test]
-    #[serial_test::serial(dst_dir_fsync_group_commit)]
-    async fn rename_data_inline_dst_dir_fsync_group_commit_failure_rolls_back_fresh_put() {
-        use tempfile::tempdir;
-
-        let _group_commit = os::set_dst_dir_fsync_group_commit_for_test(true);
-        let _mode = durability_mode_override::set(DurabilityMode::Strict);
-        let dir = tempdir().expect("temp dir should be created");
-        let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
-        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
-        let bucket = "grouped-inline-dst-fsync-failure-bucket";
-        let object = "dir/inline-object";
-        let tmp_object = "tmp-grouped-inline-dst-fsync-failure";
-        ensure_test_volume(&disk, bucket).await;
-        ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
-
-        let dst_meta_parent = dir.path().join(bucket).join(object);
-        os::fsync_dir_recorder::set_grouped_failure(&dst_meta_parent, io::ErrorKind::PermissionDenied);
-        let version_id = Uuid::parse_str("cccccccc-cccc-cccc-cccc-cccccccccccc").expect("version id should parse");
-        let new_fi = test_file_info(object, version_id, None, Some(Bytes::from_static(b"inline-payload")));
-        let err = disk
-            .rename_data(RUSTFS_META_TMP_BUCKET, tmp_object, new_fi, bucket, object)
-            .await
-            .expect_err("grouped dst dir fsync failure must fail the fresh inline PUT");
-
-        assert!(
-            matches!(err, DiskError::Io(ref io_err) if io_err.kind() == io::ErrorKind::PermissionDenied),
-            "grouped dst dir fsync failure must propagate the injected permission error"
-        );
-        assert!(
-            !dst_meta_parent.join(STORAGE_FORMAT_FILE).exists(),
-            "fresh inline PUT rollback must remove the committed xl.meta after grouped dst dir fsync failure"
-        );
-    }
-
     // Seed a first PUT of `object` (no prior version) through the non-inline
     // rename_data path and return (disk, tempdir). The object dir and any prefix
     // dirs are created during the commit.

crates/ecstore/src/disk/os.rs

@@ -676,18 +676,6 @@ pub(crate) async fn fsync_dst_dir_group_commit(dir: impl AsRef<Path>) -> io::Res
     fsync_dst_dir_group_commit_with_enabled(dir, dst_dir_fsync_group_commit_enabled()).await
 }
 
-pub(crate) async fn fsync_dst_dir_group_commit_or_namespace_file_sync_limit(
-    dir: impl AsRef<Path>,
-    lease: Arc<NamespaceMutationLease>,
-    admission: &FileSyncAdmission,
-) -> io::Result<()> {
-    if dst_dir_fsync_group_commit_enabled() {
-        fsync_dst_dir_group_commit_with_enabled(dir, true).await
-    } else {
-        fsync_dir_with_namespace_file_sync_limit(dir, lease, admission).await
-    }
-}
-
 #[cfg(test)]
 pub(crate) async fn fsync_dst_dir_group_commit_for_test(dir: impl AsRef<Path>, enabled: bool) -> io::Result<()> {
     fsync_dst_dir_group_commit_with_enabled(dir, enabled).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/config.rs

@@ -324,11 +324,18 @@ impl Default for LocalConfig {
 /// wraps data encryption keys — there is no HMAC-SHA256 derivation step — and
 /// each wrapped DEK is serialized as a RustFS `DataKeyEnvelope` JSON blob.
 ///
-/// This mirrors the *concept* of MinIO's builtin/static single-key KMS, but is
-/// not wire-compatible with it: MinIO wraps DEKs in a different (`{"aead": ...}`)
-/// blob that this backend neither produces nor accepts, so KMS ciphertext
-/// written by MinIO cannot be opened here. Reading MinIO-written SSE objects is
-/// tracked separately in rustfs/backlog#1638.
+/// This mirrors the *concept* of MinIO's builtin/static single-key KMS but is
+/// not wire-compatible with it. MinIO seals a DEK as `sealed || iv[16] ||
+/// nonce[12]` under a per-ciphertext key derived from the master secret, with
+/// a legacy JSON encoding of the same layout; this backend neither produces
+/// nor accepts either, so pointing it at MinIO's master key does **not** make
+/// MinIO-written objects readable through it.
+///
+/// Reading MinIO-written SSE objects is a property of the object read path, not
+/// of this backend: that path decodes MinIO's format directly, keyed by
+/// `RUSTFS_SSE_S3_MASTER_KEY`. See the migration section of
+/// `docs/operations/kms-backend-security.md` for which object shapes are
+/// covered, and rustfs/backlog#1638 for the remainder.
 #[derive(Clone, Default, Serialize, Deserialize)]
 pub struct StaticConfig {
     /// Key identifier (name) for the single configured key

docs/operations/kms-backend-security.md

@@ -14,24 +14,41 @@ For how the Vault backends authenticate (static token, AppRole, Kubernetes, Vaul
 | Vault Transit | `VaultTransit` | Key-encryption keys never leave Vault; only Transit ciphertext is visible outside | Vault Transit engine (cryptographic isolation) | Delegated to Vault storage | Via Vault Transit key versioning | Deployments that need key material to be unreadable through storage APIs |
 | AWS KMS | `AWS` (alias `AwsKms`) | Key material never leaves AWS KMS; RustFS mirrors no key state | AWS KMS (cryptographic isolation) + IAM | Delegated to AWS | On-demand `RotateKeyOnDemand`; prior backing keys stay usable for decryption | Deployments already rooted in AWS IAM that want AWS as the cryptographic root — read [AWS KMS: deviations from the shared backend contract](#aws-kms-deviations-from-the-shared-backend-contract) first |
 
-## Migrating from MinIO: encrypted objects do not carry over
+## Migrating from MinIO: what carries over, and what does not
 
-> **Warning: RustFS does not currently support reading objects that MinIO encrypted.**
-> This applies to SSE-S3, SSE-KMS, and SSE-C, in every released binary and container image, and it holds regardless of which KMS backend you configure. Configuring the `Static` backend with the same key material MinIO used does **not** make those objects readable — MinIO wraps data keys in a different envelope format that no RustFS backend produces or accepts (`crates/kms/src/config.rs:304-308`). Plan for this **before** moving data. Tracked in rustfs/backlog#1638.
+> **Read this before moving data.** Some MinIO-encrypted objects are readable by RustFS and some are not, and the boundary is not where you would guess. Verify against a sample of your own objects rather than assuming either answer. Tracked in rustfs/backlog#1638.
 
-The read does fail closed — ciphertext is never served as plaintext. MinIO's internal encryption headers mark the object as encrypted (`crates/utils/src/http/header_compat.rs:50-67`), so the read path demands encryption material and refuses when none resolves (`crates/ecstore/src/object_api/readers.rs:559-568`). Two properties still make the problem easy to discover late:
+Support is stated per shape below because that is how far it has been *measured* — against fixtures captured from a real MinIO server (`minio/minio:RELEASE.2025-09-07T16-13-09Z`), not inferred from the code:
+
+| MinIO object | RustFS read | Evidence |
+| --- | --- | --- |
+| SSE-S3, multipart | **Yes** | `reads_minio_generated_sse_s3_multipart_fixture` |
+| SSE-KMS, multipart | **Yes** | `reads_minio_generated_sse_kms_multipart_fixture` |
+| SSE-C, multipart | **Yes** | `reads_minio_generated_sse_c_multipart_fixture` |
+| SSE-S3 / SSE-KMS / SSE-C, single-part | **Unverified** | No fixture coverage — see below |
+| Sealed by KES, a KMS plugin, or MinKMS | **No**, and not planned | Re-encrypt at the source before migrating |
+
+SSE-C needs no KMS at all: the customer supplies the key on each request, exactly as against MinIO. Note that a MinIO SSE-C object stores no customer-key MD5, so the usual early "these parameters do not match" rejection cannot fire for it — a wrong key is refused by the decryption itself instead, which is a different error but the same outcome.
+
+Reading a supported *managed* object (SSE-S3, SSE-KMS) requires RustFS to hold the same master key MinIO used, supplied through `RUSTFS_SSE_S3_MASTER_KEY` (the production entry point, exercised by `reads_minio_generated_sse_s3_fixture_through_production_master_key_env`). MinIO's builtin KMS derives a per-ciphertext sealing key from that master secret, so the *same* secret is required — not merely an equivalently configured backend.
+
+**"Unverified" means unknown, not broken.** Single-part objects below MinIO's small-file threshold carry their data inline in `xl.meta`, sharded across disks, and the interop fixture harness cannot yet load that shape — so those objects have never been read in a test either way. Do not read the table's "Yes" rows as covering them.
+
+Whatever the table says, verify before you commit: **read a sample of encrypted objects, not just their listings.** A read that is not supported fails closed — ciphertext is never served as plaintext — but two properties still make it easy to discover late:
 
 - **The error does not say what happened.** It surfaces as a 500 `InternalError`, which reads as a RustFS fault rather than "another implementation encrypted this object".
 - **Surrounding metadata migrates fine.** The object's `xl.meta` parses, so encrypted objects list and HEAD normally and report plausible sizes. The failure appears only when something reads the payload.
 
-Read a sample of encrypted objects, not just their listings, before decommissioning the MinIO deployment.
-
-Current options for a migration whose source contains encrypted objects:
+For any shape that does not read, the options are unchanged:
 
 - Decrypt on the MinIO side first, migrate plaintext, then let RustFS re-encrypt with its own KMS.
 - Copy through the S3 API rather than moving drives — MinIO decrypts on read, and RustFS encrypts on write. This re-encrypts rather than preserving ciphertext and costs a full data transfer.
 - Leave encrypted objects on MinIO and migrate only unencrypted data.
 
+### The reverse direction does not work
+
+MinIO cannot read objects RustFS encrypted, and that is a deliberate, documented position rather than a gap awaiting a fix. RustFS fills MinIO's metadata slots — the sealed-key and IV headers are MinIO-shaped — but the data key in `X-Minio-Internal-Server-Side-Encryption-S3-Kms-Sealed-Key` is a RustFS envelope, which MinIO's KMS cannot open. **Treat the MinIO-branded headers on a RustFS-written object as RustFS-internal.** Their presence is not a statement that MinIO can read the object, and no coexistence plan should assume two-way reads.
+
 Inventory the source before choosing: bucket default-encryption settings mean objects can be encrypted without any client having sent SSE headers.
 
 The same limitation applies in reverse — objects RustFS encrypts are not readable by MinIO. For the code-level breakdown of which seams block each SSE mode, see [MinIO file-format interoperability, Part C](../architecture/minio-file-format-compat.md#part-c--server-side-encryption-sse).

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/admin/handlers/cluster_snapshot.rs

@@ -14,7 +14,7 @@
 
 use crate::admin::storage_api::cluster::{CapabilityState, CapabilityStatus, ObservabilitySnapshot, TopologySnapshot};
 use crate::admin::{
-    auth::authorize_admin_request,
+    auth::validate_admin_request,
     router::{AdminOperation, Operation, S3Router},
     runtime_sources::default_admin_usecase,
     storage_api::cluster::{
@@ -24,10 +24,11 @@ use crate::admin::{
     },
     system,
 };
+use crate::auth::{check_key_valid, get_session_token};
 use crate::cluster_snapshot::{
     ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot, cluster_has_actionable_pressure,
 };
-use crate::server::{ADMIN_PREFIX, ReadinessDegradedReason};
+use crate::server::{ADMIN_PREFIX, ReadinessDegradedReason, RemoteAddr};
 use http::{HeaderMap, HeaderValue, StatusCode};
 use hyper::Method;
 use matchit::Params;
@@ -65,15 +66,23 @@ pub(crate) struct ClusterSnapshotDiscoveryResponse {
     pub components: Option<ClusterComponentStatusView>,
 }
 
-/// The pre-check keeps this endpoint's historical missing-credentials message;
-/// the shared gate reports "get cred failed".
 async fn authorize_cluster_snapshot_request(req: &S3Request<Body>) -> S3Result<()> {
-    if req.credentials.is_none() {
+    let Some(input_cred) = &req.credentials else {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    }
+    };
 
-    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await?;
-    Ok(())
+    let (cred, owner) =
+        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+    validate_admin_request(
+        &req.headers,
+        &cred,
+        owner,
+        false,
+        vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)],
+        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+    )
+    .await
 }
 
 fn build_json_response(
@@ -944,30 +953,6 @@ mod tests {
         );
     }
 
-    /// This endpoint authorizes through the shared admin gate, which reports
-    /// "get cred failed" for a credential-less request. The pre-check keeps the
-    /// message it has always returned (rustfs/backlog#1829).
-    #[tokio::test]
-    async fn cluster_snapshot_gate_keeps_its_missing_credentials_message() {
-        let req = s3s::S3Request {
-            input: s3s::Body::from(String::new()),
-            method: http::Method::GET,
-            uri: http::Uri::from_static("/rustfs/admin/v4/cluster/snapshot"),
-            headers: http::HeaderMap::new(),
-            extensions: http::Extensions::new(),
-            credentials: None,
-            region: None,
-            service: None,
-            trailing_headers: None,
-        };
-
-        let err = super::authorize_cluster_snapshot_request(&req)
-            .await
-            .expect_err("a request without credentials must be rejected");
-        assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
-        assert_eq!(err.message(), Some("authentication required"));
-    }
-
     #[test]
     fn cluster_snapshot_response_serializes_none_snapshot() {
         let value = serde_json::to_value(ClusterSnapshotResponse { snapshot: None }).expect("serialize response");

rustfs/src/admin/handlers/extensions.rs

@@ -14,7 +14,7 @@
 
 use crate::admin::storage_api::cluster::CapabilityStatus;
 use crate::admin::{
-    auth::authorize_admin_request,
+    auth::validate_admin_request,
     handlers::{cluster_snapshot, plugins_instances, system},
     plugin_contract::{
         PluginContractDomain, PluginInstanceDiagnosticCode, PluginInstanceDiagnosticCount, PluginInstanceEntry,
@@ -22,7 +22,8 @@ use crate::admin::{
     },
     router::{AdminOperation, Operation, S3Router},
 };
-use crate::server::ADMIN_PREFIX;
+use crate::auth::{check_key_valid, get_session_token};
+use crate::server::{ADMIN_PREFIX, RemoteAddr};
 use http::{HeaderMap, HeaderValue, StatusCode};
 use hyper::Method;
 use matchit::Params;
@@ -182,26 +183,42 @@ fn map_extension_instance(instance: PluginInstanceEntry) -> ExtensionInstanceEnt
     }
 }
 
-/// The pre-check keeps this endpoint's historical missing-credentials message;
-/// the shared gate reports "get cred failed".
 async fn authorize_extension_catalog_request(req: &S3Request<Body>) -> S3Result<()> {
-    if req.credentials.is_none() {
+    let Some(input_cred) = &req.credentials else {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    }
+    };
 
-    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await?;
-    Ok(())
+    let (cred, owner) =
+        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+    validate_admin_request(
+        &req.headers,
+        &cred,
+        owner,
+        false,
+        vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)],
+        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+    )
+    .await
 }
 
-/// The pre-check keeps this endpoint's historical missing-credentials message;
-/// the shared gate reports "get cred failed".
 async fn authorize_extension_instance_request(req: &S3Request<Body>) -> S3Result<()> {
-    if req.credentials.is_none() {
+    let Some(input_cred) = &req.credentials else {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    }
+    };
 
-    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::GetBucketTargetAction)]).await?;
-    Ok(())
+    let (cred, owner) =
+        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+    validate_admin_request(
+        &req.headers,
+        &cred,
+        owner,
+        false,
+        vec![Action::AdminAction(AdminAction::GetBucketTargetAction)],
+        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+    )
+    .await
 }
 
 fn build_json_response(
@@ -303,36 +320,6 @@ mod tests {
         );
     }
 
-    /// Both extension gates authorize through the shared admin gate, which reports
-    /// "get cred failed" for a credential-less request. The pre-check keeps the
-    /// message these endpoints have always returned (rustfs/backlog#1829).
-    #[tokio::test]
-    async fn extension_gates_keep_their_missing_credentials_message() {
-        let credential_less_request = || s3s::S3Request {
-            input: s3s::Body::from(String::new()),
-            method: http::Method::GET,
-            uri: http::Uri::from_static("/rustfs/admin/v4/extensions/catalog"),
-            headers: http::HeaderMap::new(),
-            extensions: http::Extensions::new(),
-            credentials: None,
-            region: None,
-            service: None,
-            trailing_headers: None,
-        };
-
-        for err in [
-            super::authorize_extension_catalog_request(&credential_less_request())
-                .await
-                .expect_err("a request without credentials must be rejected"),
-            super::authorize_extension_instance_request(&credential_less_request())
-                .await
-                .expect_err("a request without credentials must be rejected"),
-        ] {
-            assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
-            assert_eq!(err.message(), Some("authentication required"));
-        }
-    }
-
     #[test]
     fn builtin_ops_schemas_register_cleanly_in_runtime_registries() {
         let mut diagnostics_registry = rustfs_targets::OpsDiagnosticsRegistry::new();

rustfs/src/admin/handlers/object_data_cache.rs

@@ -21,11 +21,12 @@
 //! that bucket, and with `bucket`+`object` it flushes that one identity — the
 //! only remediation for a poisoned entry short of a node restart.
 
-use crate::admin::auth::authorize_admin_request;
+use crate::admin::auth::validate_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::current_object_data_cache;
 use crate::app::object_data_cache::ObjectDataCacheAdapter;
-use crate::server::ADMIN_PREFIX;
+use crate::auth::{check_key_valid, get_session_token};
+use crate::server::{ADMIN_PREFIX, RemoteAddr};
 use http::{HeaderMap, HeaderValue};
 use hyper::{Method, StatusCode};
 use matchit::Params;
@@ -75,14 +76,17 @@ pub fn register_object_data_cache_route(r: &mut S3Router<AdminOperation>) -> std
     Ok(())
 }
 
-/// The pre-check keeps these endpoints' historical missing-credentials message;
-/// the shared gate reports "get cred failed".
 async fn authorize(req: &S3Request<Body>, action: AdminAction) -> S3Result<()> {
-    if req.credentials.is_none() {
+    let Some(input_cred) = req.credentials.as_ref() else {
         return Err(s3_error!(InvalidRequest, "missing credentials"));
-    }
-    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
-    Ok(())
+    };
+    let (cred, owner) =
+        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+    let remote_addr = req
+        .extensions
+        .get::<Option<RemoteAddr>>()
+        .and_then(|opt| opt.map(|addr| addr.0));
+    validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(action)], remote_addr).await
 }
 
 fn json_response<T: Serialize>(body: &T) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -204,30 +208,6 @@ mod tests {
         assert_eq!(invalidation_outcome(&ObjectDataCacheInvalidationResult::NoOp), ("noop", 0));
     }
 
-    /// These endpoints authorize through the shared admin gate, which reports
-    /// "get cred failed" for a credential-less request. The pre-check keeps the
-    /// message they have always returned (rustfs/backlog#1829).
-    #[tokio::test]
-    async fn authorize_keeps_its_missing_credentials_message() {
-        let req = S3Request {
-            input: Body::from(String::new()),
-            method: Method::GET,
-            uri: "/rustfs/admin/v3/object-data-cache/stats".parse().expect("uri should parse"),
-            headers: HeaderMap::new(),
-            extensions: http::Extensions::new(),
-            credentials: None,
-            region: None,
-            service: None,
-            trailing_headers: None,
-        };
-
-        let err = authorize(&req, AdminAction::ServerInfoAdminAction)
-            .await
-            .expect_err("a request without credentials must be rejected");
-        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
-        assert_eq!(err.message(), Some("missing credentials"));
-    }
-
     #[test]
     fn stats_handler_requires_server_info_action() {
         // Guard the auth contract: the stats endpoint is a read, the flush

rustfs/src/admin/handlers/plugins_catalog.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use crate::admin::{
-    auth::authorize_admin_request,
+    auth::validate_admin_request,
     plugin_contract::{
         PluginCatalogAdminDiscovery, PluginCatalogDomainEntry, PluginCatalogEntry, PluginCatalogResponse, PluginContractDomain,
         PluginContractEntrypointKind, PluginContractPackaging, PluginDistributionContract, PluginRuntimeContract,
@@ -21,7 +21,8 @@ use crate::admin::{
     router::{AdminOperation, Operation, S3Router},
     runtime_sources::default_admin_usecase,
 };
-use crate::server::ADMIN_PREFIX;
+use crate::auth::{check_key_valid, get_session_token};
+use crate::server::{ADMIN_PREFIX, RemoteAddr};
 use http::{HeaderMap, HeaderValue, StatusCode};
 use hyper::Method;
 use matchit::Params;
@@ -113,15 +114,23 @@ fn merge_catalog_descriptor(plugins: &mut HashMap<&'static str, PluginCatalogEnt
     }
 }
 
-/// The pre-check keeps this endpoint's historical missing-credentials message;
-/// the shared gate reports "get cred failed".
 async fn authorize_plugin_catalog_request(req: &S3Request<Body>) -> S3Result<()> {
-    if req.credentials.is_none() {
+    let Some(input_cred) = &req.credentials else {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    }
+    };
 
-    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await?;
-    Ok(())
+    let (cred, owner) =
+        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+    validate_admin_request(
+        &req.headers,
+        &cred,
+        owner,
+        false,
+        vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)],
+        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+    )
+    .await
 }
 
 fn build_json_response(
@@ -166,30 +175,6 @@ mod tests {
         );
     }
 
-    /// This endpoint authorizes through the shared admin gate, which reports
-    /// "get cred failed" for a credential-less request. The pre-check keeps the
-    /// message it has always returned (rustfs/backlog#1829).
-    #[tokio::test]
-    async fn plugin_catalog_gate_keeps_its_missing_credentials_message() {
-        let req = s3s::S3Request {
-            input: s3s::Body::from(String::new()),
-            method: http::Method::GET,
-            uri: http::Uri::from_static("/rustfs/admin/v4/plugins/catalog"),
-            headers: http::HeaderMap::new(),
-            extensions: http::Extensions::new(),
-            credentials: None,
-            region: None,
-            service: None,
-            trailing_headers: None,
-        };
-
-        let err = super::authorize_plugin_catalog_request(&req)
-            .await
-            .expect_err("a request without credentials must be rejected");
-        assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
-        assert_eq!(err.message(), Some("authentication required"));
-    }
-
     #[test]
     fn plugin_catalog_contains_representative_builtin_targets() {
         let response = build_catalog_response();

rustfs/src/admin/handlers/plugins_instances.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use crate::admin::{
-    auth::authorize_admin_request,
+    auth::validate_admin_request,
     handlers::audit_runtime_config::{load_server_config_from_store, remove_audit_target_config, set_audit_target_config},
     handlers::notify_runtime_access::{
         load_notification_config_snapshot, remove_notification_target_config, set_notification_target_config,
@@ -29,9 +29,10 @@ use crate::admin::{
     },
     router::{AdminOperation, Operation, S3Router},
 };
+use crate::auth::{check_key_valid, get_session_token};
 use crate::server::{
-    ADMIN_PREFIX, is_audit_module_enabled, is_notify_module_enabled, refresh_audit_module_enabled, refresh_notify_module_enabled,
-    refresh_persisted_module_switches_from_store,
+    ADMIN_PREFIX, RemoteAddr, is_audit_module_enabled, is_notify_module_enabled, refresh_audit_module_enabled,
+    refresh_notify_module_enabled, refresh_persisted_module_switches_from_store,
 };
 use hyper::{Method, StatusCode};
 use matchit::Params;
@@ -562,26 +563,42 @@ fn plugin_instance_matches_query(instance: &PluginInstanceEntry, query: &str) ->
     .any(|field| field.to_ascii_lowercase().contains(&query))
 }
 
-/// The pre-check keeps this endpoint's historical missing-credentials message;
-/// the shared gate reports "get cred failed".
 async fn authorize_plugin_instance_request(req: &S3Request<Body>) -> S3Result<()> {
-    if req.credentials.is_none() {
+    let Some(input_cred) = &req.credentials else {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    }
+    };
 
-    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::GetBucketTargetAction)]).await?;
-    Ok(())
+    let (cred, owner) =
+        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+    validate_admin_request(
+        &req.headers,
+        &cred,
+        owner,
+        false,
+        vec![Action::AdminAction(AdminAction::GetBucketTargetAction)],
+        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+    )
+    .await
 }
 
-/// The pre-check keeps this endpoint's historical missing-credentials message;
-/// the shared gate reports "get cred failed".
 async fn authorize_plugin_instance_write_request(req: &S3Request<Body>) -> S3Result<()> {
-    if req.credentials.is_none() {
+    let Some(input_cred) = &req.credentials else {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    }
+    };
 
-    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::SetBucketTargetAction)]).await?;
-    Ok(())
+    let (cred, owner) =
+        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+    validate_admin_request(
+        &req.headers,
+        &cred,
+        owner,
+        false,
+        vec![Action::AdminAction(AdminAction::SetBucketTargetAction)],
+        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+    )
+    .await
 }
 
 fn plugin_instance_mutation_block_reason(
@@ -925,36 +942,6 @@ mod tests {
         );
     }
 
-    /// Both instance gates authorize through the shared admin gate, which reports
-    /// "get cred failed" for a credential-less request. The pre-check keeps the
-    /// message these endpoints have always returned (rustfs/backlog#1829).
-    #[tokio::test]
-    async fn plugin_instance_gates_keep_their_missing_credentials_message() {
-        let credential_less_request = || S3Request {
-            input: Body::from(String::new()),
-            method: Method::GET,
-            uri: Uri::from_static("/rustfs/admin/v4/plugins/instances"),
-            headers: HeaderMap::new(),
-            extensions: Extensions::new(),
-            credentials: None,
-            region: None,
-            service: None,
-            trailing_headers: None,
-        };
-
-        for err in [
-            super::authorize_plugin_instance_request(&credential_less_request())
-                .await
-                .expect_err("a request without credentials must be rejected"),
-            super::authorize_plugin_instance_write_request(&credential_less_request())
-                .await
-                .expect_err("a request without credentials must be rejected"),
-        ] {
-            assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
-            assert_eq!(err.message(), Some("authentication required"));
-        }
-    }
-
     #[test]
     fn configured_instance_without_runtime_appears_offline() {
         let config = Config(HashMap::from([(

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/minio_generated_read_test.rs

@@ -260,6 +260,118 @@ async fn reads_minio_generated_sse_kms_multipart_fixture() {
     assert_fixture_round_trip("sse-kms-multipart-8m", 8 * 1024 * 1024).await;
 }
 
+/// Read an SSE-C fixture, supplying the customer key the way a client does.
+///
+/// SSE-C needs no KMS at all — the key arrives on the request — so this path
+/// shares nothing with the managed-SSE reads above beyond the fixture loader.
+async fn read_ssec_fixture_plaintext(
+    encrypted: Vec<u8>,
+    object_info: ObjectInfo,
+    customer_key_b64: &str,
+    customer_key_md5_b64: &str,
+) -> Result<Vec<u8>, String> {
+    let object_size = object_info.size;
+    reset_sse_dek_provider();
+
+    let mut headers = http::HeaderMap::new();
+    headers.insert(
+        http::HeaderName::from_static("x-amz-server-side-encryption-customer-algorithm"),
+        http::HeaderValue::from_static("AES256"),
+    );
+    headers.insert(
+        http::HeaderName::from_static("x-amz-server-side-encryption-customer-key"),
+        http::HeaderValue::from_str(customer_key_b64).expect("fixture customer key is a header value"),
+    );
+    headers.insert(
+        http::HeaderName::from_static("x-amz-server-side-encryption-customer-key-md5"),
+        http::HeaderValue::from_str(customer_key_md5_b64).expect("fixture customer key md5 is a header value"),
+    );
+
+    let resolver = SseObjectEncryptionResolver;
+    let (mut reader, offset, length) = GetObjectReader::new_with_resolver(
+        Box::new(Cursor::new(encrypted)),
+        None,
+        &object_info,
+        &ObjectOptions::default(),
+        &headers,
+        Some(&resolver),
+    )
+    .await
+    .map_err(|err| format!("construct GetObjectReader from MinIO SSE-C fixture: {err:?}"))?;
+
+    if offset != 0 || length != object_size {
+        return Err(format!("unexpected fixture range offset={offset} length={length} size={object_size}"));
+    }
+
+    let mut plaintext = Vec::new();
+    reader
+        .read_to_end(&mut plaintext)
+        .await
+        .map_err(|err| format!("read plaintext from MinIO SSE-C fixture: {err}"))?;
+    Ok(plaintext)
+}
+
+/// The interop claim must hold on the production key entry point, not only on
+/// the test-only injection channel every other case here uses.
+#[tokio::test]
+#[ignore = "requires generated MinIO fixture data and a local static KMS key"]
+async fn reads_minio_generated_sse_s3_fixture_through_production_master_key_env() {
+    let (object_info, encrypted, expected_sha256) = load_fixture_reader_input("sse-s3-multipart-8m").await;
+
+    let plaintext = read_fixture_plaintext_via_production_env(encrypted, object_info, minio_static_kms_key_b64())
+        .await
+        .expect("fixture must restore through RUSTFS_SSE_S3_MASTER_KEY");
+
+    assert_eq!(sha256_hex(&plaintext), expected_sha256);
+}
+
+/// SSE-C is the one managed shape needing no KMS: the customer supplies the key
+/// on every request, so this measures the read path alone.
+#[tokio::test]
+#[ignore = "requires generated MinIO fixture data"]
+async fn reads_minio_generated_sse_c_multipart_fixture() {
+    // The fixture lab's fixed SSE-C key; recorded in the case's request.json.
+    const SSEC_KEY_B64: &str = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=";
+    const SSEC_KEY_MD5_B64: &str = "tP/LI3N87DFaSk0aoqYgzg==";
+
+    let (object_info, encrypted, expected_sha256) = load_fixture_reader_input("sse-c-multipart-8m").await;
+
+    let plaintext = read_ssec_fixture_plaintext(encrypted, object_info, SSEC_KEY_B64, SSEC_KEY_MD5_B64)
+        .await
+        .expect("MinIO SSE-C fixture must restore with the customer key");
+
+    assert_eq!(sha256_hex(&plaintext), expected_sha256);
+}
+
+/// The read path skips the stored-MD5 comparison for MinIO SSE-C objects,
+/// which store no MD5. This holds the line that made that safe: the customer
+/// key is still proven by the object-key unseal, so a wrong key must fail even
+/// with nothing to compare it against.
+#[tokio::test]
+#[ignore = "requires generated MinIO fixture data"]
+async fn sse_c_wrong_customer_key_still_fails_without_a_stored_md5() {
+    // A well-formed 32-byte key that is not the one the fixture was sealed
+    // with, sent with its own correct MD5 so the request itself is valid.
+    const WRONG_KEY_B64: &str = "AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=";
+    const WRONG_KEY_MD5_B64: &str = "0YB4bMPPCf9SNlqiKmM0uQ==";
+
+    let (object_info, encrypted, expected_sha256) = load_fixture_reader_input("sse-c-multipart-8m").await;
+
+    let result = read_ssec_fixture_plaintext(encrypted, object_info, WRONG_KEY_B64, WRONG_KEY_MD5_B64).await;
+
+    match result {
+        Err(_) => {}
+        // Never reached today, and asserted rather than assumed: if a future
+        // change let a wrong key through, returning the real plaintext would be
+        // the worst possible outcome.
+        Ok(plaintext) => assert_ne!(
+            sha256_hex(&plaintext),
+            expected_sha256,
+            "a wrong SSE-C customer key must never restore the original plaintext"
+        ),
+    }
+}
+
 #[tokio::test]
 #[ignore = "requires generated MinIO fixture data and a local static KMS key"]
 async fn rejects_minio_generated_sse_s3_fixture_with_wrong_kms_key() {
@@ -288,6 +400,55 @@ async fn rejects_minio_generated_sse_s3_fixture_with_truncated_ciphertext() {
     }
 }
 
+/// Read a fixture through the **production** provider selection.
+///
+/// [`read_fixture_plaintext`] injects the master key through
+/// `__RUSTFS_SSE_SIMPLE_CMK`, which is `#[cfg(test)]`-only, so on its own it
+/// proves nothing about a deployment: it never reaches
+/// `LocalSseDekProvider::new_from_env`. This variant sets only
+/// `RUSTFS_SSE_S3_MASTER_KEY` — the sole production entry point — so the
+/// interop claim rests on the path operators actually run (backlog#1638).
+async fn read_fixture_plaintext_via_production_env(
+    encrypted: Vec<u8>,
+    object_info: ObjectInfo,
+    master_key_b64: String,
+) -> Result<Vec<u8>, String> {
+    let object_size = object_info.size;
+    reset_sse_dek_provider();
+
+    async_with_vars(
+        [
+            ("__RUSTFS_SSE_SIMPLE_CMK", None::<String>),
+            ("RUSTFS_SSE_S3_MASTER_KEY", Some(master_key_b64)),
+        ],
+        async move {
+            let resolver = SseObjectEncryptionResolver;
+            let (mut reader, offset, length) = GetObjectReader::new_with_resolver(
+                Box::new(Cursor::new(encrypted)),
+                None,
+                &object_info,
+                &ObjectOptions::default(),
+                &http::HeaderMap::new(),
+                Some(&resolver),
+            )
+            .await
+            .map_err(|err| format!("construct GetObjectReader from MinIO raw fixture: {err:?}"))?;
+
+            if offset != 0 || length != object_size {
+                return Err(format!("unexpected fixture range offset={offset} length={length} size={object_size}"));
+            }
+
+            let mut plaintext = Vec::new();
+            reader
+                .read_to_end(&mut plaintext)
+                .await
+                .map_err(|err| format!("read plaintext from MinIO raw fixture: {err}"))?;
+            Ok(plaintext)
+        },
+    )
+    .await
+}
+
 async fn assert_fixture_round_trip(case_id: &str, expected_size: i64) {
     let (object_info, encrypted, expected_sha256) = load_fixture_reader_input(case_id).await;
     // `ObjectInfo.size` is the on-disk size. For SSE objects that is the

rustfs/src/storage/sse.rs

@@ -2061,10 +2061,20 @@ pub async fn sse_prepare_encryption(request: PrepareEncryptionRequest<'_>) -> Re
 /// }
 /// ```
 pub async fn sse_decryption(request: DecryptionRequest<'_>) -> Result<Option<DecryptionMaterial>, ApiError> {
-    // Check for SSE-C encryption
+    // Check for SSE-C encryption.
+    //
+    // The stored customer-algorithm marker is what RustFS writes, but a
+    // MinIO-written object has only the internal sealed-key slot: MinIO keeps
+    // the customer algorithm on the request and synthesizes it back onto the
+    // response, never persisting it. Recognizing that slot as well is what lets
+    // a migrated SSE-C object be read at all; the customer key still has to be
+    // supplied, and is still checked against the stored MD5 below.
     if request
         .metadata
         .contains_key("x-amz-server-side-encryption-customer-algorithm")
+        || request
+            .metadata
+            .contains_key(MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER)
     {
         let (key, key_md5) = match (request.sse_customer_key, request.sse_customer_key_md5) {
             (Some(k), Some(md5)) => (k, md5),
@@ -2078,7 +2088,21 @@ pub async fn sse_decryption(request: DecryptionRequest<'_>) -> Result<Option<Dec
 
         // Verify that the provided key MD5 matches the stored MD5 for security
         let stored_md5 = request.metadata.get("x-amz-server-side-encryption-customer-key-md5");
-        verify_ssec_key_match(key_md5, stored_md5)?;
+        // MinIO stores no customer-key MD5 — it keeps that header on the request
+        // and returns it on the response — so requiring one would make every
+        // migrated SSE-C object unreadable. Skipping the comparison when there is
+        // nothing to compare against does not weaken the check it performs: the
+        // stored MD5 is an early, friendlier rejection, while the key itself is
+        // proven by the object-key unseal below, whose AEAD fails on a wrong key.
+        // `sse_c_wrong_customer_key_still_fails_without_a_stored_md5` holds that
+        // line. Objects that *do* carry a stored MD5 are unaffected.
+        let minio_ssec_without_stored_md5 = stored_md5.is_none()
+            && request
+                .metadata
+                .contains_key(MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER);
+        if !minio_ssec_without_stored_md5 {
+            verify_ssec_key_match(key_md5, stored_md5)?;
+        }
 
         let mut material = apply_ssec_decryption_material(request.bucket, request.key, request.metadata, key, key_md5).await?;
         material.customer_key_md5 = Some(key_md5.clone());
@@ -4739,6 +4763,43 @@ mod tests {
     }
 
     #[cfg(feature = "rio-v2")]
+    /// A stored customer-key MD5 must still be compared against the one the
+    /// request presents.
+    ///
+    /// The read path skips that comparison for MinIO SSE-C objects, which store
+    /// no MD5. Nothing pinned the check for objects that *do* store one —
+    /// disabling it outright left this file's 115 tests green — so a later
+    /// widening of that skip would have gone unnoticed. The mismatch has to be
+    /// refused here, at the request boundary, rather than surfacing later as a
+    /// decryption failure.
+    #[tokio::test]
+    async fn ssec_stored_md5_mismatch_is_refused_when_an_md5_is_stored() {
+        let key = SSECustomerKey::from(BASE64_STANDARD.encode([0x11u8; 32]));
+        let provided_md5 = SSECustomerKeyMD5::from(md5_base64(&[0x11u8; 32]));
+
+        let metadata = HashMap::from([
+            (SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string()),
+            // A stored MD5 that belongs to a different key.
+            ("x-amz-server-side-encryption-customer-key-md5".to_string(), md5_base64(&[0x22u8; 32])),
+        ]);
+
+        let error = sse_decryption(DecryptionRequest {
+            bucket: "bucket",
+            key: "object",
+            metadata: &metadata,
+            sse_customer_key: Some(&key),
+            sse_customer_key_md5: Some(&provided_md5),
+            principal: None,
+        })
+        .await
+        .expect_err("a stored MD5 that does not match the request must be refused");
+
+        assert!(
+            format!("{error:?}").contains("did not match"),
+            "expected the parameter-mismatch refusal, got {error:?}"
+        );
+    }
+
     #[tokio::test]
     async fn test_sse_kms_roundtrip_persists_and_uses_minio_context() {
         use rustfs_kms::types::{CreateKeyRequest, KeyUsage};