refactor(storage): authorize bucket configuration changes on both edges

`authorize_bucket_config_mutation(server_ctx, req, bucket, action)` is
now generic over `AuthorizationRequest`, so the legacy hooks and the
gateway edge run the same shape. The shape records the bucket on
`ReqInfo` and snapshots the bucket incarnation under the lifecycle
namespace read lock before the decision. It surfaces a snapshot failure
only after an Allow, and on Allow it installs the snapshot (new trait
installer `install_bucket_config_mutation`) for
`bucket_config_mutation_incarnation`.

The 22 legacy hooks of the shape now call it, with no other change:

- Delete: BucketCors, BucketEncryption, BucketLifecycle, BucketPolicy,
  BucketReplication, BucketTagging, BucketWebsite and
  PublicAccessBlock.
- Put: BucketAccelerateConfiguration, BucketCors, BucketEncryption,
  BucketLifecycleConfiguration, BucketLogging,
  BucketNotificationConfiguration, BucketPolicy, BucketReplication,
  BucketRequestPayment, BucketTagging, BucketVersioning, BucketWebsite,
  ObjectLockConfiguration and PublicAccessBlock.

DeleteBucketTagging keeps its `with_bucket_tag_deletion` scope at the
call site.

Tests: the decision fixture of the owned-versus-legacy parity tests is
now one shared function. A new parity test runs eight cases through
both request forms and compares the decision and the snapshot handed
on. Three are allowed: IAM user, anonymous via bucket policy, owner.
Five are refused: wrong resource, no policy for the action, anonymous
on a private bucket, anonymous on a missing bucket, owner on a missing
bucket.

Not built here: the gateway edges of these operations (batches A1-A4).

rustfs/src/app/context/envelope.rs

@@ -507,6 +507,9 @@ impl AuthorizationRequest for RequestEnvelope {
     fn bucket_config_mutation(&self) -> Option<&BucketConfigMutationSnapshot> {
         self.bucket_config_mutation.as_ref()
     }
+    fn install_bucket_config_mutation(&mut self, snapshot: BucketConfigMutationSnapshot) {
+        self.bucket_config_mutation = Some(snapshot);
+    }
     #[cfg(test)]
     fn test_extensions(&self) -> &Extensions {
         &self.residual_extensions

rustfs/src/app/owned_authorization_test.rs

@@ -1172,9 +1172,19 @@ fn peer(req: &mut S3Request<()>, addr: [u8; 4]) {
         .insert(Some(crate::shared_types::RemoteAddr(std::net::SocketAddr::from((addr, 40000)))));
 }
 
-#[tokio::test]
-#[serial]
-async fn owned_and_legacy_entry_points_reach_the_same_decisions() {
+const PRIVATE: &str = "parity-private";
+const PUBLIC: &str = "parity-public";
+
+/// What the decision parity tests share: two buckets on an isolated store, `PUBLIC` with a bucket
+/// policy for anonymous callers, an IAM user whose policy covers `PRIVATE`, the owner, and a
+/// request slot bound to that store. The first element keeps the store's directories alive.
+async fn decision_world() -> (
+    impl Sized,
+    Arc<crate::app::storage_api::test::ECStore>,
+    Arc<ServerContextSlot>,
+    rustfs_credentials::Credentials,
+    rustfs_credentials::Credentials,
+) {
     use rustfs_iam::store::{Store as _, UserType};
     use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq};
     use rustfs_policy::policy::Policy;
@@ -1190,9 +1200,7 @@ async fn owned_and_legacy_entry_points_reach_the_same_decisions() {
         }
     }
 
-    const PRIVATE: &str = "parity-private";
-    const PUBLIC: &str = "parity-public";
-    let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    let (temp_dir, disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
     for bucket in [PRIVATE, PUBLIC] {
         store
             .make_bucket(bucket, &MakeBucketOptions::default())
@@ -1215,7 +1223,8 @@ async fn owned_and_legacy_entry_points_reach_the_same_decisions() {
         {"Effect":"Allow","Principal":{"AWS":"*"},"Action":"s3:PutObject","Resource":format!("arn:aws:s3:::{PUBLIC}/stage/*"),
             "Condition":{"Bool":{"s3:ObjectCreationOperation":"false"}}},
         {"Effect":"Allow","Principal":{"AWS":"*"},"Action":"s3:PutObject","Resource":format!("arn:aws:s3:::{PUBLIC}/post/*"),
-            "Condition":{"StringEquals":{"s3:RequestObjectTag/class":"public"}}}
+            "Condition":{"StringEquals":{"s3:RequestObjectTag/class":"public"}}},
+        {"Effect":"Allow","Principal":{"AWS":"*"},"Action":"s3:PutBucketCors","Resource":format!("arn:aws:s3:::{PUBLIC}")}
     ]});
     store
         .update_bucket_metadata_config(
@@ -1251,7 +1260,8 @@ async fn owned_and_legacy_entry_points_reach_the_same_decisions() {
             format!(
                 r#"{{"Version":"2012-10-17","Statement":[
                     {{"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::{PRIVATE}/*"}},
-                    {{"Effect":"Deny","Action":"s3:GetObject","Resource":"arn:aws:s3:::{PRIVATE}/iam-denied"}}
+                    {{"Effect":"Deny","Action":"s3:GetObject","Resource":"arn:aws:s3:::{PRIVATE}/iam-denied"}},
+                    {{"Effect":"Allow","Action":"s3:PutBucketTagging","Resource":"arn:aws:s3:::{PRIVATE}"}}
                 ]}}"#
             )
             .as_bytes(),
