fix(iam): require an explicit permission for force-delete (#8154)

* fix(iam): require an explicit permission for force-delete A force-delete header no longer inherits s3:* or consoleAdmin. Bucket force-delete requires s3:ForceDeleteBucket whenever the header is present, and recursive object force-delete requires s3:ForceDeleteObject. A plain delete keeps the existing checks. Co-authored-by: RustFS <hello@rustfs.com> Signed-off-by: loverustfs <155562731+loverustfs@users.noreply.github.com> * test(e2e): keep force-delete header names static The bucket force-delete helper must pass a static header name into the SDK request mutator. Co-authored-by: RustFS <hello@rustfs.com> Signed-off-by: loverustfs <155562731+loverustfs@users.noreply.github.com> * test(e2e): move the force-delete header into the request mutator The SDK request customizer requires a static header name owned by the closure. Co-authored-by: RustFS <hello@rustfs.com> Signed-off-by: loverustfs <155562731+loverustfs@users.noreply.github.com> * fix(iam): keep force-delete out of NotAction grants NotAction now uses plain wildcard matching, so NotAction "s3:*" still excludes force-delete. An Allow statement grants s3:ForceDeleteObject or s3:ForceDeleteBucket only when its Action list names the action; a NotAction-only Allow never does. The rule applies to both IAM and bucket policy statements. Also build the invalid-header errors with S3Error::with_message to keep the s3s footprint at its baseline, and fix a clippy single_match. --------- Signed-off-by: loverustfs <155562731+loverustfs@users.noreply.github.com> Co-authored-by: Hauser <housemecn@gmail.com> Co-authored-by: overtrue <anzhengchao@gmail.com>
2026-10-02 05:14:35 +08:00 · 2026-09-28 19:08:33 +08:00
parent fc5609bbb0
commit 9442e89f5f
10 changed files with 427 additions and 44 deletions
@@ -282,7 +282,7 @@ async fn sdk_list_bucket_and_list_bucket_versions_permissions_are_independent()
 }

 #[tokio::test]
