refactor(admin): route plugin handler auth through authorize_admin_request

The plugin/extension admin family carried seven byte-near copies of the
admin auth preamble (extract credentials, check_key_valid, read RemoteAddr
out of the extensions, call validate_admin_request). Each copy is a place
the gate can drift, which is exactly the review surface rustfs/backlog#1829
tracks.

Every one of the seven is a per-file wrapper with no resource scope and no
audit seam, so it folds onto the shared `authorize_admin_request` gate
without changing the decision. The wrappers keep their own missing-
credentials pre-check, following the pattern established in
`kms_management.rs`: the shared gate reports "get cred failed", while these
endpoints have always reported "authentication required" (six sites) and
"missing credentials" (object_data_cache), and that response must stay
byte-identical. New tests pin each message.

Action sets, `deny_only=false`, the RemoteAddr lookup, and the wrapper
signatures are unchanged, so authenticated-but-unauthorized and
wrong-credential responses are unchanged too.

crates/e2e_test/src/multipart_auth_test.rs

@@ -17,6 +17,7 @@
 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,
@@ -348,6 +349,71 @@ 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
@@ -1534,59 +1600,6 @@ 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>> {
@@ -2584,512 +2597,182 @@ 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_metadata_field_covered_by_starts_with()
+async fn test_anonymous_post_object_accepts_fields_covered_by_policy_conditions()
 -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
 
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    // (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 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();
+    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 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()
+        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,
+        )
         .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(())
 }
@@ -3440,64 +3123,6 @@ 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, ObjectIO as _, ObjectOperations as _},
+    object::{EcstoreObjectIO, HTTPPreconditions, ObjectOperations as _},
 };
 use crate::{
     bucket::{metadata_sys::get_replication_config, versioning::VersioningApi as _, versioning_sys::BucketVersioningSys},
@@ -2009,80 +2009,93 @@ pub async fn apply_bucket_usage_memory_overlay(data_usage_info: &mut DataUsageIn
 }
 
 // Helper functions for DataUsageCache operations
-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};
+
+/// 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;
     use crate::object_api::ObjectOptions;
     use http::HeaderMap;
-    use rand::RngExt;
-    use std::path::Path;
-    use std::time::Duration;
-    use tokio::time::sleep;
 
-    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;
-            }
-            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;
-                }
+    match store
+        .get_object_reader(
+            RUSTFS_META_BUCKET,
+            key,
+            None,
+            HeaderMap::new(),
+            &ObjectOptions {
+                no_lock: true,
+                ..Default::default()
             },
-        }
-        retries += 1;
-        let dur = {
-            let mut rng = rand::rng();
-            rng.random_range(0..1_000)
-        };
-        sleep(Duration::from_millis(dur)).await;
+        )
+        .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),
     }
-    Ok(d)
+}
+
+/// 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 std::path::Path;
+
+    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()),
+                },
+            }
+        },
+        DATA_USAGE_CACHE_LOAD_ATTEMPTS,
+        DATA_USAGE_CACHE_LOAD_BASE_DELAY,
+        DATA_USAGE_CACHE_LOAD_MAX_DELAY,
+    )
+    .await
 }
 
 /// Persist the current in-memory compression total to the backend.
@@ -2220,6 +2233,7 @@ 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;
@@ -2452,6 +2466,128 @@ 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].clone(),
+                store.pools[d.pool_index as usize].disk_set[d.set_index as usize].as_ref(),
                 DATA_USAGE_CACHE_NAME,
             )
             .await

crates/ecstore/src/disk/local.rs

