feat(gateway): serve bucket CORS mutations

Route PutBucketCors and DeleteBucketCors through owned use cases and shared authorization, preserving audit, storage, and legacy edge behavior.

Coverage includes eight refusal and two root matrix scenarios per stack, a wrong-resource refusal, missing-context refusal, and CORS persistence cases. The scoped suite passed 125 tests on 05554bee0. Bypassing Put or Delete authorization failed the gateway refusal matrix; skipping each storage mutation failed the CORS persistence test; allowing a missing context failed the access unit test.

Not verified: full workspace CI and the missing-bucket plus invalid-configuration error ordering. The delete audit status remains 200 while the response is 204.

rustfs/src/app/bucket_usecase/cors.rs

@@ -15,42 +15,53 @@
 //! Bucket CORS configuration use-case operations.
 
 use super::*;
-use crate::app::storage_api::s3::dto::{
-    DeleteBucketCorsInput, DeleteBucketCorsOutput, GetBucketCorsInput, PutBucketCorsInput, PutBucketCorsOutput,
-};
-use crate::app::storage_api::s3::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
+use crate::app::storage_api::s3::dto::GetBucketCorsInput;
+use crate::app::storage_api::s3::{S3Error, S3ErrorCode, S3Request, S3Result};
+use rustfs_gateway_types::cors_tagging::parse_cors;
+use rustfs_gateway_types::persistence::serialize_cors_dto;
 
 impl DefaultBucketUsecase {
-    #[instrument(level = "debug", skip(self, req))]
+    /// Removes the CORS document of `input.bucket`; audited as an audit-only operation.
+    ///
+    /// # Errors
+    ///
+    /// `InternalError` when the store is not initialized or the access phase left no mutation
+    /// snapshot; the storage error of the bucket lookup or of the write.
+    #[instrument(level = "debug", skip(self, envelope, input))]
     pub async fn execute_delete_bucket_cors(
         &self,
-        req: S3Request<DeleteBucketCorsInput>,
-    ) -> S3Result<S3Response<DeleteBucketCorsOutput>> {
-        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
-        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
-        let DeleteBucketCorsInput { bucket, .. } = req.input;
+        envelope: RequestEnvelope,
+        input: gateway_dto::DeleteBucketCorsInput,
+    ) -> rustfs_s3_types::S3Result<gateway_dto::DeleteBucketCorsOutput> {
+        // The audit entry records 200, as on main; the wire answer is the operation's 204.
+        RequestEnvelope::audit_only_operation(envelope, S3Operation::DeleteBucketCors, StatusCode::OK, |envelope| async move {
+            let bucket = input.bucket.as_str();
+            let expected_incarnation_id = bucket_config_mutation_incarnation(&envelope, bucket)?;
+            let request_context = envelope.request_context().cloned();
 
-        let Some(store) = self.object_store() else {
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
-        };
+            let Some(store) = self.object_store() else {
+                return Err(rustfs_s3_types::s3_error!(InternalError, "Not init"));
+            };
 
-        store
-            .get_bucket_info(&bucket, &BucketOptions::default())
-            .await
-            .map_err(ApiError::from)?;
+            store
+                .get_bucket_info(bucket, &BucketOptions::default())
+                .await
+                .map_err(ApiError::from)?;
 
-        delete_bucket_config_for_incarnation(&bucket, BUCKET_CORS_CONFIG, expected_incarnation_id)
-            .await
-            .map_err(ApiError::from)?;
+            delete_bucket_config_for_incarnation(bucket, BUCKET_CORS_CONFIG, expected_incarnation_id)
+                .await
+                .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket cors", request_context, false).await;
+            notify_bucket_metadata_reload(bucket.to_owned(), "delete bucket cors", request_context, false).await;
 
-        let item = sr_bucket_meta_item(bucket.clone(), "cors-config");
-        if let Err(err) = site_replication_bucket_meta_hook(item).await {
-            warn!(bucket = %bucket, error = ?err, "site replication bucket cors delete hook failed");
-        }
+            let item = sr_bucket_meta_item(bucket.to_owned(), "cors-config");
+            if let Err(err) = site_replication_bucket_meta_hook(item).await {
+                warn!(bucket = %bucket, error = ?err, "site replication bucket cors delete hook failed");
+            }
 
-        Ok(S3Response::new(DeleteBucketCorsOutput {}))
+            Ok(gateway_dto::DeleteBucketCorsOutput::default())
+        })
+        .await
     }
 
     #[instrument(level = "debug", skip(self, req))]
@@ -90,63 +101,75 @@ impl DefaultBucketUsecase {
         Ok(cors_configuration)
     }
 
-    #[instrument(level = "debug", skip(self, req))]
+    /// Stores the CORS document of `input.bucket` with the persisted writer; audited as an
+    /// audit-only operation.
+    ///
+    /// # Errors
+    ///
+    /// `InternalError` when the store is not initialized or the access phase left no mutation
+    /// snapshot; the storage error of the bucket lookup; `MalformedXML` when the persisted reader
+    /// refuses the document, after the bucket lookup as before.
+    #[instrument(level = "debug", skip(self, envelope, input))]
     pub async fn execute_put_bucket_cors(
         &self,
-        req: S3Request<PutBucketCorsInput>,
-        prepared: S3Result<PersistedCorsConfiguration>,
-    ) -> S3Result<S3Response<PutBucketCorsOutput>> {
-        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
-        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
-        let PutBucketCorsInput { bucket, .. } = req.input;
+        envelope: RequestEnvelope,
+        input: gateway_dto::PutBucketCorsInput,
+    ) -> rustfs_s3_types::S3Result<gateway_dto::PutBucketCorsOutput> {
+        RequestEnvelope::audit_only_operation(envelope, S3Operation::PutBucketCors, StatusCode::OK, |envelope| async move {
+            let bucket = input.bucket.as_str();
+            let expected_incarnation_id = bucket_config_mutation_incarnation(&envelope, bucket)?;
+            let request_context = envelope.request_context().cloned();
 
-        let Some(store) = self.object_store() else {
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
-        };
+            let Some(store) = self.object_store() else {
+                return Err(rustfs_s3_types::s3_error!(InternalError, "Not init"));
+            };
 
-        store
-            .get_bucket_info(&bucket, &BucketOptions::default())
-            .await
-            .map_err(ApiError::from)?;
+            store
+                .get_bucket_info(bucket, &BucketOptions::default())
+                .await
+                .map_err(ApiError::from)?;
 
-        let cors_configuration = prepared?;
-        let data = serialize_config(&cors_configuration)?;
-        update_bucket_config_for_incarnation(&bucket, BUCKET_CORS_CONFIG, data, expected_incarnation_id)
-            .await
-            .map_err(ApiError::from)?;
+            // The persisted reader decides what may be stored, on both edges.
+            let cors_configuration = parse_cors(&serialize_cors_dto(&input.cors_configuration))
+                .map_err(|err| rustfs_s3_types::s3_error!(MalformedXML, "{err}"))?;
+            let data = serialize_config(&cors_configuration)?;
+            update_bucket_config_for_incarnation(bucket, BUCKET_CORS_CONFIG, data.clone(), expected_incarnation_id)
+                .await
+                .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket cors", request_context, false).await;
+            notify_bucket_metadata_reload(bucket.to_owned(), "put bucket cors", request_context, false).await;
 
-        let mut item = sr_bucket_meta_item(bucket.clone(), "cors-config");
-        item.cors =
-            Some(serialize_config(&cors_configuration).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?);
-        if let Err(err) = site_replication_bucket_meta_hook(item).await {
-            warn!(bucket = %bucket, error = ?err, "site replication bucket cors hook failed");
-        }
+            let mut item = sr_bucket_meta_item(bucket.to_owned(), "cors-config");
+            item.cors = Some(String::from_utf8(data).map_err(to_internal_error)?);
+            if let Err(err) = site_replication_bucket_meta_hook(item).await {
+                warn!(bucket = %bucket, error = ?err, "site replication bucket cors hook failed");
+            }
 
-        Ok(S3Response::new(PutBucketCorsOutput::default()))
+            Ok(gateway_dto::PutBucketCorsOutput::default())
+        })
+        .await
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::app::bucket_usecase::test_support::build_request;
-    use crate::app::storage_api::s3::dto::CORSConfiguration;
+    use crate::app::bucket_usecase::test_support::{build_request, gateway_bucket, owned_envelope};
     use http::Method;
 
     #[tokio::test]
     async fn execute_delete_bucket_cors_returns_internal_error_when_store_uninitialized() {
-        let input = DeleteBucketCorsInput::builder()
-            .bucket("test-bucket".to_string())
-            .build()
-            .unwrap();
-
-        let req = build_request(input, Method::DELETE);
+        let input = gateway_dto::DeleteBucketCorsInput {
+            bucket: gateway_bucket("test-bucket"),
+            ..Default::default()
+        };
         let usecase = DefaultBucketUsecase::without_context();
 
-        let err = usecase.execute_delete_bucket_cors(req).await.unwrap_err();
-        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        let err = usecase
+            .execute_delete_bucket_cors(owned_envelope(Method::DELETE), input)
+            .await
+            .unwrap_err();
+        assert_eq!(err.code(), &rustfs_s3_types::S3ErrorCode::InternalError);
     }
 
     #[tokio::test]
@@ -165,19 +188,16 @@ mod tests {
 
     #[tokio::test]
     async fn execute_put_bucket_cors_returns_internal_error_when_store_uninitialized() {
-        let input = PutBucketCorsInput::builder()
-            .bucket("test-bucket".to_string())
-            .cors_configuration(CORSConfiguration::default())
-            .build()
-            .unwrap();
-
-        let req = build_request(input, Method::PUT);
+        let input = gateway_dto::PutBucketCorsInput {
+            bucket: gateway_bucket("test-bucket"),
+            ..Default::default()
+        };
         let usecase = DefaultBucketUsecase::without_context();
 
         let err = usecase
-            .execute_put_bucket_cors(req, Ok(PersistedCorsConfiguration::default()))
+            .execute_put_bucket_cors(owned_envelope(Method::PUT), input)
             .await
             .unwrap_err();
-        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        assert_eq!(err.code(), &rustfs_s3_types::S3ErrorCode::InternalError);
     }
 }

rustfs/src/app/bucket_usecase/test_support.rs

@@ -40,6 +40,16 @@ pub(super) fn build_request<T>(input: T, method: Method) -> S3Request<T> {
     }
 }
 
+/// A bucket name for an owned use-case input.
+pub(super) fn gateway_bucket(name: &str) -> rustfs_gateway_types::BucketName {
+    rustfs_gateway_types::BucketName::new(name).expect("valid bucket name")
+}
+
+/// The envelope of a request that carries nothing but `method`.
+pub(super) fn owned_envelope(method: Method) -> RequestEnvelope {
+    RequestEnvelope::from_s3s(&build_request((), method))
+}
+
 pub(super) fn build_request_with_req_info<T>(input: T, method: Method, req_info: ReqInfo) -> S3Request<T> {
     let mut req = build_request(input, method);
     req.extensions.insert(req_info);

rustfs/src/app/context/envelope.rs

@@ -714,13 +714,6 @@ impl RequestEnvelope {
     }
 }
 
-#[cfg_attr(
-    not(test),
-    expect(
-        dead_code,
-        reason = "the operation batches of rustfs/backlog#2749 are its first callers (task H2)"
-    )
-)]
 impl RequestEnvelope {
     /// Runs `handler` on this request and records its outcome as an audit-only
     /// entry for `op`, the owned form of the legacy edge's `audit_only_operation`.

rustfs/src/server/gateway_handlers/access.rs

@@ -18,15 +18,20 @@
 //! Responsible for: [`admit`], which builds the envelope with the access markers, installs the
 //! caller and runs the operation-aware admission, each through the function the legacy check
 //! calls (`storage/access.rs`).
+//! Also the gateway-edge form of the one authorization shape several families share:
+//! [`authorize_bucket_config_mutation`].
 //! NOT responsible for: an operation's own authorization (`Served::authorize`), or verifying the
 //! signature (the gateway did, and only a verified caller has a principal).
 //! Upstream: `GatewayBackend::serve`. Downstream: the storage layer's access functions.
 
 use crate::app::RequestEnvelope;
 use crate::runtime_sources::ServerContextSlot;
-use crate::storage_api::server::http::access::{VerifiedSignature, admit_operation, resolve_req_info};
+use crate::storage_api::server::http::access::{
+    VerifiedSignature, admit_operation, authorize_bucket_config_mutation as authorize_config_mutation, resolve_req_info,
+};
 use rustfs_gateway::sig::{SigFamily, SigLocation};
 use rustfs_gateway::{RequestContextView, RequestPrincipal};
+use rustfs_policy::policy::action::Action;
 use rustfs_s3_types::S3Result;
 use std::sync::Arc;
 
@@ -53,6 +58,25 @@ pub(super) async fn admit(server_ctx: &Arc<ServerContextSlot>, context: &Request
     Ok(envelope)
 }
 
+/// The bucket-configuration mutation shape on the gateway edge: the shared function the legacy
+/// hook calls, with the server context the envelope carries.
+///
+/// # Errors
+///
+/// `InternalError` when the envelope carries no server context; otherwise what the shared
+/// function answers.
+pub(super) async fn authorize_bucket_config_mutation(
+    envelope: &mut RequestEnvelope,
+    bucket: &str,
+    action: Action,
+) -> S3Result<()> {
+    let server_ctx = envelope
+        .server_context()
+        .cloned()
+        .ok_or_else(|| rustfs_s3_types::s3_error!(InternalError, "the request carries no server context"))?;
+    Ok(authorize_config_mutation(&server_ctx, envelope, bucket, action).await?)
+}
+
 /// The gateway's verdict as the access markers read it: a SigV4 signature in the header or the
 /// query. SigV2, a POST policy and an anonymous request mark nothing, as on the legacy edge.
 pub(super) fn verified_signature(context: &RequestContextView) -> VerifiedSignature {
@@ -64,3 +88,34 @@ pub(super) fn verified_signature(context: &RequestContextView) -> VerifiedSignat
         _ => VerifiedSignature::default(),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::RequestHead;
+    use http::{Extensions, HeaderMap, Method};
+    use rustfs_policy::policy::action::S3Action;
+
+    /// Without a server context there is no lock to take and no incarnation to snapshot, so the
+    /// mutation is refused rather than decided.
+    #[tokio::test]
+    async fn a_bucket_config_mutation_without_a_server_context_is_refused() {
+        let head = RequestHead {
+            method: Method::PUT,
+            uri: "/bucket?cors".parse().expect("valid URI"),
+            headers: HeaderMap::new(),
+            credentials: None,
+            region: None,
+            service: None,
+        };
+        let mut envelope = RequestEnvelope::from_parts(head, Extensions::new());
+
+        let action = Action::S3Action(S3Action::PutBucketCorsAction);
+        let refusal = authorize_bucket_config_mutation(&mut envelope, "bucket", action)
+            .await
+            .unwrap_err();
+
+        assert_eq!(refusal.code(), &rustfs_s3_types::S3ErrorCode::InternalError);
+        assert_eq!(refusal.message(), Some("the request carries no server context"));
+    }
+}

rustfs/src/server/gateway_handlers/bucket_configs.rs

@@ -13,6 +13,49 @@
 // limitations under the License.
 
 //! The gateway operations of migration batch A1 (rustfs/backlog#2765): bucket CORS, tagging, public access block and encryption.
-//! None is served yet; the batch appends its rows here.
+//! The stored CORS document is applied to requests by `ConditionalCorsLayer`, not by the gateway.
 
-pub(super) const ROWS: &[super::OpRow] = &[];
+use super::access::authorize_bucket_config_mutation;
+use super::{OpRow, Served, row};
+use crate::app::RequestEnvelope;
+use crate::app::bucket_usecase::DefaultBucketUsecase;
+use crate::runtime_sources::AppContext;
+use rustfs_gateway::{Req, Resp, dto};
+use rustfs_policy::policy::action::{Action, S3Action};
+use rustfs_s3_ops::S3Operation;
+use rustfs_s3_types::S3Result;
+use std::sync::Arc;
+
+pub(super) const ROWS: &[OpRow] = &[row::<dto::PutBucketCors>(), row::<dto::DeleteBucketCors>()];
+
+impl Served for dto::PutBucketCors {
+    const METRIC: S3Operation = S3Operation::PutBucketCors;
+
+    async fn authorize(envelope: &mut RequestEnvelope, request: &Req<Self>) -> S3Result<()> {
+        let action = Action::S3Action(S3Action::PutBucketCorsAction);
+        authorize_bucket_config_mutation(envelope, request.input().bucket.as_str(), action).await
+    }
+
+    async fn execute(context: Option<Arc<AppContext>>, envelope: RequestEnvelope, request: Req<Self>) -> S3Result<Resp<Self>> {
+        let output = DefaultBucketUsecase::with_context(context)
+            .execute_put_bucket_cors(envelope, request.into_input())
+            .await?;
+        Ok(Resp::new(output))
+    }
+}
+
+impl Served for dto::DeleteBucketCors {
+    const METRIC: S3Operation = S3Operation::DeleteBucketCors;
+
+    async fn authorize(envelope: &mut RequestEnvelope, request: &Req<Self>) -> S3Result<()> {
+        let action = Action::S3Action(S3Action::DeleteBucketCorsAction);
+        authorize_bucket_config_mutation(envelope, request.input().bucket.as_str(), action).await
+    }
+
+    async fn execute(context: Option<Arc<AppContext>>, envelope: RequestEnvelope, request: Req<Self>) -> S3Result<Resp<Self>> {
+        let output = DefaultBucketUsecase::with_context(context)
+            .execute_delete_bucket_cors(envelope, request.into_input())
+            .await?;
+        Ok(Resp::new(output))
+    }
+}

rustfs/src/storage/legacy_edge/bucket_configs.rs

@@ -23,6 +23,7 @@ use crate::storage::ecfs::*;
 use crate::storage::helper::audit_only_operation;
 use crate::storage::s3_api;
 use crate::storage::{persisted_from_s3s, persisted_to_s3s, tagging_from_s3s, tagging_to_s3s};
+use rustfs_gateway_types::compat::s3s_5761ddfe::generated::ops::{delete_bucket_cors, put_bucket_cors};
 use rustfs_io_metrics::record_s3_op;
 use rustfs_s3_ops::S3Operation;
 use tracing::instrument;
@@ -33,8 +34,15 @@ impl FS {
         &self,
         req: S3Request<DeleteBucketCorsInput>,
     ) -> S3Result<S3Response<DeleteBucketCorsOutput>> {
+        record_s3_op(S3Operation::DeleteBucketCors);
         let usecase = s3_api::bucket_usecase_for(self);
-        audit_only_operation(req, S3Operation::DeleteBucketCors, |req| usecase.execute_delete_bucket_cors(req)).await
+        through_gateway_seam(
+            req,
+            delete_bucket_cors::input_from_s3s,
+            delete_bucket_cors::output_to_s3s,
+            |envelope, input| usecase.execute_delete_bucket_cors(envelope, input),
+        )
+        .await
     }
 
     pub(in crate::storage::ecfs) async fn delete_bucket_encryption_edge(
@@ -131,9 +139,12 @@ impl FS {
         &self,
         req: S3Request<PutBucketCorsInput>,
     ) -> S3Result<S3Response<PutBucketCorsOutput>> {
-        let prepared = persisted_from_s3s(&req.input.cors_configuration);
+        record_s3_op(S3Operation::PutBucketCors);
         let usecase = s3_api::bucket_usecase_for(self);
-        audit_only_operation(req, S3Operation::PutBucketCors, |req| usecase.execute_put_bucket_cors(req, prepared)).await
+        through_gateway_seam(req, put_bucket_cors::input_from_s3s, put_bucket_cors::output_to_s3s, |envelope, input| {
+            usecase.execute_put_bucket_cors(envelope, input)
+        })
+        .await
     }
 
     pub(in crate::storage::ecfs) async fn put_bucket_encryption_edge(

rustfs/src/storage/storage_api.rs

@@ -150,7 +150,8 @@ pub(crate) mod access_consumer {
     pub(crate) use super::super::access::access_owned;
     #[cfg(feature = "gateway-stack")]
     pub(crate) use super::super::access::{
-        AccessMarkers, VerifiedSignature, admit_operation, authorize_bucket_request, resolve_req_info,
+        AccessMarkers, VerifiedSignature, admit_operation, authorize_bucket_config_mutation, authorize_bucket_request,
+        resolve_req_info,
     };
     pub(crate) use super::super::access::{
         AuthorizationRequest, BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard,

rustfs/src/storage_api.rs

@@ -238,7 +238,8 @@ pub(crate) mod server {
             /// The per-batch authorization shapes, one module per migration batch.
             pub(crate) use crate::storage::storage_api::access_consumer::access_owned;
             pub(crate) use crate::storage::storage_api::access_consumer::{
-                AccessMarkers, VerifiedSignature, admit_operation, authorize_bucket_request, resolve_req_info,
+                AccessMarkers, VerifiedSignature, admit_operation, authorize_bucket_config_mutation, authorize_bucket_request,
+                resolve_req_info,
             };
         }
 

rustfs/tests/embedded_gateway_stack_test/bucket_configs.rs

@@ -12,6 +12,258 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Refusal-matrix rows of migration batch A1 (rustfs/backlog#2765). None is served yet.
+//! Refusal-matrix rows and behaviour of migration batch A1 (rustfs/backlog#2765): bucket CORS,
+//! tagging, public access block and encryption. Every behaviour test runs on both stacks.
 
-pub(crate) const MATRIX: &[super::MatrixRow] = &[];
+use super::{
+    GATEWAY, LEGACY, MatrixRow, Outcome, USER_SECRET_KEY, create_bucket_in_the_store, create_user_allowed_on_one_bucket, outcome,
+    s3_client, with_stack,
+};
+use aws_sdk_s3::Client;
+use aws_sdk_s3::config::interceptors::{
+    BeforeDeserializationInterceptorContextRef, BeforeTransmitInterceptorContextMut, BeforeTransmitInterceptorContextRef,
+};
+use aws_sdk_s3::config::{ConfigBag, Intercept, RuntimeComponents};
+use aws_sdk_s3::error::BoxError;
+use aws_sdk_s3::operation::put_bucket_cors::builders::PutBucketCorsFluentBuilder;
+use aws_sdk_s3::types::{CorsConfiguration, CorsRule};
+use futures::future::BoxFuture;
+use rustfs::embedded::RustFSServer;
+use rustfs_ecstore::api::bucket::metadata_sys;
+use rustfs_s3_ops::S3Operation;
+use std::sync::{Arc, Mutex};
+
+pub(crate) const MATRIX: &[MatrixRow] = &[
+    MatrixRow {
+        operation: "PutBucketCors",
+        metric: S3Operation::PutBucketCors,
+        has_error_body: true,
+        call: put_bucket_cors,
+    },
+    MatrixRow {
+        operation: "DeleteBucketCors",
+        metric: S3Operation::DeleteBucketCors,
+        has_error_body: true,
+        call: delete_bucket_cors,
+    },
+];
+
+fn put_bucket_cors<'a>(client: &'a Client, bucket: &'a str) -> BoxFuture<'a, Outcome> {
+    Box::pin(async move { outcome(put_cors(client, bucket, cors_document()).send().await) })
+}
+
+fn delete_bucket_cors<'a>(client: &'a Client, bucket: &'a str) -> BoxFuture<'a, Outcome> {
+    Box::pin(async move { outcome(client.delete_bucket_cors().bucket(bucket).send().await) })
+}
+
+fn put_cors(client: &Client, bucket: &str, document: CorsConfiguration) -> PutBucketCorsFluentBuilder {
+    client.put_bucket_cors().bucket(bucket).cors_configuration(document)
+}
+
+/// A document with every member of a rule, and a rule with only the required ones.
+fn cors_document() -> CorsConfiguration {
+    let full = CorsRule::builder()
+        .id("full")
+        .allowed_headers("x-a")
+        .allowed_headers("x-b")
+        .allowed_methods("GET")
+        .allowed_methods("PUT")
+        .allowed_origins("https://a.example")
+        .expose_headers("ETag")
+        .max_age_seconds(300)
+        .build();
+    let minimal = CorsRule::builder().allowed_methods("HEAD").allowed_origins("*").build();
+    CorsConfiguration::builder()
+        .set_cors_rules(Some([full, minimal].map(|rule| rule.expect("valid rule")).to_vec()))
+        .build()
+        .expect("valid configuration")
+}
+
+/// One rule allowing `method` from `origin`.
+fn single_rule(method: &str, origin: &str) -> CorsConfiguration {
+    let rule = CorsRule::builder().allowed_methods(method).allowed_origins(origin).build();
+    CorsConfiguration::builder()
+        .cors_rules(rule.expect("valid rule"))
+        .build()
+        .expect("valid configuration")
+}
+
+/// Records the status of the one response a request receives.
+#[derive(Clone, Debug, Default)]
+struct StatusCapture(Arc<Mutex<Option<u16>>>);
+
+impl StatusCapture {
+    fn take(&self) -> Option<u16> {
+        self.0.lock().ok().and_then(|mut slot| slot.take())
+    }
+}
+
+impl Intercept for StatusCapture {
+    fn name(&self) -> &'static str {
+        "bucket-configs-status-capture"
+    }
+
+    fn read_before_deserialization(
+        &self,
+        context: &BeforeDeserializationInterceptorContextRef<'_>,
+        _runtime_components: &RuntimeComponents,
+        _cfg: &mut ConfigBag,
+    ) -> Result<(), BoxError> {
+        let status = context.response().status().as_u16();
+        *self
+            .0
+            .lock()
+            .map_err(|_| std::io::Error::other("status capture mutex was poisoned"))? = Some(status);
+        Ok(())
+    }
+}
+
+fn is_integrity_claim(name: &str) -> bool {
+    name == "content-md5" || name.starts_with("x-amz-checksum-") || name == "x-amz-sdk-checksum-algorithm"
+}
+
+/// Removes every integrity claim the SDK attaches to a request body, before the request is signed,
+/// and records the claims still on the request as it is sent.
+#[derive(Clone, Debug, Default)]
+struct WithoutIntegrityClaim(Arc<Mutex<Option<Vec<String>>>>);
+
+impl WithoutIntegrityClaim {
+    fn sent_claims(&self) -> Option<Vec<String>> {
+        self.0.lock().ok().and_then(|mut slot| slot.take())
+    }
+}
+
+impl Intercept for WithoutIntegrityClaim {
+    fn name(&self) -> &'static str {
+        "bucket-configs-without-integrity-claim"
+    }
+
+    fn modify_before_signing(
+        &self,
+        context: &mut BeforeTransmitInterceptorContextMut<'_>,
+        _runtime_components: &RuntimeComponents,
+        _cfg: &mut ConfigBag,
+    ) -> Result<(), BoxError> {
+        let headers = context.request_mut().headers_mut();
+        let claims: Vec<String> = headers
+            .iter()
+            .map(|(name, _)| name.to_owned())
+            .filter(|name| is_integrity_claim(name))
+            .collect();
+        for name in claims {
+            headers.remove(name);
+        }
+        Ok(())
+    }
+
+    fn read_before_transmit(
+        &self,
+        context: &BeforeTransmitInterceptorContextRef<'_>,
+        _runtime_components: &RuntimeComponents,
+        _cfg: &mut ConfigBag,
+    ) -> Result<(), BoxError> {
+        let sent = context
+            .request()
+            .headers()
+            .iter()
+            .map(|(name, _)| name.to_owned())
+            .filter(|name| is_integrity_claim(name))
+            .collect();
+        *self
+            .0
+            .lock()
+            .map_err(|_| std::io::Error::other("claim capture mutex was poisoned"))? = Some(sent);
+        Ok(())
+    }
+}
+
+/// The bytes PutBucketCors stores for [`cors_document`]: the persisted writer's form.
+const PERSISTED_CORS: &str = concat!(
+    "<CORSConfiguration><CORSRule><AllowedHeader>x-a</AllowedHeader><AllowedHeader>x-b</AllowedHeader>",
+    "<AllowedMethod>GET</AllowedMethod><AllowedMethod>PUT</AllowedMethod><AllowedOrigin>https://a.example</AllowedOrigin>",
+    "<ExposeHeader>ETag</ExposeHeader><ID>full</ID><MaxAgeSeconds>300</MaxAgeSeconds></CORSRule>",
+    "<CORSRule><AllowedMethod>HEAD</AllowedMethod><AllowedOrigin>*</AllowedOrigin></CORSRule></CORSConfiguration>",
+);
+
+const CORS_BUCKET: &str = "gateway-stack-cors";
+const CORS_GRANTED_BUCKET: &str = "gateway-stack-cors-granted";
+/// A name RustFS creates and the AWS bucket grammar refuses (reserved suffix).
+const CORS_RUSTFS_ONLY_BUCKET: &str = "gateway-stack-cors-s3alias";
+const CORS_WRITER: &str = "gatewaystackcorswriter";
+
+async fn stored_cors(bucket: &str) -> String {
+    let metadata = metadata_sys::get(bucket).await.expect("bucket metadata");
+    String::from_utf8_lossy(&metadata.cors_config_xml).into_owned()
+}
+
+/// PutBucketCors stores the persisted writer's bytes for a document, also on a bucket name only
+/// RustFS admits; DeleteBucketCors answers 204 and removes them, whether or not a document exists.
+/// Refused: a caller whose policy names another bucket, and the documents the codecs refuse.
+async fn the_cors_document_is_stored_and_removed(server: RustFSServer) {
+    for bucket in [CORS_BUCKET, CORS_GRANTED_BUCKET, CORS_RUSTFS_ONLY_BUCKET] {
+        create_bucket_in_the_store(bucket).await;
+    }
+    create_user_allowed_on_one_bucket(CORS_WRITER, "s3:PutBucketCors", CORS_GRANTED_BUCKET).await;
+    let endpoint = server.endpoint();
+    let root = s3_client(&endpoint, server.access_key(), server.secret_key());
+    let writer = s3_client(&endpoint, CORS_WRITER, USER_SECRET_KEY);
+    let refused = |result: Outcome| result.map_err(|refusal| (refusal.code, refusal.status));
+
+    let wrong_bucket = outcome(put_cors(&writer, CORS_BUCKET, cors_document()).send().await);
+    assert_eq!(refused(wrong_bucket), Err((Some("AccessDenied".to_owned()), Some(403))));
+    assert_eq!(stored_cors(CORS_BUCKET).await, "", "a refused write stores nothing");
+    let granted = outcome(put_cors(&writer, CORS_GRANTED_BUCKET, cors_document()).send().await);
+    assert_eq!(refused(granted), Ok(()));
+
+    let unknown_method = outcome(put_cors(&root, CORS_BUCKET, single_rule("PATCH", "*")).send().await);
+    let two_wildcards = outcome(
+        put_cors(&root, CORS_BUCKET, single_rule("GET", "https://*.*.example"))
+            .send()
+            .await,
+    );
+    let without_claim = WithoutIntegrityClaim::default();
+    let no_integrity_claim = outcome(
+        put_cors(&root, CORS_BUCKET, cors_document())
+            .customize()
+            .interceptor(without_claim.clone())
+            .send()
+            .await,
+    );
+    assert_eq!(without_claim.sent_claims(), Some(vec![]), "the request left without an integrity claim");
+    let observed = [unknown_method, two_wildcards, no_integrity_claim].map(refused);
+    assert_eq!(observed, [Ok(()), Ok(()), Ok(())], "unknown method, two wildcards, no integrity claim");
+
+    for bucket in [CORS_BUCKET, CORS_RUSTFS_ONLY_BUCKET] {
+        assert_eq!(
+            refused(outcome(put_cors(&root, bucket, cors_document()).send().await)),
+            Ok(()),
+            "{bucket}"
+        );
+        assert_eq!(stored_cors(bucket).await, PERSISTED_CORS, "{bucket}");
+    }
+
+    for attempt in ["configured", "unconfigured"] {
+        let status = StatusCapture::default();
+        let deleted = root
+            .delete_bucket_cors()
+            .bucket(CORS_BUCKET)
+            .customize()
+            .interceptor(status.clone())
+            .send()
+            .await;
+        assert_eq!((refused(outcome(deleted)), status.take()), (Ok(()), Some(204)), "{attempt}");
+        assert_eq!(stored_cors(CORS_BUCKET).await, "", "{attempt}");
+    }
+
+    server.shutdown().await;
+}
+
+#[test]
+fn the_gateway_stack_stores_and_removes_the_cors_document() {
+    with_stack(GATEWAY, the_cors_document_is_stored_and_removed);
+}
+
+#[test]
+fn the_legacy_stack_stores_and_removes_the_cors_document() {
+    with_stack(LEGACY, the_cors_document_is_stored_and_removed);
+}

rustfs/tests/embedded_gateway_stack_test/main.rs

@@ -235,6 +235,21 @@ async fn create_user_without_a_policy(access_key: &str) {
         .expect("create user without a policy");
 }
 
+/// Creates `access_key` with the user secret and attaches a policy granting `action` on `bucket`
+/// alone, so a call on any other bucket is a wrong-resource refusal.
+async fn create_user_allowed_on_one_bucket(access_key: &str, action: &str, bucket: &str) {
+    create_user_without_a_policy(access_key).await;
+    let document = format!(
+        r#"{{"Version":"2012-10-17","Statement":[{{"Effect":"Allow","Action":["{action}"],"Resource":["arn:aws:s3:::{bucket}"]}}]}}"#
+    );
+    let policy = rustfs_policy::policy::Policy::parse_config(document.as_bytes()).expect("parse policy");
+    let iam = get_global_app_context().expect("embedded server context").iam().handle();
+    iam.set_policy(access_key, policy).await.expect("store policy");
+    iam.policy_db_set(access_key, rustfs_iam::store::UserType::Reg, false, access_key)
+        .await
+        .expect("attach policy");
+}
+
 #[test]
 fn every_served_operation_has_one_matrix_row_and_every_row_a_served_operation() {
     let rows: Vec<&str> = matrix().map(|row| row.operation).collect();