fix: allow root to bypass bucket policy deny for policy management APIs (#2102)

Co-authored-by: GatewayJ <8352692332qq.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: loverustfs <hello@rustfs.com>
Co-authored-by: 安正超 <anzhengchao@gmail.com>

crates/ecstore/src/bucket/metadata.rs

@@ -372,10 +372,17 @@ impl BucketMetadata {
         Ok(())
     }
 
-    fn parse_all_configs(&mut self, _api: Arc<ECStore>) -> Result<()> {
+    fn parse_policy_config(&mut self) -> Result<()> {
         if !self.policy_config_json.is_empty() {
             self.policy_config = Some(serde_json::from_slice(&self.policy_config_json)?);
+        } else {
+            self.policy_config = None;
         }
+        Ok(())
+    }
+
+    fn parse_all_configs(&mut self, _api: Arc<ECStore>) -> Result<()> {
+        self.parse_policy_config()?;
         if !self.notification_config_xml.is_empty() {
             self.notification_config = Some(deserialize::<NotificationConfiguration>(&self.notification_config_xml)?);
         }
@@ -666,4 +673,21 @@ mod test {
         println!("   - Lifecycle config size: {} bytes", deserialized_bm.lifecycle_config_xml.len());
         println!("   - Serialized buffer size: {} bytes", buf.len());
     }
+
+    /// After policy deletion (policy_config_json cleared), parse_policy_config sets policy_config to None.
+    #[test]
+    fn test_parse_policy_config_clears_cache_when_json_empty() {
+        let policy_json = r#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3:GetObject","Resource":"arn:aws:s3:::b/*"}]}"#;
+        let mut bm = BucketMetadata::new("b");
+        bm.policy_config_json = policy_json.as_bytes().to_vec();
+        bm.parse_policy_config().unwrap();
+        assert!(bm.policy_config.is_some(), "policy_config should be set when JSON non-empty");
+
+        bm.policy_config_json.clear();
+        bm.parse_policy_config().unwrap();
+        assert!(
+            bm.policy_config.is_none(),
+            "policy_config should be None after JSON cleared (e.g. policy deleted)"
+        );
+    }
 }

rustfs/src/storage/access.rs

@@ -93,6 +93,19 @@ impl FS {
     }
 }
 
+/// Returns true when the owner (root or parent=root credentials) may bypass bucket policy
+/// explicit Deny for this action. Per AWS S3, only GetBucketPolicy, PutBucketPolicy, and
+/// DeleteBucketPolicy have this bypass so the admin can recover from a misconfigured policy.
+pub(crate) fn owner_can_bypass_policy_deny(is_owner: bool, action: &Action) -> bool {
+    is_owner
+        && matches!(
+            action,
+            Action::S3Action(S3Action::GetBucketPolicyAction)
+                | Action::S3Action(S3Action::PutBucketPolicyAction)
+                | Action::S3Action(S3Action::DeleteBucketPolicyAction)
+        )
+}
+
 /// Authorizes the request based on the action and credentials.
 pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3Result<()> {
     let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
@@ -129,7 +142,13 @@ pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3R
         }
         let bucket_name = req_info.bucket.as_deref().unwrap_or("");
 
+        // Per AWS S3: root can always perform GetBucketPolicy, PutBucketPolicy, DeleteBucketPolicy
+        // even if bucket policy explicitly denies. Other actions (ListBucket, GetObject, etc.) are
+        // subject to bucket policy Deny for root as well. See: repost.aws/knowledge-center/s3-accidentally-denied-access
+        // Here "owner" means root or credentials whose parent_user is root (e.g. Console admin via STS).
+        let owner_can_bypass_deny = owner_can_bypass_policy_deny(req_info.is_owner, &action);
         if !bucket_name.is_empty()
+            && !owner_can_bypass_deny
             && !PolicySys::is_allowed(&BucketPolicyArgs {
                 bucket: bucket_name,
                 action,
@@ -1475,4 +1494,24 @@ mod tests {
         let result = bucket_policy_uses_existing_object_tag("test-bucket-no-policy-xyz-absent").await;
         assert!(!result, "bucket with no policy should not use ExistingObjectTag");
     }
+
+    /// Owner can bypass bucket policy Deny only for the three policy management APIs (per AWS S3).
+    #[test]
+    fn test_owner_can_bypass_policy_deny_only_for_policy_apis() {
+        // Owner + policy management actions -> bypass allowed
+        assert!(owner_can_bypass_policy_deny(true, &Action::S3Action(S3Action::GetBucketPolicyAction)));
+        assert!(owner_can_bypass_policy_deny(true, &Action::S3Action(S3Action::PutBucketPolicyAction)));
+        assert!(owner_can_bypass_policy_deny(true, &Action::S3Action(S3Action::DeleteBucketPolicyAction)));
+
+        // Owner + other actions -> no bypass (still subject to bucket policy Deny)
+        assert!(!owner_can_bypass_policy_deny(true, &Action::S3Action(S3Action::ListBucketAction)));
+        assert!(!owner_can_bypass_policy_deny(true, &Action::S3Action(S3Action::GetObjectAction)));
+
+        // Non-owner -> no bypass for any action
+        assert!(!owner_can_bypass_policy_deny(false, &Action::S3Action(S3Action::GetBucketPolicyAction)));
+        assert!(!owner_can_bypass_policy_deny(
+            false,
+            &Action::S3Action(S3Action::DeleteBucketPolicyAction)
+        ));
+    }
 }