@@ -1272,6 +1282,13 @@ async fn owned_and_legacy_entry_points_reach_the_same_decisions() {
     let context = Arc::new(AppContext::new(Arc::clone(&store), Arc::new(ReadyIam(iam)), Arc::new(TestKms)));
     let slot = ServerContextSlot::new();
     assert!(slot.install(context), "install request context");
+    ((temp_dir, disk_paths), store, slot, user, owner)
+}
+
+#[tokio::test]
+#[serial]
+async fn owned_and_legacy_entry_points_reach_the_same_decisions() {
+    let (_store_dirs, _store, slot, user, owner) = decision_world().await;
 
     let cases = [
         DecisionCase {
@@ -1594,3 +1611,134 @@ fn post_form(req: &mut S3Request<()>, form_tagging: &'static str, header_tagging
     req.extensions.insert(PostObjectRequestMarker);
     req.extensions.insert(PostObjectPolicyHeaders(form));
 }
+
+/// What a bucket-configuration mutation answered, compared across the two request forms.
+#[derive(Debug, PartialEq)]
+enum MutationOutcome {
+    /// Allowed, with the incarnation snapshot the use case reads back.
+    Allowed(Option<uuid::Uuid>),
+    Denied,
+    /// Refused with another code after the decision, such as a snapshot failure.
+    Failed(String),
+}
+
+#[tokio::test]
+#[serial]
+async fn owned_and_legacy_bucket_config_mutations_hand_on_the_same_decision_and_snapshot() {
+    use crate::app::storage_api::owned_parity::{authorize_bucket_config_mutation, bucket_config_mutation_incarnation};
+
+    const MISSING: &str = "parity-missing";
+    let (_store_dirs, store, slot, user, owner) = decision_world().await;
+    let private = Some(store.bucket_incarnation_id(PRIVATE).await.expect("private incarnation"));
+    let public = Some(store.bucket_incarnation_id(PUBLIC).await.expect("public incarnation"));
+    let failed = |code: &str| MutationOutcome::Failed(code.to_owned());
+    let cases = [
+        (
+            "IAM policy allows the user",
+            PRIVATE,
+            Principal::User,
+            S3Action::PutBucketTaggingAction,
+            MutationOutcome::Allowed(private),
+        ),
+        (
+            "bucket policy allows anonymous",
+            PUBLIC,
+            Principal::Anonymous,
+            S3Action::PutBucketCorsAction,
+            MutationOutcome::Allowed(public),
+        ),
+        (
+            "owner",
+            PRIVATE,
+            Principal::Owner,
+            S3Action::PutBucketVersioningAction,
+            MutationOutcome::Allowed(private),
+        ),
+        (
+            "user policy names another bucket",
+            PUBLIC,
+            Principal::User,
+            S3Action::PutBucketTaggingAction,
+            MutationOutcome::Denied,
+        ),
+        (
+            "user without a policy for the action",
+            PRIVATE,
+            Principal::User,
+            S3Action::PutBucketCorsAction,
+            MutationOutcome::Denied,
+        ),
+        (
+            "anonymous on a private bucket",
+            PRIVATE,
+            Principal::Anonymous,
+            S3Action::PutBucketTaggingAction,
+            MutationOutcome::Denied,
+        ),
+        (
+            "anonymous on a missing bucket",
+            MISSING,
+            Principal::Anonymous,
+            S3Action::PutBucketCorsAction,
+            MutationOutcome::Denied,
+        ),
+        (
+            "owner on a missing bucket",
+            MISSING,
+            Principal::Owner,
+            S3Action::PutBucketVersioningAction,
+            failed("NoSuchBucket"),
+        ),
+    ];
+
+    let mut mismatches = Vec::new();
+    for (name, bucket, principal, action, expected) in cases {
+        let mut legacy = build_request((), Method::PUT);
+        legacy.uri = format!("/{bucket}").parse().expect("case uri");
+        legacy.extensions.insert(Arc::clone(&slot));
+        let (cred, is_owner) = match principal {
+            Principal::Anonymous => (None, false),
+            Principal::User => (Some(user.clone()), false),
+            Principal::Owner => (Some(owner.clone()), true),
+        };
+        legacy.extensions.insert(ReqInfo {
+            cred,
+            is_owner,
+            ..Default::default()
+        });
+        let mut owned = RequestEnvelope::from_s3s(&legacy);
+        let action = Action::S3Action(action);
+
+        let legacy_result = authorize_bucket_config_mutation(&slot, &mut legacy, bucket, action).await;
+        let owned_result = authorize_bucket_config_mutation(&slot, &mut owned, bucket, action).await;
+        let outcome = |result: Result<(), S3ErrorCode>, snapshot: Result<Option<uuid::Uuid>, S3ErrorCode>| match result {
+            Ok(()) => MutationOutcome::Allowed(snapshot.ok().flatten()),
+            Err(S3ErrorCode::AccessDenied) => MutationOutcome::Denied,
+            Err(code) => MutationOutcome::Failed(code.as_str().to_owned()),
+        };
+        let legacy_outcome = outcome(
+            legacy_result.map_err(|err| err.code().clone()),
+            bucket_config_mutation_incarnation(&legacy, bucket).map_err(|err| err.code().clone()),
+        );
+        let owned_outcome = outcome(
+            owned_result.map_err(|err| err.code().clone()),
+            bucket_config_mutation_incarnation(&owned, bucket).map_err(|err| err.code().clone()),
+        );
+        // A refused request hands no snapshot on: the use case's reader refuses to run without one.
+        let refused_without_snapshot = |snapshot: Result<Option<uuid::Uuid>, S3ErrorCode>| {
+            matches!(expected, MutationOutcome::Allowed(_)) || snapshot == Err(S3ErrorCode::InternalError)
+        };
+        let both_refused_without_snapshot =
+            refused_without_snapshot(bucket_config_mutation_incarnation(&legacy, bucket).map_err(|err| err.code().clone()))
+                && refused_without_snapshot(bucket_config_mutation_incarnation(&owned, bucket).map_err(|err| err.code().clone()));
+        let bucket_recorded = [legacy.extensions.get::<ReqInfo>(), owned.req_info()]
+            .iter()
+            .all(|info| info.and_then(|info| info.bucket.as_deref()) == Some(bucket));
+        if legacy_outcome != expected || owned_outcome != expected || !both_refused_without_snapshot || !bucket_recorded {
+            mismatches.push(format!(
+                "{name}: expected {expected:?}, legacy {legacy_outcome:?}, owned {owned_outcome:?}"
+            ));
+        }
+    }
+    assert!(mismatches.is_empty(), "bucket config mutation mismatches:\n{}", mismatches.join("\n"));
+}

rustfs/src/app/storage_api.rs

@@ -1848,8 +1848,8 @@ pub(crate) mod owned_parity {
         DenialAuditor, DenialRateLimiter, DenialScope, FS, ObjectTagConditions, PostObjectPolicyHeaders, PostObjectRequestMarker,
         ReqInfo, RequestContext, RequestedCopySource, RequestedObjectLegalHold, RequestedObjectLockRetainUntil,
         RequestedObjectLockRetention, RequestedObjectTags, SUPPRESSED_DENIALS_TAG, TableDataPlanePublicationGuards,
-        authorization_denial_entry, authorization_denial_entry_for_view, authorize_object_request, authorize_request,
-        parity_test_support,
+        authorization_denial_entry, authorization_denial_entry_for_view, authorize_bucket_config_mutation,
+        authorize_object_request, authorize_request, bucket_config_mutation_incarnation, parity_test_support,
     };
 }
 

rustfs/src/storage/access.rs

@@ -455,13 +455,12 @@ pub(crate) fn bucket_config_mutation_incarnation<R: AuthorizationRequest + ?Size
     Ok(Some(snapshot.incarnation_id))
 }
 
-async fn load_bucket_config_mutation_snapshot<T>(
-    fs: &FS,
-    req: &S3Request<T>,
+async fn load_bucket_config_mutation_snapshot<R: AuthorizationRequest + ?Sized>(
+    server_ctx: &ServerContextSlot,
+    req: &R,
     bucket: &str,
 ) -> S3Result<BucketConfigMutationSnapshot> {
-    let store = fs
-        .server_ctx()
+    let store = server_ctx
         .object_store()
         .ok_or_else(|| s3_error!(InternalError, "object store is not initialized"))?;
     let lock = store
@@ -489,14 +488,28 @@ async fn load_bucket_config_mutation_snapshot<T>(
     })
 }
 
-async fn authorize_bucket_config_mutation<T>(fs: &FS, req: &mut S3Request<T>, action: Action) -> S3Result<()> {
-    let bucket = req_info_ref(req)?
-        .bucket
-        .clone()
-        .ok_or_else(|| s3_error!(InternalError, "bucket config mutation request has no bucket"))?;
-    let snapshot = load_bucket_config_mutation_snapshot(fs, req, &bucket).await;
+/// Authorizes a bucket-configuration mutation for either edge: `action` on `bucket`, with the
+/// bucket incarnation snapshotted under the lifecycle namespace read lock BEFORE the decision, so
+/// a bucket recreated during authorization is refused by the use case's conditional write. A
+/// snapshot failure surfaces only after an Allow, which keeps AccessDenied first. On Allow the
+/// snapshot is installed for [`bucket_config_mutation_incarnation`].
+///
+/// The legacy hooks pass `FS::server_ctx`; the gateway edge passes the envelope's server context.
+///
+/// # Errors
+///
+/// `InternalError` when the access phase installed no `ReqInfo`; what [`authorize_request`]
+/// refuses; then the snapshot's lock or store error.
+pub(crate) async fn authorize_bucket_config_mutation<R: AuthorizationRequest + ?Sized>(
+    server_ctx: &ServerContextSlot,
+    req: &mut R,
+    bucket: &str,
+    action: Action,
+) -> S3Result<()> {
+    req_info_mut(req)?.bucket = Some(bucket.to_owned());
+    let snapshot = load_bucket_config_mutation_snapshot(server_ctx, req, bucket).await;
     authorize_request(req, action).await?;
-    req.extensions.insert(snapshot?);
+    req.install_bucket_config_mutation(snapshot?);
     Ok(())
 }
 
@@ -1021,6 +1034,7 @@ pub(crate) trait AuthorizationRequest {
     fn odm_read_generation(&self) -> Option<&OdmReadGenerationGuard>;
     fn install_odm_read_generation(&mut self, guard: OdmReadGenerationGuard);
     fn bucket_config_mutation(&self) -> Option<&BucketConfigMutationSnapshot>;
+    fn install_bucket_config_mutation(&mut self, snapshot: BucketConfigMutationSnapshot);
     /// The extensions no typed accessor covers, for test-only payloads.
     #[cfg(test)]
     fn test_extensions(&self) -> &http::Extensions;
@@ -1096,6 +1110,9 @@ impl<T> AuthorizationRequest for S3Request<T> {
     fn bucket_config_mutation(&self) -> Option<&BucketConfigMutationSnapshot> {
         self.extensions.get::<BucketConfigMutationSnapshot>()
     }
+    fn install_bucket_config_mutation(&mut self, snapshot: BucketConfigMutationSnapshot) {
+        self.extensions.insert(snapshot);
+    }
     #[cfg(test)]
     fn test_extensions(&self) -> &http::Extensions {
         &self.extensions
@@ -3224,20 +3241,18 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn delete_bucket_cors(&self, req: &mut S3Request<DeleteBucketCorsInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::DeleteBucketCorsAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::DeleteBucketCorsAction))
+            .await
     }
 
     /// Checks whether the DeleteBucketEncryption request has accesses to the resources.
     ///
     /// This method returns `Ok(())` by default.
     async fn delete_bucket_encryption(&self, req: &mut S3Request<DeleteBucketEncryptionInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketEncryptionAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::PutBucketEncryptionAction))
+            .await
     }
 
     /// Checks whether the DeleteBucketIntelligentTieringConfiguration request has accesses to the resources.
@@ -3264,10 +3279,9 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn delete_bucket_lifecycle(&self, req: &mut S3Request<DeleteBucketLifecycleInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketLifecycleAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::PutBucketLifecycleAction))
+            .await
     }
 
     /// Checks whether the DeleteBucketMetricsConfiguration request has accesses to the resources.
