fix(authz): restrict recursive force-delete scope (#5610)

rustfs/src/app/object_usecase.rs

@@ -20,7 +20,8 @@ use rustfs_io_metrics::buffered_write;
 use crate::storage_api::table::get_bucket_metadata;
 
 use super::storage_api::object_usecase::access::{
-    PostObjectRequestMarker, authorize_request, has_bypass_governance_header, req_info_mut,
+    PostObjectRequestMarker, authorize_request, has_bypass_governance_header, recursive_force_delete_is_authorized, req_info_mut,
+    req_info_ref,
 };
 use super::storage_api::object_usecase::bucket::quota::checker::QuotaChecker;
 #[cfg(test)]
@@ -6614,6 +6615,13 @@ impl DefaultObjectUsecase {
             ));
         }
 
+        if !recursive_force_delete_is_authorized(&req.headers, req_info_ref(&req)?.is_owner, false) {
+            return Err(S3Error::with_message(
+                S3ErrorCode::AccessDenied,
+                "Recursive force-delete is restricted to administrative requests",
+            ));
+        }
+
         let Some(store) = self.object_store() else {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
         };
@@ -7073,6 +7081,14 @@ impl DefaultObjectUsecase {
             authorize_request(&mut req, Action::S3Action(S3Action::ReplicateDeleteAction)).await?;
         }
 
+        let is_owner = req_info_ref(&req)?.is_owner;
+        if !recursive_force_delete_is_authorized(&req.headers, is_owner, replica) {
+            return Err(S3Error::with_message(
+                S3ErrorCode::AccessDenied,
+                "Recursive force-delete is restricted to internal or administrative requests",
+            ));
+        }
+
         // Establish bucket existence before any bucket-metadata work (matches
         // PUT/GET): nonexistent buckets fail here instead of paying the
         // versioning lookups in del_opts/get_opts first. Resolve the store
@@ -13386,6 +13402,60 @@ mod tests {
         assert_eq!(internal_version_id, None);
     }
 
+    #[test]
+    fn recursive_force_delete_requires_administrative_or_replica_context() {
+        let mut headers = HeaderMap::new();
+        headers.insert("x-rustfs-force-delete", HeaderValue::from_static("true"));
+
+        assert!(!recursive_force_delete_is_authorized(&headers, false, false));
+        assert!(recursive_force_delete_is_authorized(&headers, true, false));
+        assert!(recursive_force_delete_is_authorized(&headers, false, true));
+        assert!(recursive_force_delete_is_authorized(&HeaderMap::new(), false, false));
+    }
+
+    #[tokio::test]
+    async fn execute_delete_object_rejects_untrusted_force_delete_before_store_access() {
+        let input = DeleteObjectInput::builder()
+            .bucket("test-bucket".to_string())
+            .key("prefix/object".to_string())
+            .build()
+            .unwrap();
+        let mut req = build_request(input, Method::DELETE);
+        req.headers.insert("x-rustfs-force-delete", HeaderValue::from_static("true"));
+        req.extensions.insert(crate::storage::access::ReqInfo::default());
+
+        let err = DefaultObjectUsecase::without_context()
+            .execute_delete_object(req)
+            .await
+            .expect_err("untrusted force-delete must be rejected before storage lookup");
+        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+    }
+
+    #[tokio::test]
+    async fn execute_delete_objects_rejects_untrusted_force_delete_before_store_access() {
+        let input = DeleteObjectsInput::builder()
+            .bucket("test-bucket".to_string())
+            .delete(Delete {
+                objects: vec![ObjectIdentifier {
+                    key: "prefix/object".to_string(),
+                    version_id: None,
+                    ..Default::default()
+                }],
+                quiet: None,
+            })
+            .build()
+            .unwrap();
+        let mut req = build_request(input, Method::POST);
+        req.headers.insert("x-rustfs-force-delete", HeaderValue::from_static("true"));
+        req.extensions.insert(crate::storage::access::ReqInfo::default());
+
+        let err = DefaultObjectUsecase::without_context()
+            .execute_delete_objects(req)
+            .await
+            .expect_err("untrusted force-delete must be rejected before storage lookup");
+        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+    }
+
     // backlog#929 (HP-8): the pre-delete stat may only be skipped when every
     // consumer of its result is provably idle. Each guard flips one condition
     // to prove the skip is fenced on all four data dependencies.