@@ -9745,7 +9745,8 @@ impl DiskAPI for LocalDisk {
                 {
                     let fsync_started = rustfs_io_metrics::put_stage_timer();
                     if let Err(err) =
-                        os::fsync_dir_with_namespace_file_sync_limit(dst_parent, mutation_lease.clone(), admission).await
+                        os::fsync_dst_dir_group_commit_or_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,
@@ -13092,6 +13093,80 @@ 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,6 +676,18 @@ 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,4 +157,219 @@ 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,18 +324,11 @@ 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 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.
+/// 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.
 #[derive(Clone, Default, Serialize, Deserialize)]
 pub struct StaticConfig {
     /// Key identifier (name) for the single configured key

docs/operations/kms-backend-security.md

@@ -14,41 +14,24 @@ 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: what carries over, and what does not
+## Migrating from MinIO: encrypted objects do not carry over
 
-> **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.
+> **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.
 
-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 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:
 
 - **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.
 
-For any shape that does not read, the options are unchanged:
+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:
 
 - 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 | 85 |  |
+| multipart_auth_test | 75 |  |
 | 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::validate_admin_request,
+    auth::authorize_admin_request,
     router::{AdminOperation, Operation, S3Router},
     runtime_sources::default_admin_usecase,
     storage_api::cluster::{
@@ -24,11 +24,10 @@ 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, RemoteAddr};
+use crate::server::{ADMIN_PREFIX, ReadinessDegradedReason};
 use http::{HeaderMap, HeaderValue, StatusCode};
 use hyper::Method;
 use matchit::Params;
@@ -66,23 +65,15 @@ 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<()> {
-    let Some(input_cred) = &req.credentials else {
+    if req.credentials.is_none() {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    };
+    }
 
-    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
+    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await?;
+    Ok(())
 }
 
 fn build_json_response(
@@ -953,6 +944,30 @@ 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::validate_admin_request,
+    auth::authorize_admin_request,
     handlers::{cluster_snapshot, plugins_instances, system},
     plugin_contract::{
         PluginContractDomain, PluginInstanceDiagnosticCode, PluginInstanceDiagnosticCount, PluginInstanceEntry,
@@ -22,8 +22,7 @@ use crate::admin::{
     },
     router::{AdminOperation, Operation, S3Router},
 };
-use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use http::{HeaderMap, HeaderValue, StatusCode};
 use hyper::Method;
 use matchit::Params;
@@ -183,42 +182,26 @@ 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<()> {
-    let Some(input_cred) = &req.credentials else {
+    if req.credentials.is_none() {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    };
+    }
 
-    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
+    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await?;
+    Ok(())
 }
 
+/// 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<()> {
-    let Some(input_cred) = &req.credentials else {
+    if req.credentials.is_none() {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    };
+    }
 
-    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
+    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::GetBucketTargetAction)]).await?;
+    Ok(())
 }
 
 fn build_json_response(
@@ -320,6 +303,36 @@ 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,12 +21,11 @@
 //! 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::validate_admin_request;
+use crate::admin::auth::authorize_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::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use http::{HeaderMap, HeaderValue};
 use hyper::{Method, StatusCode};
 use matchit::Params;
@@ -76,17 +75,14 @@ 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<()> {
-    let Some(input_cred) = req.credentials.as_ref() else {
+    if req.credentials.is_none() {
         return Err(s3_error!(InvalidRequest, "missing credentials"));
-    };
-    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
+    }
+    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
+    Ok(())
 }
 
 fn json_response<T: Serialize>(body: &T) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -208,6 +204,30 @@ 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::validate_admin_request,
+    auth::authorize_admin_request,
     plugin_contract::{
         PluginCatalogAdminDiscovery, PluginCatalogDomainEntry, PluginCatalogEntry, PluginCatalogResponse, PluginContractDomain,
         PluginContractEntrypointKind, PluginContractPackaging, PluginDistributionContract, PluginRuntimeContract,
@@ -21,8 +21,7 @@ use crate::admin::{
     router::{AdminOperation, Operation, S3Router},
     runtime_sources::default_admin_usecase,
 };
-use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use http::{HeaderMap, HeaderValue, StatusCode};
 use hyper::Method;
 use matchit::Params;
@@ -114,23 +113,15 @@ 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<()> {
-    let Some(input_cred) = &req.credentials else {
+    if req.credentials.is_none() {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    };
+    }
 
-    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
+    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await?;
+    Ok(())
 }
 
 fn build_json_response(
@@ -175,6 +166,30 @@ 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::validate_admin_request,
+    auth::authorize_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,10 +29,9 @@ use crate::admin::{
     },
     router::{AdminOperation, Operation, S3Router},
 };
-use crate::auth::{check_key_valid, get_session_token};
 use crate::server::{
-    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,
+    ADMIN_PREFIX, 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;
@@ -563,42 +562,26 @@ 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<()> {
-    let Some(input_cred) = &req.credentials else {
+    if req.credentials.is_none() {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    };
+    }
 
-    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
+    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::GetBucketTargetAction)]).await?;
+    Ok(())
 }
 
+/// 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<()> {
-    let Some(input_cred) = &req.credentials else {
+    if req.credentials.is_none() {
         return Err(s3_error!(InvalidRequest, "authentication required"));
-    };
+    }
 
-    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
+    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::SetBucketTargetAction)]).await?;
+    Ok(())
 }
 
 fn plugin_instance_mutation_block_reason(
@@ -942,6 +925,36 @@ 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,11 +747,12 @@ pub struct S3ErrorMessageCompatService<S> {
     inner: S,
 }
 
-impl<S, RestBody, GrpcBody> Service<HttpRequest<Incoming>> for S3ErrorMessageCompatService<S>
+impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for S3ErrorMessageCompatService<S>
 where
-    S: Service<HttpRequest<Incoming>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
+    S: Service<HttpRequest<ReqBody>, 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,
@@ -764,28 +765,27 @@ where
         self.inner.poll_ready(cx)
     }
 