@@ -3291,32 +3305,34 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn delete_bucket_policy(&self, req: &mut S3Request<DeleteBucketPolicyInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::DeleteBucketPolicyAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::DeleteBucketPolicyAction))
+            .await
     }
 
     /// Checks whether the DeleteBucketReplication request has accesses to the resources.
     ///
     /// This method returns `Ok(())` by default.
     async fn delete_bucket_replication(&self, req: &mut S3Request<DeleteBucketReplicationInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutReplicationConfigurationAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(
+            self.server_ctx(),
+            req,
+            &bucket,
+            Action::S3Action(S3Action::PutReplicationConfigurationAction),
+        )
+        .await
     }
 
     /// Checks whether the DeleteBucketTagging request has accesses to the resources.
     ///
     /// This method returns `Ok(())` by default.
     async fn delete_bucket_tagging(&self, req: &mut S3Request<DeleteBucketTaggingInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
+        let bucket = req.input.bucket.clone();
         rustfs_iam::sys::with_bucket_tag_deletion(authorize_bucket_config_mutation(
-            self,
+            self.server_ctx(),
             req,
+            &bucket,
             Action::S3Action(S3Action::PutBucketTaggingAction),
         ))
         .await
@@ -3326,10 +3342,8 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn delete_bucket_website(&self, req: &mut S3Request<DeleteBucketWebsiteInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, bucket_website_config_authorize_action()).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, bucket_website_config_authorize_action()).await
     }
 
     /// Checks whether the DeleteObject request has accesses to the resources.