rustfs/src/app/storage_api.rs

@@ -217,7 +217,8 @@ pub(crate) mod runtime_sources {
 
 pub(crate) mod access {
     pub(crate) use crate::storage::storage_api::access_consumer::{
-        PostObjectRequestMarker, ReqInfo, authorize_request, has_bypass_governance_header, req_info_mut, req_info_ref,
+        PostObjectRequestMarker, ReqInfo, authorize_request, has_bypass_governance_header, recursive_force_delete_is_authorized,
+        req_info_mut, req_info_ref,
     };
 }
 

rustfs/src/storage/access.rs

@@ -15,7 +15,8 @@
 use super::ECStore;
 use super::ecfs::FS;
 use super::{
-    PolicySys, StorageError, get_bucket_metadata, get_bucket_policy_raw, get_public_access_block_config, is_err_bucket_not_found,
+    PolicySys, ReplicationStatusType, StorageError, get_bucket_metadata, get_bucket_policy_raw, get_public_access_block_config,
+    is_err_bucket_not_found,
 };
 use crate::auth::{
     check_key_valid_with_context, get_condition_values_with_client_info, get_condition_values_with_query_and_client_info,
@@ -36,7 +37,7 @@ use rustfs_policy::policy::{
     bucket_policy_uses_existing_object_tag_conditions,
 };
 use rustfs_trusted_proxies::ClientInfo;
-use rustfs_utils::http::AMZ_OBJECT_LOCK_BYPASS_GOVERNANCE;
+use rustfs_utils::http::{AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_BYPASS_GOVERNANCE, SUFFIX_FORCE_DELETE, get_header};
 use s3s::access::{S3Access, S3AccessContext};
 use s3s::{S3Error, S3ErrorCode, S3Request, S3Result, dto::*, s3_error};
 use std::collections::HashMap;
@@ -55,6 +56,12 @@ pub(crate) struct ReqInfo {
     pub request_context: Option<RequestContext>,
 }
 
+pub(crate) fn recursive_force_delete_is_authorized(headers: &HeaderMap, is_owner: bool, replica_request: bool) -> bool {
+    !get_header(headers, SUFFIX_FORCE_DELETE).is_some_and(|value| value.eq_ignore_ascii_case("true"))
+        || is_owner
+        || replica_request
+}
+
 #[derive(Clone, Debug)]
 pub(crate) struct PostObjectRequestMarker;
 
@@ -1310,9 +1317,22 @@ impl S3Access for FS {
         req_info.bucket = Some(req.input.bucket.clone());
         req_info.object = Some(req.input.key.clone());
         req_info.version_id = req.input.version_id.clone();
+        let is_owner = req_info.is_owner;
 
         authorize_request(req, Action::S3Action(S3Action::DeleteObjectAction)).await?;
 
+        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());
+        if !recursive_force_delete_is_authorized(&req.headers, is_owner, replica_request) {
+            return Err(s3_error!(
+                AccessDenied,
+                "Recursive force-delete is restricted to internal or administrative requests"
+            ));
+        }
+
         // S3 Standard: When bypass_governance header is set, must have s3:BypassGovernanceRetention permission
         if has_bypass_governance_header(&req.headers) {
             authorize_request(req, Action::S3Action(S3Action::BypassGovernanceRetentionAction)).await?;

rustfs/src/storage/storage_api.rs

@@ -102,7 +102,8 @@ pub(crate) use super::sse::{
 
 pub(crate) mod access_consumer {
     pub(crate) use super::super::access::{
-        PostObjectRequestMarker, ReqInfo, authorize_request, has_bypass_governance_header, req_info_mut, req_info_ref,
+        PostObjectRequestMarker, ReqInfo, authorize_request, has_bypass_governance_header, recursive_force_delete_is_authorized,
+        req_info_mut, req_info_ref,
     };
 
     pub(crate) mod contract {