-    fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
+    fn call(&mut self, req: HttpRequest<ReqBody>) -> 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 } => {
-                    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 })
+                    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 })
                 }
                 HybridBody::Grpc { grpc_body } => Response::from_parts(parts, HybridBody::Grpc { grpc_body }),
             };
@@ -886,11 +886,12 @@ pub struct IcebergRestErrorCompatService<S> {
     inner: S,
 }
 
-impl<S, RestBody, GrpcBody> Service<HttpRequest<Incoming>> for IcebergRestErrorCompatService<S>
+impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for IcebergRestErrorCompatService<S>
 where
-    S: Service<HttpRequest<Incoming>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
+    S: Service<HttpRequest<ReqBody>, 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,
@@ -903,18 +904,21 @@ where
         self.inner.poll_ready(cx)
     }
 
-    fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
+    fn call(&mut self, req: HttpRequest<ReqBody>) -> 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 } if should_convert => {
+                HybridBody::Rest { rest_body } => {
                     let (rest_body, converted_status) = convert_iceberg_error_in_xml(
                         rest_body,
                         parts.status,
@@ -932,7 +936,6 @@ 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 }),
             };
 
@@ -1045,11 +1048,12 @@ pub struct ObjectAttributesEtagFixService<S> {
     inner: S,
 }
 
-impl<S, RestBody, GrpcBody> Service<HttpRequest<Incoming>> for ObjectAttributesEtagFixService<S>
+impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for ObjectAttributesEtagFixService<S>
 where
-    S: Service<HttpRequest<Incoming>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
+    S: Service<HttpRequest<ReqBody>, 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,
@@ -1062,27 +1066,26 @@ where
         self.inner.poll_ready(cx)
     }
 
-    fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
+    fn call(&mut self, req: HttpRequest<ReqBody>) -> 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 } => {
-                    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 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 }),
             };
@@ -1144,12 +1147,11 @@ where
 
         Box::pin(async move {
             let response = inner.call(req).await?;
-            let (mut parts, body) = response.into_parts();
-
-            if !is_bodyless_status(parts.status) {
-                return Ok(Response::from_parts(parts, body));
+            if !is_bodyless_status(response.status()) {
+                return Ok(response);
             }
 
+            let (mut parts, body) = response.into_parts();
             let response = match body {
                 HybridBody::Rest { .. } => {
                     parts.headers.remove(http::header::CONTENT_LENGTH);
@@ -1802,7 +1804,7 @@ fn strip_quotes_from_first_etag(xml: String) -> String {
     fixed
 }
 
-fn is_object_attributes_request(req: &HttpRequest<Incoming>) -> bool {
+fn is_object_attributes_request<B>(req: &HttpRequest<B>) -> bool {
     if req.method() != Method::GET {
         return false;
     }
@@ -1967,11 +1969,12 @@ fn apply_bucket_cors_result(response_headers: &mut HeaderMap, bucket_cors_header
     }
 }
 
-impl<S, ResBody> Service<HttpRequest<Incoming>> for ConditionalCorsService<S>
+impl<S, ReqBody, ResBody> Service<HttpRequest<ReqBody>> for ConditionalCorsService<S>
 where
-    S: Service<HttpRequest<Incoming>, Response = Response<ResBody>> + Clone + Send + 'static,
+    S: Service<HttpRequest<ReqBody>, 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>;
@@ -1982,7 +1985,14 @@ where
         self.inner.poll_ready(cx).map_err(Into::into)
     }
 
-    fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
+    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) });
+        }
+
         let path = req.uri().path().to_string();
         let method = req.method().clone();
         let request_headers = req.headers().clone();
@@ -1990,7 +2000,7 @@ where
         let is_s3 = ConditionalCorsLayer::is_s3_path(&path);
         let is_root = path == "/";
 
-        if method == Method::OPTIONS {
+        if is_options {
             let has_acrm = request_headers.contains_key(cors::request::ACCESS_CONTROL_REQUEST_METHOD);
 
             if is_root {
@@ -2192,6 +2202,7 @@ 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;
@@ -3783,6 +3794,188 @@ 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,
@@ -4270,6 +4463,63 @@ 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,118 +260,6 @@ 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() {
@@ -400,55 +288,6 @@ 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,20 +2061,10 @@ 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.
-    //
-    // 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.
+    // Check for SSE-C encryption
     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),
@@ -2088,21 +2078,7 @@ 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");
-        // 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)?;
-        }
+        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());
@@ -4763,43 +4739,6 @@ 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};