@@ -3379,10 +3393,14 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn delete_public_access_block(&self, req: &mut S3Request<DeletePublicAccessBlockInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::DeleteBucketPublicAccessBlockAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(
+            self.server_ctx(),
+            req,
+            &bucket,
+            Action::S3Action(S3Action::DeleteBucketPublicAccessBlockAction),
+        )
+        .await
     }
 
     /// Checks whether the GetBucketAccelerateConfiguration request has accesses to the resources.
@@ -3872,10 +3890,8 @@ impl S3Access for FS {
         &self,
         req: &mut S3Request<PutBucketAccelerateConfigurationInput>,
     ) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketPolicyAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::PutBucketPolicyAction)).await
     }
 
     /// Checks whether the PutBucketAcl request has accesses to the resources.
@@ -3902,20 +3918,17 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn put_bucket_cors(&self, req: &mut S3Request<PutBucketCorsInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketCorsAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::PutBucketCorsAction)).await
     }
 
     /// Checks whether the PutBucketEncryption request has accesses to the resources.
     ///
     /// This method returns `Ok(())` by default.
     async fn put_bucket_encryption(&self, req: &mut S3Request<PutBucketEncryptionInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketEncryptionAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::PutBucketEncryptionAction))
+            .await
     }
 
     /// Checks whether the PutBucketIntelligentTieringConfiguration request has accesses to the resources.