-async fn console_admin_force_delete_removes_prefix_versions_and_delete_markers() -> TestResult {
+async fn explicit_force_delete_permission_removes_prefix_versions_and_delete_markers() -> TestResult {
    init_logging();
    let mut env = RustFSTestEnvironment::new().await?;
    env.start_rustfs_server(vec![]).await?;
@@ -300,7 +300,14 @@ async fn console_admin_force_delete_removes_prefix_versions_and_delete_markers()
    put(&root, bucket, "folder-sibling/keep.txt").await?;
    let keep = versions(&root, bucket, "keep.txt").await?;
    let sibling = versions(&root, bucket, "folder-sibling/").await?;
-    let user = policy_user(&env, "consoleAdmin", None).await?;
+    let user = policy_user(
+        &env,
+        "explicit-force-delete",
+        Some(json!({"Version":"2012-10-17","Statement":[
+            {"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{bucket}/*")}
+        ]})),
+    )
+    .await?;

    force_delete(&user, bucket, "folder/").await?;
    assert!(
@@ -322,6 +329,126 @@ async fn console_admin_force_delete_removes_prefix_versions_and_delete_markers()
    Ok(())
 }

+#[tokio::test]
+async fn force_delete_header_requires_explicit_permission() -> TestResult {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    let root = env.create_s3_client();
+    let admin = policy_user(&env, "consoleAdmin", None).await?;
+
+    let owner_bucket = "force-owner-bucket";
+    root.create_bucket().bucket(owner_bucket).send().await?;
+    put(&root, owner_bucket, "kept-until-force.txt").await?;
+    force_delete_bucket(&root, owner_bucket, "x-rustfs-force-delete").await?;
+    assert!(
+        head_bucket_missing(&root, owner_bucket).await?,
+        "root force-delete must remove a non-empty bucket"
+    );
+
+    let denied_bucket = "force-admin-denied";
+    root.create_bucket().bucket(denied_bucket).send().await?;
+    put(&root, denied_bucket, "folder/child.txt").await?;
+    put(&root, denied_bucket, "plain.txt").await?;
+    assert_boxed_denied(force_delete(&admin, denied_bucket, "folder/").await);
+    assert!(
+        !versions(&root, denied_bucket, "folder/").await?.is_empty(),
+        "consoleAdmin without s3:ForceDeleteObject must not recursively delete"
+    );
+    assert_boxed_denied(force_delete_bucket(&admin, denied_bucket, "x-rustfs-force-delete").await);
+    assert_boxed_denied(force_delete_bucket(&admin, denied_bucket, "x-minio-force-delete").await);
+    assert!(
+        !head_bucket_missing(&root, denied_bucket).await?,
+        "consoleAdmin without s3:ForceDeleteBucket must not force-delete a non-empty bucket"
+    );
+    let not_empty = admin.delete_bucket().bucket(denied_bucket).send().await;
+    let not_empty = not_empty.expect_err("plain delete of a non-empty bucket must fail");
+    assert_eq!(
+        not_empty.as_service_error().and_then(|error| error.code()),
+        Some("BucketNotEmpty"),
+        "plain delete must keep the emptiness check, got {not_empty:?}"
+    );
+    admin.delete_object().bucket(denied_bucket).key("plain.txt").send().await?;
+    assert!(
+        versions(&root, denied_bucket, "plain.txt").await?.is_empty(),
+        "plain DeleteObject without the force header must stay allowed for consoleAdmin"
+    );
+
+    let object_bucket = "force-explicit-object";
+    root.create_bucket().bucket(object_bucket).send().await?;
+    put(&root, object_bucket, "folder/child.txt").await?;
+    put(&root, object_bucket, "keep.txt").await?;
+    let keep = versions(&root, object_bucket, "keep.txt").await?;
+    let object_user = policy_user(
+        &env,
+        "explicit-object-force",
+        Some(json!({"Version":"2012-10-17","Statement":[
+            {"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{object_bucket}/*")}
+        ]})),
+    )
+    .await?;
+    force_delete(&object_user, object_bucket, "folder/").await?;
+    assert!(versions(&root, object_bucket, "folder/").await?.is_empty());
+    assert_eq!(versions(&root, object_bucket, "keep.txt").await?, keep);
+
+    let allowed_bucket = "force-explicit-bucket";
+    root.create_bucket().bucket(allowed_bucket).send().await?;
+    put(&root, allowed_bucket, "inside.txt").await?;
+    let bucket_user = policy_user(
+        &env,
+        "explicit-bucket-force",
+        Some(json!({"Version":"2012-10-17","Statement":[
+            {"Effect":"Allow","Action":["s3:DeleteBucket","s3:ForceDeleteBucket"],"Resource":[
+                format!("arn:aws:s3:::{allowed_bucket}"), format!("arn:aws:s3:::{allowed_bucket}/*")
+            ]}
+        ]})),
+    )
+    .await?;
+    force_delete_bucket(&bucket_user, allowed_bucket, "x-minio-force-delete").await?;
+    assert!(
+        head_bucket_missing(&root, allowed_bucket).await?,
+        "an explicit s3:ForceDeleteBucket grant must remove a non-empty bucket"
+    );
+    Ok(())
+}
+
+async fn force_delete_bucket(
+    client: &Client,
+    bucket: &str,
+    header: &str,
+) -> Result<(), Box<SdkError<aws_sdk_s3::operation::delete_bucket::DeleteBucketError>>> {
+    let header: &'static str = match header {
+        "x-rustfs-force-delete" => "x-rustfs-force-delete",
+        "x-minio-force-delete" => "x-minio-force-delete",
+        other => panic!("unexpected force-delete header {other}"),
+    };
+    client
+        .delete_bucket()
+        .bucket(bucket)
+        .customize()
+        .mutate_request(move |request| {
+            request.headers_mut().insert(header, "true");
+        })
+        .send()
+        .await
+        .map(|_| ())
+        .map_err(Box::new)
+}
+
+async fn head_bucket_missing(client: &Client, bucket: &str) -> TestResult<bool> {
+    match client.head_bucket().bucket(bucket).send().await {
+        Ok(_) => Ok(false),
+        Err(error) => {
+            let code = error.as_service_error().and_then(|service| service.code());
+            if matches!(code, Some("NotFound" | "NoSuchBucket")) {
+                Ok(true)
+            } else {
+                Err(error.into())
+            }
+        }
+    }
+}
+
 #[tokio::test]
 async fn force_delete_authorizes_only_its_path_scope_without_list_permissions() -> TestResult {
    init_logging();
@@ -339,7 +466,7 @@ async fn force_delete_authorizes_only_its_path_scope_without_list_permissions()
        &env,
        "delete-only",
        Some(json!({"Version":"2012-10-17","Statement":[
-            {"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion"],"Resource":[
+            {"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject"],"Resource":[
                format!("arn:aws:s3:::{bucket}/selected.txt"), format!("arn:aws:s3:::{bucket}/selected.txt/*")
            ]},
            {"Effect":"Deny","Action":["s3:DeleteObject","s3:DeleteObjectVersion"],"Resource":format!("arn:aws:s3:::{bucket}/selected.txt-sibling")}
@@ -376,7 +503,7 @@ async fn force_directory_delete_cannot_remove_an_unauthorized_colliding_parent()
        &env,
        "parent-denier",
        Some(json!({"Version":"2012-10-17","Statement":[
-            {"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
+            {"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
            {"Effect":"Deny","Action":"s3:DeleteObjectVersion","Resource":format!("arn:aws:s3:::{bucket}/collision.txt"),
             "Condition":{"StringEquals":{"s3:VersionId":protected_parent_version}}}
        ]})),
@@ -409,7 +536,7 @@ async fn force_unversioned_directory_requires_only_delete_object() -> TestResult
        &env,
        "unversioned-deleter",
        Some(json!({"Version":"2012-10-17","Statement":[
-            {"Effect":"Allow","Action":"s3:DeleteObject","Resource":format!("arn:aws:s3:::{bucket}/*")},
+            {"Effect":"Allow","Action":["s3:DeleteObject","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
            {"Effect":"Deny","Action":"s3:DeleteObjectVersion","Resource":format!("arn:aws:s3:::{bucket}/*")}
        ]})),
    )
@@ -438,7 +565,7 @@ async fn force_delete_denied_child_preserves_every_object_despite_bucket_allow()
        &env,
        "child-denier",
        Some(json!({"Version":"2012-10-17","Statement":[
-            {"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ReplicateDelete"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
+            {"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject","s3:ReplicateDelete"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
            {"Effect":"Deny","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ReplicateDelete"],"Resource":format!("arn:aws:s3:::{bucket}/folder/z-denied.txt")}
        ]})),
    )
@@ -468,7 +595,7 @@ async fn force_delete_denied_child_preserves_every_object_despite_bucket_allow()
        &env,
        "replica-deleter",
        Some(json!({"Version":"2012-10-17","Statement":[
-            {"Effect":"Allow","Action":"s3:DeleteObject","Resource":format!("arn:aws:s3:::{bucket}/*")},
+            {"Effect":"Allow","Action":["s3:DeleteObject","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
            {"Effect":"Allow","Action":"s3:ReplicateDelete","Resource":format!("arn:aws:s3:::{bucket}/*")}
        ]})),
    )
@@ -500,7 +627,7 @@ async fn force_delete_denied_historical_version_preserves_versions_and_markers()
    put(&root, bucket, "folder/a-allowed.txt").await?;
    let policy = |version: &str| {
        json!({"Version":"2012-10-17","Statement":[
-            {"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
+            {"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
            {"Effect":"Deny","Action":"s3:DeleteObjectVersion","Resource":format!("arn:aws:s3:::{bucket}/folder/*"),
             "Condition":{"StringEquals":{"s3:VersionId":version}}}
        ]})
@@ -628,7 +755,7 @@ async fn force_delete_checks_every_version_page_before_mutation() -> TestResult
        .try_collect::<Vec<_>>()
        .await?;
    put(&root, bucket, "folder/z-denied.txt").await?;
-    let allow = json!({"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion"],"Resource":format!("arn:aws:s3:::{bucket}/*")});
+    let allow = json!({"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{bucket}/*")});
    let user = policy_user(
        &env,
        "paged-deleter",
@@ -58,8 +58,14 @@ impl ActionSet {
    }

    pub fn is_match(&self, action: &Action) -> bool {
+        self.is_match_for_effect(action, false)
+    }
+
+    /// `deny` lets a blanket `s3:*` cover force-delete actions so an explicit
+    /// Deny still wins. Allow matching stays name-only for those actions.
+    pub fn is_match_for_effect(&self, action: &Action, deny: bool) -> bool {
        for act in self.0.iter() {
-            if act.is_match(action) {
+            if act.is_match_for_effect(action, deny) {
                return true;
            }

@@ -72,6 +78,22 @@ impl ActionSet {

        false
    }
+
+    /// Whether a statement with this `Action` list and `not_actions` as its
+    /// `NotAction` list covers `action`.
+    ///
+    /// `NotAction` always uses plain wildcard matching, so `NotAction: "s3:*"`
+    /// still excludes force-delete. An Allow grants force-delete only when its
+    /// `Action` list names it: an Allow built from `NotAction` alone never does.
+    pub fn statement_covers(&self, not_actions: &ActionSet, action: &Action, deny: bool) -> bool {
+        if not_actions.is_match_for_effect(action, true) {
+            return false;
+        }
+        if self.is_empty() {
+            return deny || !action_requires_explicit_grant(action);
+        }
+        self.is_match_for_effect(action, deny)
+    }
 }

 impl Deref for ActionSet {
@@ -155,10 +177,28 @@ pub enum Action {

 impl Action {
    pub fn is_match(&self, action: &Action) -> bool {
+        self.is_match_for_effect(action, false)
+    }
+
+    pub fn is_match_for_effect(&self, action: &Action, deny: bool) -> bool {
+        // Force-delete bypasses emptiness and version checks. A blanket `s3:*`
+        // / `*` Allow (including canned consoleAdmin) must not confer it; the
+        // statement has to name `s3:ForceDeleteBucket` or `s3:ForceDeleteObject`.
+        // The same wildcard on a Deny still matches, so explicit deny wins.
+        if !deny && matches!(self, Action::S3Action(S3Action::AllActions)) && action_requires_explicit_grant(action) {
+            return false;
+        }
        wildcard::is_match::<&str, &str>(self.into(), action.into())
    }
 }

+fn action_requires_explicit_grant(action: &Action) -> bool {
+    matches!(
+        action,
+        Action::S3Action(S3Action::ForceDeleteBucketAction | S3Action::ForceDeleteObjectAction)
+    )
+}
+
 impl From<&Action> for &str {
    fn from(value: &Action) -> &'static str {
        match value {
@@ -220,8 +260,14 @@ pub enum S3Action {
    CreateBucketAction,
    #[strum(serialize = "s3:DeleteBucket")]
    DeleteBucketAction,
+    /// DeleteBucket when `x-minio-force-delete` / `x-rustfs-force-delete` is set.
+    /// Not implied by `s3:*`; the action must be named on the statement.
    #[strum(serialize = "s3:ForceDeleteBucket")]
    ForceDeleteBucketAction,
+    /// Recursive DeleteObject selected by the same force-delete header.
+    /// Not implied by `s3:*`; the action must be named on the statement.
+    #[strum(serialize = "s3:ForceDeleteObject")]
+    ForceDeleteObjectAction,
    #[strum(serialize = "s3:DeleteBucketPolicy")]
    DeleteBucketPolicyAction,
    #[strum(serialize = "s3:DeleteBucketPublicAccessBlock")]
@@ -732,6 +778,26 @@ pub enum KmsAction {
 mod tests {
    use super::*;

+    #[test]
+    fn blanket_s3_wildcard_does_not_grant_force_delete() {
+        let wildcard = Action::try_from("s3:*").expect("s3:* parses");
+        let star = Action::try_from("*").expect("* parses as s3:*");
+        let force_bucket = Action::try_from("s3:ForceDeleteBucket").expect("force bucket action parses");
+        let force_object = Action::try_from("s3:ForceDeleteObject").expect("force object action parses");
+        let delete_object = Action::try_from("s3:DeleteObject").expect("delete object parses");
+
+        assert!(!wildcard.is_match(&force_bucket));
+        assert!(!wildcard.is_match(&force_object));
+        assert!(!star.is_match(&force_bucket));
+        assert!(!star.is_match(&force_object));
+        assert!(wildcard.is_match(&delete_object));
+        assert!(wildcard.is_match_for_effect(&force_bucket, true));
+        assert!(wildcard.is_match_for_effect(&force_object, true));
+        assert!(force_bucket.is_match(&force_bucket));
+        assert!(force_object.is_match(&force_object));
+        assert!(!force_bucket.is_match(&force_object));
+    }
+
    #[test]
    fn test_action_wildcard_parsing() {
        // Test that "*" parses to S3Action::AllActions
@@ -3368,4 +3368,123 @@ mod test {
        assert_eq!(round_trip.statements.len(), policy.statements.len());
        assert_eq!(round_trip.statements[0].effect, policy.statements[0].effect);
    }
+
+    #[tokio::test]
+    async fn force_delete_actions_require_an_explicit_grant() {
+        let wildcard = Policy::parse_config(
+            br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::*"]}]}"#,
+        )
+        .expect("wildcard policy parses");
+        let explicit = Policy::parse_config(
+            br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:ForceDeleteObject","s3:ForceDeleteBucket","s3:DeleteObject"],"Resource":["arn:aws:s3:::*"]}]}"#,
+        )
+        .expect("explicit policy parses");
+        let denied = Policy::parse_config(
+            br#"{"Version":"2012-10-17","Statement":[
+                {"Effect":"Allow","Action":["s3:ForceDeleteObject"],"Resource":["arn:aws:s3:::*"]},
+                {"Effect":"Deny","Action":["s3:ForceDeleteObject"],"Resource":["arn:aws:s3:::*"]}
+            ]}"#,
+        )
+        .expect("deny policy parses");
+
+        let wildcard_deny = Policy::parse_config(
+            br#"{"Version":"2012-10-17","Statement":[
+                {"Effect":"Allow","Action":["s3:ForceDeleteObject","s3:DeleteObject"],"Resource":["arn:aws:s3:::*"]},
+                {"Effect":"Deny","Action":["s3:*"],"Resource":["arn:aws:s3:::*"]}
+            ]}"#,
+        )
+        .expect("wildcard deny policy parses");
+
+        assert!(allows(&wildcard, "s3:DeleteObject").await);
+        assert!(!allows(&wildcard, "s3:ForceDeleteObject").await);
+        assert!(!allows(&wildcard, "s3:ForceDeleteBucket").await);
+        assert!(allows(&explicit, "s3:ForceDeleteObject").await);
+        assert!(allows(&explicit, "s3:ForceDeleteBucket").await);
+        assert!(!allows(&denied, "s3:ForceDeleteObject").await);
+        assert!(!allows(&wildcard_deny, "s3:ForceDeleteObject").await);
+        assert!(!allows(&wildcard_deny, "s3:DeleteObject").await);
+    }
+
+    #[tokio::test]
+    async fn not_action_allow_does_not_grant_force_delete() {
+        for not_action in ["s3:*", "s3:GetObject"] {
+            let iam = Policy::parse_config(
+                format!(
+                    r#"{{"Version":"2012-10-17","Statement":[{{"Effect":"Allow","NotAction":["{not_action}"],"Resource":["arn:aws:s3:::*"]}}]}}"#
+                )
+                .as_bytes(),
+            )
+            .expect("NotAction policy parses");
+            let bucket: BucketPolicy = serde_json::from_str(&format!(
+                r#"{{"Version":"2012-10-17","Statement":[{{"Effect":"Allow","Principal":{{"AWS":["*"]}},"NotAction":["{not_action}"],"Resource":["arn:aws:s3:::bucket","arn:aws:s3:::bucket/*"]}}]}}"#
+            ))
+            .expect("NotAction bucket policy parses");
+
+            for action in ["s3:ForceDeleteObject", "s3:ForceDeleteBucket"] {
+                assert!(!allows(&iam, action).await, "IAM NotAction {not_action} must not grant {action}");
+                assert!(
+                    !bucket_allows(&bucket, action).await,
+                    "bucket NotAction {not_action} must not grant {action}"
+                );
+            }
+        }
+
+        // NotAction still grants ordinary actions it does not exclude.
+        let iam = Policy::parse_config(
+            br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","NotAction":["s3:GetObject"],"Resource":["arn:aws:s3:::*"]}]}"#,
+        )
+        .expect("NotAction policy parses");
+        assert!(allows(&iam, "s3:DeleteObject").await);
+
+        // Deny + NotAction keeps plain wildcard semantics: s3:* excludes
+        // force-delete, a narrower exclusion still denies it.
+        let deny_all_but = |not_action: &str| {
+            Policy::parse_config(
+                format!(
+                    r#"{{"Version":"2012-10-17","Statement":[
+                        {{"Effect":"Allow","Action":["s3:ForceDeleteObject"],"Resource":["arn:aws:s3:::*"]}},
+                        {{"Effect":"Deny","NotAction":["{not_action}"],"Resource":["arn:aws:s3:::*"]}}
+                    ]}}"#
+                )
+                .as_bytes(),
+            )
+            .expect("deny NotAction policy parses")
+        };
+        assert!(allows(&deny_all_but("s3:*"), "s3:ForceDeleteObject").await);
+        assert!(!allows(&deny_all_but("s3:GetObject"), "s3:ForceDeleteObject").await);
+    }
+
+    async fn bucket_allows(policy: &BucketPolicy, action: &str) -> bool {
+        let conditions = HashMap::new();
+        policy
+            .is_allowed(&BucketPolicyArgs {
+                account: "user",
+                groups: &None,
+                action: Action::try_from(action).expect("action parses"),
+                bucket: "bucket",
+                conditions: &conditions,
+                is_owner: false,
+                object: "folder/a",
+            })
+            .await
+    }
+
+    async fn allows(policy: &Policy, action: &str) -> bool {
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let groups = None;
+        policy
+            .is_allowed(&Args {
+                account: "user",
+                groups: &groups,
+                action: Action::try_from(action).expect("action parses"),
+                bucket: "bucket",
+                conditions: &conditions,
+                is_owner: false,
+                object: "folder/a",
+                claims: &claims,
+                deny_only: false,
+            })
+            .await
+    }
 }
@@ -247,7 +247,8 @@ impl Statement {
    /// Returns true when this statement would reach `conditions.evaluate_with_resolver` in
    /// [`Statement::is_allowed`] (including the KMS resource path). Does not evaluate conditions.
    pub(crate) async fn request_reaches_condition_eval(&self, args: &Args<'_>, resolver: &VariableResolver) -> bool {
-        if (!self.actions.is_match(&args.action) && !self.actions.is_empty()) || self.not_actions.is_match(&args.action) {
+        let deny = matches!(self.effect, Effect::Deny);
+        if !self.actions.statement_covers(&self.not_actions, &args.action, deny) {
            return false;
        }

@@ -423,7 +424,8 @@ impl BPStatement {
            return false;
        }

-        if (!self.actions.is_match(&args.action) && !self.actions.is_empty()) || self.not_actions.is_match(&args.action) {
+        let deny = matches!(self.effect, Effect::Deny);
+        if !self.actions.statement_covers(&self.not_actions, &args.action, deny) {
            return false;
        }

@@ -91,6 +91,6 @@ crypto = ["dep:base64-simd", "dep:hex-simd", "dep:hmac", "dep:hyper", "dep:sha1"
 hash = ["dep:highway", "dep:md-5", "dep:sha2", "dep:blake2", "dep:serde", "dep:siphasher", "dep:hex-simd", "dep:crc-fast"]
 os = ["dep:rustix", "dep:tempfile", "dep:windows"]  # operating system utilities
 integration = []  # integration test features
-http = ["dep:convert_case", "dep:http", "dep:regex"]
+http = ["dep:convert_case", "dep:http", "dep:regex", "string"]
 obj = ["http"] # object storage features
 full = ["ip", "net", "egress", "io", "hash", "os", "integration", "path", "crypto", "string", "compress", "http", "obj"]  # all features
@@ -130,6 +130,19 @@ fn minio_key(suffix: &str) -> String {
    format!("{MINIO_PREFIX}{suffix}")
 }

+/// Reads `x-rustfs-force-delete` / `x-minio-force-delete`.
+///
+/// `Ok(None)` means the header is absent. `Ok(Some(_))` is a boolean accepted
+/// by [`crate::string::parse_bool`]. `Err` means the header is present but not
+/// a boolean; callers must reject that request instead of treating the flag as
+/// absent, which would turn a malformed force-delete into a normal delete.
+pub fn force_delete_header(headers: &HeaderMap) -> std::io::Result<Option<bool>> {
+    match get_header(headers, SUFFIX_FORCE_DELETE) {
+        None => Ok(None),
+        Some(value) => crate::string::parse_bool(value.as_ref()).map(Some),
+    }
+}
+
 /// Get header value: tries x-rustfs-{suffix} first, then x-minio-{suffix}. Case-insensitive.
 pub fn get_header<'a>(headers: &'a HeaderMap, suffix: &str) -> Option<Cow<'a, str>> {
    let rk = rustfs_key(suffix);
@@ -245,4 +258,20 @@ mod tests {
        headers2.insert("X-Rustfs-Force-Delete", HeaderValue::from_static("true"));
        assert_eq!(get_header(&headers2, SUFFIX_FORCE_DELETE).as_deref(), Some("true"));
    }
+
+    #[test]
+    fn force_delete_header_accepts_minio_bools_and_rejects_garbage() {
+        let mut headers = HeaderMap::new();
+        assert_eq!(force_delete_header(&headers).expect("absent header"), None);
+
+        headers.insert("x-minio-force-delete", HeaderValue::from_static("TRUE"));
+        assert_eq!(force_delete_header(&headers).expect("minio true"), Some(true));
+
+        headers.insert("x-rustfs-force-delete", HeaderValue::from_static("false"));
+        assert_eq!(force_delete_header(&headers).expect("rustfs header wins"), Some(false));
+
+        headers.clear();
+        headers.insert("x-rustfs-force-delete", HeaderValue::from_static("maybe"));
+        assert!(force_delete_header(&headers).is_err());
+    }
 }
@@ -95,9 +95,8 @@ use rustfs_targets::{
    arn::{ARN, TargetID, TargetIDError},
 };
 use rustfs_trusted_proxies::ClientInfo;
-use rustfs_utils::http::{SUFFIX_FORCE_DELETE, get_header};
+use rustfs_utils::http::{force_delete_header, get_header};
 use rustfs_utils::obj::extract_user_defined_metadata;
-use rustfs_utils::string::parse_bool;
 use s3s::dto::{
    BucketLifecycleConfiguration, BucketLocationConstraint, BucketVersioningStatus, CommonPrefix, CreateBucketInput,
    CreateBucketOutput, DeleteBucketCorsInput, DeleteBucketCorsOutput, DeleteBucketEncryptionInput, DeleteBucketEncryptionOutput,
@@ -1455,11 +1454,10 @@ impl DefaultBucketUsecase {
            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
        };

-        let force_str = get_header(&req.headers, SUFFIX_FORCE_DELETE)
-            .map(|v| v.into_owned())
-            .unwrap_or_default();
-
-        let force = parse_bool(&force_str).unwrap_or_default();
+        let force = match force_delete_header(&req.headers) {
+            Ok(value) => value.unwrap_or(false),
+            Err(_) => return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "Invalid force-delete header value")),
+        };

        if force {
            authorize_request(&mut req, Action::S3Action(S3Action::ForceDeleteBucketAction)).await?;
@@ -619,16 +619,22 @@ impl DefaultObjectUsecase {

        // Phase 1 (serial): derive storage options from the request-scoped
        // configuration after every candidate has passed authorization.
+        // DeleteObjects names each key explicitly. The force-delete header is a
+        // DeleteObject/DeleteBucket extension and must not change this batch,
+        // including by rejecting a non-boolean value.
+        let mut batch_headers = req.headers.clone();
+        batch_headers.remove("x-rustfs-force-delete");
+        batch_headers.remove("x-minio-force-delete");
        let mut prepared_deletes: Vec<PreparedDelete> = Vec::with_capacity(authorized_deletes.len());
        for authorized in authorized_deletes {
            let AuthorizedDelete { idx, object } = authorized;

-            let metadata = extract_metadata(&req.headers);
+            let metadata = extract_metadata(&batch_headers);
            let opts: ObjectOptions = del_opts_with_versioning(
                &bucket,
                &object.object_name,
                object.version_id.map(|f| f.to_string()),
-                &req.headers,
+                &batch_headers,
                metadata,
                version_cfg,
                false,
@@ -933,6 +939,10 @@ impl DefaultObjectUsecase {
            .get(AMZ_BUCKET_REPLICATION_STATUS)
            .map(|v| v.to_str().unwrap_or_default() == ReplicationStatusType::Replica.as_str())
            .unwrap_or_default();
+        let force_header = match rustfs_utils::http::force_delete_header(&req.headers) {
+            Ok(value) => value.unwrap_or(false),
+            Err(_) => return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "Invalid force-delete header value")),
+        };

        if replica {
            authorize_request(&mut req, Action::S3Action(S3Action::ReplicateDeleteAction)).await?;
@@ -945,6 +955,12 @@ impl DefaultObjectUsecase {
                "Recursive force-delete requires an authenticated caller",
            ));
        }
+        // The access layer checks the same action. Repeat it here so a direct
+        // usecase call cannot skip the dedicated permission. Root bypasses IAM.
+        // A caller-supplied REPLICA header does not.
+        if force_header && !req_info_ref(&req).is_ok_and(|info| info.is_owner) {
+            authorize_request(&mut req, Action::S3Action(S3Action::ForceDeleteObjectAction)).await?;
+        }
        validate_table_catalog_object_mutation(&bucket, &key).await?;

        // Establish bucket existence before any bucket-metadata work (matches
@@ -960,9 +976,7 @@ impl DefaultObjectUsecase {
        // Lock order is bucket lifecycle, then object/commit locks in storage.
        // Keep this guard alive through the physical delete: a preflight without
        // writer exclusion could authorize one subtree and delete a newer one.
-        let recursive_delete_guard = if rustfs_utils::http::get_header(&req.headers, rustfs_utils::http::SUFFIX_FORCE_DELETE)
-            .is_some_and(|value| value == "true")
-        {
+        let recursive_delete_guard = if force_header {
            Some(
                store
                    .lock_bucket_for_recursive_delete(&bucket)
@@ -49,9 +49,9 @@ use rustfs_policy::policy::{
 };
 use rustfs_trusted_proxies::ClientInfo;
 use rustfs_utils::http::{
-    AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_BYPASS_GOVERNANCE, SUFFIX_FORCE_DELETE, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE,
+    AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_BYPASS_GOVERNANCE, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE,
    SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK,
-    SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, get_header,
+    SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, force_delete_header, get_header,
 };
 use s3s::access::{S3Access, S3AccessContext};
 use s3s::{S3Error, S3ErrorCode, S3Request, S3Result, dto::*, s3_error};
@@ -118,9 +118,14 @@ pub(crate) fn recursive_force_delete_has_authenticated_caller(
    authenticated: bool,
    replica_request: bool,
 ) -> bool {
-    !get_header(headers, SUFFIX_FORCE_DELETE).is_some_and(|value| value.eq_ignore_ascii_case("true"))
-        || authenticated
-        || replica_request
+    match force_delete_header(headers) {
+        Ok(Some(true)) => authenticated || replica_request,
+        _ => true,
+    }
+}
+
+fn invalid_force_delete_header() -> S3Error {
+    S3Error::with_message(S3ErrorCode::InvalidRequest, "Invalid force-delete header value")
 }

 #[derive(Clone, Debug)]
@@ -1294,6 +1299,8 @@ pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3R
                    Action::S3Action(
                        S3Action::DeleteObjectAction
                            | S3Action::DeleteObjectVersionAction
+                            | S3Action::ForceDeleteBucketAction
+                            | S3Action::ForceDeleteObjectAction
                            | S3Action::ListBucketVersionsAction
                            | S3Action::BypassGovernanceRetentionAction
                            | S3Action::ReplicateDeleteAction
@@ -2262,7 +2269,13 @@ impl S3Access for FS {

        authorize_request(req, Action::S3Action(S3Action::DeleteBucketAction)).await?;

-        if req.input.force_delete.is_some_and(|v| v) {
+        // MinIO evaluates s3:ForceDeleteBucket whenever the header is present,
+        // including the value `false`. Only a parsed `true` later skips the
+        // emptiness check. `s3:*` does not grant this action.
+        if force_delete_header(&req.headers)
+            .map_err(|_| invalid_force_delete_header())?
+            .is_some()
+        {
            authorize_request(req, Action::S3Action(S3Action::ForceDeleteBucketAction)).await?;
        }
        Ok(())
@@ -2411,11 +2424,21 @@ impl S3Access for FS {

        authorize_request(req, action).await?;

+        let force_delete = match force_delete_header(&req.headers) {
+            Ok(value) => value.unwrap_or(false),
+            Err(_) => return Err(invalid_force_delete_header()),
+        };
        let replica_request = req
            .headers
            .get(AMZ_BUCKET_REPLICATION_STATUS)
            .and_then(|value| value.to_str().ok())
            .is_some_and(|value| value == ReplicationStatusType::Replica.as_str());
+        // The REPLICA header is caller-supplied, so it must not skip this
+        // check. s3:ReplicateDelete is still required separately for replica
+        // deletes; s3:* does not grant s3:ForceDeleteObject.
+        if force_delete {
+            authorize_request(req, Action::S3Action(S3Action::ForceDeleteObjectAction)).await?;
+        }
        if !recursive_force_delete_has_authenticated_caller(&req.headers, authenticated, replica_request) {
            return Err(s3_error!(AccessDenied, "Recursive force-delete requires an authenticated caller"));
        }
@@ -17,12 +17,11 @@ use crate::storage::storage_api::options_consumer::contract::{object::HTTPPrecon
 use http::header::{IF_MATCH, IF_NONE_MATCH};
 use http::{HeaderMap, HeaderValue};
 use rustfs_utils::http::{
-    AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP,
-    SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC,
-    SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_PROXY_REQUEST,
-    SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
-    SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID,
-    SUFFIX_TAGGING_TIMESTAMP, get_header,
+    AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP,
+    SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG,
+    SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_PROXY_REQUEST, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
+    SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
+    SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID, SUFFIX_TAGGING_TIMESTAMP, get_header,
    header_compat::{MINIO_ENCRYPTION_PREFIX, RUSTFS_ENCRYPTION_PREFIX},
    insert_header_map, insert_str,
    metadata_compat::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX, starts_with_ignore_ascii_case},
@@ -219,9 +218,17 @@ pub fn del_opts_with_versioning(
        StorageError::InvalidArgument(bucket.to_owned(), object.to_owned(), err.to_string())
    })?;

-    opts.delete_prefix = get_header(headers, SUFFIX_FORCE_DELETE)
-        .map(|v| v.as_ref() == "true")
-        .unwrap_or_default();
+    opts.delete_prefix = match rustfs_utils::http::force_delete_header(headers) {
+        Ok(Some(true)) => true,
+        Ok(_) => false,
+        Err(_) => {
+            return Err(StorageError::InvalidArgument(
+                bucket.to_owned(),
+                object.to_owned(),
+                "invalid force-delete header".to_owned(),
+            ));
+        }
+    };

    opts.version_id = synthetic_version_id.then(|| Uuid::nil().to_string()).or(vid);
    opts.synthetic_version_id = synthetic_version_id;
@@ -1387,12 +1394,10 @@ mod tests {
        let opts = result.unwrap();
        assert!(!opts.delete_prefix);

-        // Test with RUSTFS_FORCE_DELETE header set to other value
+        // A non-boolean force-delete header is rejected rather than ignored.
        insert_header(&mut headers, SUFFIX_FORCE_DELETE, "maybe");
        let result = del_opts("test-bucket", "test-object", None, &headers, metadata).await;
-        assert!(result.is_ok());
-        let opts = result.unwrap();
-        assert!(!opts.delete_prefix);
+        assert!(result.is_err());
    }

    #[tokio::test]