@@ -3945,17 +3958,15 @@ impl S3Access for FS {
         &self,
         req: &mut S3Request<PutBucketLifecycleConfigurationInput>,
     ) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketLifecycleAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::PutBucketLifecycleAction))
+            .await
     }
 
     async fn put_bucket_logging(&self, req: &mut S3Request<PutBucketLoggingInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketLoggingAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::PutBucketLoggingAction))
+            .await
     }
 
     /// Checks whether the PutBucketMetricsConfiguration request has accesses to the resources.
@@ -3972,10 +3983,9 @@ impl S3Access for FS {
         &self,
         req: &mut S3Request<PutBucketNotificationConfigurationInput>,
     ) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketNotificationAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::PutBucketNotificationAction))
+            .await
     }
 
     /// Checks whether the PutBucketOwnershipControls request has accesses to the resources.
@@ -3989,60 +3999,56 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn put_bucket_policy(&self, req: &mut S3Request<PutBucketPolicyInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, put_bucket_policy_authorize_action()).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, put_bucket_policy_authorize_action()).await
     }
 
     /// Checks whether the PutBucketReplication request has accesses to the resources.
     ///
     /// This method returns `Ok(())` by default.
     async fn put_bucket_replication(&self, req: &mut S3Request<PutBucketReplicationInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutReplicationConfigurationAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(
+            self.server_ctx(),
+            req,
+            &bucket,
+            Action::S3Action(S3Action::PutReplicationConfigurationAction),
+        )
+        .await
     }
 
     /// Checks whether the PutBucketRequestPayment request has accesses to the resources.
     ///
     /// This method returns `Ok(())` by default.
     async fn put_bucket_request_payment(&self, req: &mut S3Request<PutBucketRequestPaymentInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketPolicyAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::PutBucketPolicyAction)).await
     }
 
     /// Checks whether the PutBucketTagging request has accesses to the resources.
     ///
     /// This method returns `Ok(())` by default.
     async fn put_bucket_tagging(&self, req: &mut S3Request<PutBucketTaggingInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketTaggingAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::PutBucketTaggingAction))
+            .await
     }
 
     /// Checks whether the PutBucketVersioning request has accesses to the resources.
     ///
     /// This method returns `Ok(())` by default.
     async fn put_bucket_versioning(&self, req: &mut S3Request<PutBucketVersioningInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketVersioningAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, Action::S3Action(S3Action::PutBucketVersioningAction))
+            .await
     }
 
     /// Checks whether the PutBucketWebsite request has accesses to the resources.
     ///
     /// This method returns `Ok(())` by default.
     async fn put_bucket_website(&self, req: &mut S3Request<PutBucketWebsiteInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, bucket_website_config_authorize_action()).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(self.server_ctx(), req, &bucket, bucket_website_config_authorize_action()).await
     }
 
     /// Checks whether the PutObject request has accesses to the resources.
@@ -4128,10 +4134,14 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn put_object_lock_configuration(&self, req: &mut S3Request<PutObjectLockConfigurationInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketObjectLockConfigurationAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(
+            self.server_ctx(),
+            req,
+            &bucket,
+            Action::S3Action(S3Action::PutBucketObjectLockConfigurationAction),
+        )
+        .await
     }
 
     /// Checks whether the PutObjectRetention request has accesses to the resources.
@@ -4176,10 +4186,14 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn put_public_access_block(&self, req: &mut S3Request<PutPublicAccessBlockInput>) -> S3Result<()> {
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-
-        authorize_bucket_config_mutation(self, req, Action::S3Action(S3Action::PutBucketPublicAccessBlockAction)).await
+        let bucket = req.input.bucket.clone();
+        authorize_bucket_config_mutation(
+            self.server_ctx(),
+            req,
+            &bucket,
+            Action::S3Action(S3Action::PutBucketPublicAccessBlockAction),
+        )
+        .await
     }
 
     /// Checks whether the RestoreObject request has accesses to the resources.

rustfs/src/storage/storage_api.rs

@@ -175,7 +175,8 @@ pub(crate) mod owned_parity_consumer {
     pub(crate) use super::super::access::{
         ObjectTagConditions, PostObjectPolicyHeaders, PostObjectRequestMarker, ReqInfo, RequestedCopySource,
         RequestedObjectLegalHold, RequestedObjectLockRetainUntil, RequestedObjectLockRetention, RequestedObjectTags,
-        TableDataPlanePublicationGuards, authorize_object_request, authorize_request, parity_test_support,
+        TableDataPlanePublicationGuards, authorize_bucket_config_mutation, authorize_object_request, authorize_request,
+        bucket_config_mutation_incarnation, parity_test_support,
     };
     pub(crate) use super::super::denial_audit::{DenialAuditor, DenialRateLimiter, SUPPRESSED_DENIALS_TAG};
     pub(crate) use super::super::ecfs::FS;