fix(ecstore): gate null-version overwrites under Object Lock (#9157)

A put-like write without a version id to a key that is not
versioning-Enabled replaces the key's null version in place. The
commit-time Object Lock gate in put_object_with_old_current_size_inner
ran only when the write carried an explicit version id, and the
request-layer pre-check reads the key's latest version. When the
retained null version was not the latest entry (a UUID version or a
delete marker sat above it), a user holding only s3:PutObject could
overwrite it together with its COMPLIANCE retention or legal hold, and
CompleteMultipartUpload had no commit-time gate at all.

Name the version a commit replaces (the explicit id, or the nil id when
the write is not versioned and the bucket's Object Lock snapshot is not
ConfirmedAbsent) and run the existing WORM gate against it, tolerating
a null delete marker. complete_multipart_upload gets the same gate at
its commit point, and ECStore::handle_complete_multipart_upload attaches
the bucket's Object Lock snapshot the way a single PUT already does.

Refs rustfs/backlog#3019 (ANT-2026-EH17ZP59).

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -57,7 +57,6 @@ use super::object::{
     object_transaction_fencing_requested, read_object_transaction_epoch_fence, verify_object_transaction_epoch_fence,
 };
 use crate::api::config::storageclass;
-#[cfg(test)]
 use crate::bucket::metadata_sys::ObjectLockConfigState;
 use crate::bucket::quota::reservation;
 use crate::bucket::replication::ReplicationStatusType;
@@ -4114,6 +4113,40 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         )
         .await?;
 
+        if !opts.versioned
+            && !crate::bucket::utils::is_meta_bucketname(bucket)
+            && let Some(snapshot) = opts.object_lock_config_snapshot.as_deref()
+            && !matches!(snapshot.state(), ObjectLockConfigState::ConfirmedAbsent)
+        {
+            // Completion replaces the null version, not necessarily the latest
+            // version inspected by the request-level Object Lock check.
+            let current = self
+                .get_object_info(
+                    bucket,
+                    object,
+                    &ObjectOptions {
+                        version_id: Some(Uuid::nil().to_string()),
+                        no_lock: true,
+                        metadata_cache_safe: false,
+                        versioned: opts.versioned,
+                        version_suspended: opts.version_suspended,
+                        ..Default::default()
+                    },
+                )
+                .await;
+            match current {
+                Ok(existing) => {
+                    if check_object_lock_for_deletion_with_state(snapshot.state(), &existing, false)?.is_some() {
+                        return Err(StorageError::PrefixAccessDenied(bucket.to_string(), object.to_string()));
+                    }
+                }
+                // A null delete marker has no protected payload to replace.
+                Err(StorageError::MethodNotAllowed) => {}
+                Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {}
+                Err(err) => return Err(err),
+            }
+        }
+
         if opts.data_movement
             && opts.http_preconditions.is_some()
             && let Some(err) = self.check_write_precondition(bucket, object, opts).await

crates/ecstore/src/set_disk/ops/object.rs

@@ -4958,7 +4958,20 @@ impl SetDisks {
                 );
             }
 
-            if let Some(version_id) = opts.version_id.as_deref()
+            // A write without a version id to a key that is not versioning-Enabled
+            // replaces the key's null version in place, so the WORM gate must judge
+            // that version too. The request-layer pre-check reads the latest
+            // version, which is not the one replaced when a UUID version or a
+            // delete marker sits above a retained null version.
+            let worm_gate_version_id = opts.version_id.clone().or_else(|| {
+                (!opts.versioned
+                    && opts
+                        .object_lock_config_snapshot
+                        .as_ref()
+                        .is_some_and(|snapshot| !matches!(snapshot.state(), ObjectLockConfigState::ConfirmedAbsent)))
+                .then(|| Uuid::nil().to_string())
+            });
+            if let Some(version_id) = worm_gate_version_id.as_deref()
                 && !is_meta_bucketname(bucket)
             {
                 let explicit_version_lookup_started = rustfs_io_metrics::put_stage_timer();
@@ -4978,9 +4991,10 @@ impl SetDisks {
                     .await;
                 match current {
                     Ok(existing) => {
-                        let object_lock_config = opts.object_lock_config_snapshot.as_deref().ok_or_else(|| {
-                            Error::other("explicit-version PUT is missing its Object Lock configuration snapshot")
-                        })?;
+                        let object_lock_config = opts
+                            .object_lock_config_snapshot
+                            .as_deref()
+                            .ok_or_else(|| Error::other("destructive PUT is missing its Object Lock configuration snapshot"))?;
                         // The WORM gate protects the locked version from local
                         // overwrites; an authorized replication write passes it
                         // only when the LWW merge below will judge every
@@ -5011,6 +5025,8 @@ impl SetDisks {
                             }
                         }
                     }
+                    // A null delete marker has no protected payload to replace.
+                    Err(StorageError::MethodNotAllowed) if version_id == Uuid::nil().to_string() => {}
                     Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {}
                     Err(err) => return Err(err),
                 }

crates/ecstore/src/store/init.rs

@@ -27549,4 +27549,85 @@ mod tests {
             }
         }
     }
+
+    /// A versionless PUT or multipart completion replaces the null version in
+    /// place, so the commit-time WORM gate must judge that version even when a
+    /// UUID version or delete marker is the key's latest entry
+    /// (rustfs/backlog#3019).
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn null_version_object_lock_put_and_multipart_preserve_retained_target() {
+        let temp = tempfile::tempdir().expect("create null-version overwrite store");
+        let (_ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp.path(), "null-version-overwrite-lock", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("null-overwrite-lock-{}", Uuid::new_v4());
+        store
+            .make_bucket(
+                &bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("create Object Lock bucket");
+        for protection in ["COMPLIANCE", "GOVERNANCE", "legal-hold"] {
+            for latest_is_marker in [false, true] {
+                let object = format!("{protection}-{latest_is_marker}");
+                seed_retained_null_version(&store, &bucket, &object, protection, latest_is_marker).await;
+                let mut replacement = PutObjReader::from_vec(b"replacement".to_vec());
+                let error = store
+                    .put_object(&bucket, &object, &mut replacement, &ObjectOptions::default())
+                    .await
+                    .expect_err("PUT must reject replacement of a retained null version");
+                assert!(matches!(error, StorageError::PrefixAccessDenied(_, _)), "{object}: {error:?}");
+
+                let upload = store
+                    .clone()
+                    .new_multipart_upload(&bucket, &object, &ObjectOptions::default())
+                    .await
+                    .expect("initiate multipart upload");
+                let mut part_reader = PutObjReader::from_vec(b"multipart replacement".to_vec());
+                let part = store
+                    .clone()
+                    .put_object_part(&bucket, &object, &upload.upload_id, 1, &mut part_reader, &ObjectOptions::default())
+                    .await
+                    .expect("stage multipart replacement");
+                let error = store
+                    .clone()
+                    .complete_multipart_upload(
+                        &bucket,
+                        &object,
+                        &upload.upload_id,
+                        vec![crate::storage_api_contracts::multipart::CompletePart {
+                            part_num: part.part_num,
+                            etag: part.etag,
+                            ..Default::default()
+                        }],
+                        &ObjectOptions::default(),
+                    )
+                    .await
+                    .expect_err("multipart completion must reject replacement of a retained null version");
+                assert!(matches!(error, StorageError::PrefixAccessDenied(_, _)), "{object}: {error:?}");
+                assert_retained_null_version_body(&store, &bucket, &object).await;
+
+                let mut next_version = PutObjReader::from_vec(b"new UUID version".to_vec());
+                store
+                    .put_object(
+                        &bucket,
+                        &object,
+                        &mut next_version,
+                        &ObjectOptions {
+                            versioned: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("a new UUID version must remain allowed");
+                assert_retained_null_version_body(&store, &bucket, &object).await;
+            }
+        }
+    }
 }

crates/ecstore/src/store/multipart.rs

@@ -1210,7 +1210,34 @@ impl ECStore {
         opts: &ObjectOptions,
     ) -> Result<ObjectInfo> {
         check_complete_multipart_args(bucket, object, upload_id)?;
+        let encoded_object = encode_dir_object_ref(object);
+        let object = encoded_object.as_ref();
         let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        if opts.overwrites_existing_version() && !is_meta_bucketname(bucket) {
+            // The set-level commit judges the null version this completion
+            // replaces; give it the same authoritative snapshot a PUT carries.
+            let expected_incarnation_id = opts
+                .expected_bucket_incarnation_id
+                .ok_or_else(|| Error::other("multipart completion is missing its bucket incarnation"))?;
+            let lifecycle_fence = opts
+                .bucket_lifecycle_lock_fence
+                .as_ref()
+                .ok_or_else(|| Error::other("multipart completion is missing its bucket lifecycle fence"))?;
+            let snapshot = match opts.object_lock_config_snapshot.as_ref() {
+                Some(snapshot) => Arc::clone(snapshot),
+                None => {
+                    self.object_lock_config_snapshot_under_lifecycle_fence(bucket, lifecycle_fence)
+                        .await?
+                }
+            };
+            if !snapshot.is_valid_for_destructive_put(self.id, bucket, expected_incarnation_id) {
+                return Err(Error::other(
+                    "Object Lock snapshot does not hold valid target bucket generation and configuration fences",
+                ));
+            }
+            snapshot.add_lock_fences(&mut opts);
+            opts.object_lock_config_snapshot = Some(snapshot);
+        }
         if !self.single_pool() {
             opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
         }

crates/policy/src/arn.rs

@@ -20,7 +20,7 @@ const ARN_PARTITION_RUSTFS: &str = "rustfs";
 const ARN_SERVICE_IAM: &str = "iam";
 const ARN_RESOURCE_TYPE_ROLE: &str = "role";
 
-#[derive(Debug, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct ARN {
     pub partition: String,
     pub service: String,

crates/policy/src/policy/action.rs

@@ -165,9 +165,10 @@ impl<'de> Deserialize<'de> for ActionSet {
     }
 }
 
-#[derive(Serialize, Deserialize, Hash, PartialEq, Eq, Clone, Debug, Copy)]
+#[derive(Deserialize, Hash, PartialEq, Eq, Clone, Debug, Copy)]
 #[serde(try_from = "&str", untagged)]
 pub enum Action {
+    All,
     S3Action(S3Action),
     AdminAction(AdminAction),
     StsAction(StsAction),
@@ -175,12 +176,29 @@ pub enum Action {
     None,
 }
 
+impl Serialize for Action {
+    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        match self {
+            Self::None => serializer.serialize_unit(),
+            _ => serializer.serialize_str(self.into()),
+        }
+    }
+}
+
 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 {
+        if matches!(self, Self::All) {
+            // Preserve the historical S3-only Allow while allowing deny-all and
+            // NotAction exclusions to cover every action family.
+            return deny || Self::S3Action(S3Action::AllActions).is_match_for_effect(action, false);
+        }
         // 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`.
@@ -202,6 +220,7 @@ fn action_requires_explicit_grant(action: &Action) -> bool {
 impl From<&Action> for &str {
     fn from(value: &Action) -> &'static str {
         match value {
+            Action::All => "*",
             Action::S3Action(s) => s.into(),
             Action::AdminAction(s) => s.into(),
             Action::StsAction(s) => s.into(),
@@ -221,9 +240,8 @@ impl Action {
 impl TryFrom<&str> for Action {
     type Error = Error;
     fn try_from(value: &str) -> std::result::Result<Self, Self::Error> {
-        // Support wildcard "*" which matches all S3 actions (AWS S3 standard)
         if value == "*" {
-            return Ok(Self::S3Action(S3Action::AllActions));
+            return Ok(Self::All);
         }
         if value.starts_with(Self::S3_PREFIX) {
             Ok(Self::S3Action(
@@ -822,7 +840,7 @@ mod tests {
     #[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 star = Action::try_from("*").expect("* parses");
         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");
@@ -841,9 +859,10 @@ mod tests {
 
     #[test]
     fn test_action_wildcard_parsing() {
-        // Test that "*" parses to S3Action::AllActions
+        // Preserve the bare wildcard through serialization.
         let action = Action::try_from("*").expect("Should parse wildcard");
-        assert!(matches!(action, Action::S3Action(S3Action::AllActions)));
+        assert!(matches!(action, Action::All));
+        assert_eq!(serde_json::to_string(&action).expect("serialize wildcard"), r#""*""#);
     }
 
     #[test]
@@ -929,15 +948,13 @@ mod tests {
 
     #[test]
     fn test_actionset_wildcard_serialization() {
-        // Wildcard action should serialize as array for S3 specification compliance
-        let actionset = ActionSet(vec![Action::try_from("*").expect("Should parse wildcard")]);
-
-        let json = serde_json::to_string(&actionset).expect("Should serialize");
-        let parsed: serde_json::Value = serde_json::from_str(&json).expect("Should parse");
-        assert!(parsed.is_array(), "Should serialize as array");
-        let arr = parsed.as_array().expect("Should be array");
-        assert_eq!(arr.len(), 1);
-        assert_eq!(arr[0].as_str().unwrap(), "s3:*");
+        for wildcard in ["*", "s3:*"] {
+            let actionset = ActionSet(vec![Action::try_from(wildcard).expect("parse wildcard")]);
+            let json = serde_json::to_string(&actionset).expect("serialize action set");
+            assert_eq!(json, format!(r#"["{wildcard}"]"#));
+            let round_trip: ActionSet = serde_json::from_str(&json).expect("deserialize action set");
+            assert_eq!(round_trip, actionset);
+        }
     }
 
     #[test]

crates/policy/src/policy/function/addr.rs

@@ -15,7 +15,7 @@
 use super::func::InnerFunc;
 use ipnetwork::IpNetwork;
 use serde::{Deserialize, Serialize, de::Visitor};
-use std::{borrow::Cow, collections::HashMap, net::IpAddr};
+use std::{collections::HashMap, net::IpAddr};
 
 pub type AddrFunc = InnerFunc<AddrFuncValue>;
 
@@ -30,7 +30,16 @@ impl AddrFunc {
                 };
 
                 for ip_net in inner.values.0.iter() {
-                    if ip_net.contains(ip) {
+                    let canonical = ip.to_canonical();
+                    let matches_mapped_network = match (ip_net, canonical) {
+                        (IpNetwork::V6(network), IpAddr::V4(address))
+                            if network.prefix() >= 96 && network.ip().to_ipv4_mapped().is_some() =>
+                        {
+                            network.contains(address.to_ipv6_mapped())
+                        }
+                        _ => false,
+                    };
+                    if ip_net.contains(ip) || ip_net.contains(canonical) || matches_mapped_network {
                         return true;
                     }
                 }
@@ -81,13 +90,7 @@ impl<'de> Deserialize<'de> for AddrFuncValue {
 
         impl AddrFuncValueVisitor {
             fn cidr<E: serde::de::Error>(v: &str) -> Result<IpNetwork, E> {
-                let mut cidr_str = Cow::from(v);
-                if v.find('/').is_none() {
-                    cidr_str.to_mut().push_str("/32");
-                }
-
-                cidr_str
-                    .parse::<IpNetwork>()
+                v.parse::<IpNetwork>()
                     .map_err(|_| E::custom(format!("{v} can not be parsed to CIDR")))
             }
         }
@@ -107,6 +110,40 @@ mod tests {
     };
     use test_case::test_case;
 
+    #[test]
+    fn source_ip_matches_ipv4_and_mapped_forms_across_protocols() {
+        for network in ["203.0.113.0/24", "::ffff:203.0.113.0/120"] {
+            let function: AddrFunc =
+                serde_json::from_str(&format!(r#"{{"aws:SourceIp":"{network}"}}"#)).expect("parse source IP condition");
+            for address in ["203.0.113.9", "::ffff:203.0.113.9"] {
+                assert!(
+                    function.evaluate(&std::collections::HashMap::from([("SourceIp".to_string(), vec![address.to_string()]),])),
+                    "{network} must match {address}"
+                );
+            }
+            for address in ["198.51.100.9", "::ffff:198.51.100.9", "2001:db8::9"] {
+                assert!(
+                    !function.evaluate(&std::collections::HashMap::from([("SourceIp".to_string(), vec![address.to_string()]),])),
+                    "{network} must not match {address}"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn bare_address_matches_only_the_named_host() {
+        for (host, neighbor, prefix) in [
+            ("2001:db8:85a3::8a2e:370:7334", "2001:db8:ffff::1", 128),
+            ("203.0.113.7", "203.0.113.8", 32),
+        ] {
+            let function: AddrFunc =
+                serde_json::from_str(&format!(r#"{{"aws:SourceIp":"{host}"}}"#)).expect("parse bare IP address condition");
+            assert_eq!(function.0[0].values.0[0].prefix(), prefix);
+            assert!(function.evaluate(&std::collections::HashMap::from([("SourceIp".to_string(), vec![host.to_string()])])));
+            assert!(!function.evaluate(&std::collections::HashMap::from([("SourceIp".to_string(), vec![neighbor.to_string()])])));
+        }
+    }
+
     fn new_func(name: KeyName, variable: Option<String>, value: Vec<&str>) -> AddrFunc {
         AddrFunc {
             0: vec![FuncKeyValue {
@@ -119,20 +156,20 @@ mod tests {
     #[test_case(r#"{"aws:SourceIp": "203.0.113.0/24"}"#, new_func(Aws(AWSSourceIP), None, vec!["203.0.113.0/24"]); "1")]
     #[test_case(r#"{"aws:SourceIp": "203.0.113.0"}"#, new_func(Aws(AWSSourceIP), None, vec!["203.0.113.0/32"]); "2")]
     #[test_case(r#"{"aws:SourceIp": "2001:DB8:1234:5678::/64"}"#, new_func(Aws(AWSSourceIP),None, vec!["2001:DB8:1234:5678::/64"]); "3")]
-    #[test_case(r#"{"aws:SourceIp": "2001:DB8:1234:5678::"}"#, new_func(Aws(AWSSourceIP), None, vec!["2001:DB8:1234:5678::/32"]); "4")]
+    #[test_case(r#"{"aws:SourceIp": "2001:DB8:1234:5678::"}"#, new_func(Aws(AWSSourceIP), None, vec!["2001:DB8:1234:5678::/128"]); "4")]
     #[test_case(r#"{"aws:SourceIp": ["203.0.113.0/24","203.0.113.0"]}"#, new_func(Aws(AWSSourceIP), None, vec!["203.0.113.0/24", "203.0.113.0/32"]); "5")]
     #[test_case(r#"{"aws:SourceIp": ["2001:DB8:1234:5678::/64","203.0.113.0/24"]}"#, new_func(Aws(AWSSourceIP), None, vec!["2001:DB8:1234:5678::/64", "203.0.113.0/24"]); "6")]
-    #[test_case(r#"{"aws:SourceIp": ["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::"]}"#, new_func(Aws(AWSSourceIP),None, vec!["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::/32"]); "7")]
-    #[test_case(r#"{"aws:SourceIp": ["2001:DB8:1234:5678::", "203.0.113.0"]}"#, new_func(Aws(AWSSourceIP), None, vec!["2001:DB8:1234:5678::/32", "203.0.113.0/32"]); "8")]
+    #[test_case(r#"{"aws:SourceIp": ["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::"]}"#, new_func(Aws(AWSSourceIP),None, vec!["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::/128"]); "7")]
+    #[test_case(r#"{"aws:SourceIp": ["2001:DB8:1234:5678::", "203.0.113.0"]}"#, new_func(Aws(AWSSourceIP), None, vec!["2001:DB8:1234:5678::/128", "203.0.113.0/32"]); "8")]
     #[test_case(r#"{"aws:SourceIp/a": "203.0.113.0/24"}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["203.0.113.0/24"]); "9")]
     #[test_case(r#"{"aws:SourceIp/a": "203.0.113.0/24"}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["203.0.113.0/24"]); "10")]
     #[test_case(r#"{"aws:SourceIp/a": "203.0.113.0"}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["203.0.113.0/32"]); "11")]
     #[test_case(r#"{"aws:SourceIp/a": "2001:DB8:1234:5678::/64"}"#, new_func(Aws(AWSSourceIP),Some("a".into()), vec!["2001:DB8:1234:5678::/64"]); "12")]
-    #[test_case(r#"{"aws:SourceIp/a": "2001:DB8:1234:5678::"}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["2001:DB8:1234:5678::/32"]); "13")]
+    #[test_case(r#"{"aws:SourceIp/a": "2001:DB8:1234:5678::"}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["2001:DB8:1234:5678::/128"]); "13")]
     #[test_case(r#"{"aws:SourceIp/a": ["203.0.113.0/24", "203.0.113.0"]}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["203.0.113.0/24", "203.0.113.0/32"]); "14")]
     #[test_case(r#"{"aws:SourceIp/a": ["2001:DB8:1234:5678::/64", "203.0.113.0/24"]}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["2001:DB8:1234:5678::/64", "203.0.113.0/24"]); "15")]
-    #[test_case(r#"{"aws:SourceIp/a": ["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::"]}"#, new_func(Aws(AWSSourceIP),Some("a".into()), vec!["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::/32"]); "16")]
-    #[test_case(r#"{"aws:SourceIp/a": ["2001:DB8:1234:5678::", "203.0.113.0"]}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["2001:DB8:1234:5678::/32", "203.0.113.0/32"]); "17")]
+    #[test_case(r#"{"aws:SourceIp/a": ["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::"]}"#, new_func(Aws(AWSSourceIP),Some("a".into()), vec!["2001:DB8:1234:5678::/64", "2001:DB8:1234:5678::/128"]); "16")]
+    #[test_case(r#"{"aws:SourceIp/a": ["2001:DB8:1234:5678::", "203.0.113.0"]}"#, new_func(Aws(AWSSourceIP), Some("a".into()), vec!["2001:DB8:1234:5678::/128", "203.0.113.0/32"]); "17")]
     fn test_deser(input: &str, expect: AddrFunc) -> Result<(), serde_json::Error> {
         let v: AddrFunc = serde_json::from_str(input)?;
         assert_eq!(v, expect);

crates/policy/src/policy/function/condition.rs

@@ -61,6 +61,11 @@ pub enum Condition {
 
 impl Condition {
     pub fn from_deserializer<'a, D: MapAccess<'a>>(key: &str, d: &mut D) -> Result<Self, D::Error> {
+        // Two suffixes are retained for existing serialized policies. Bound the
+        // chain before parsing so Clone, evaluation and Drop stay bounded too.
+        if key.ends_with("IfExistsIfExistsIfExists") {
+            return Err(Error::custom("condition operator has too many IfExists suffixes"));
+        }
         Ok(match key {
             "StringEquals" => Self::StringEquals(d.next_value()?),
             "StringNotEquals" => Self::StringNotEquals(d.next_value()?),
@@ -262,18 +267,18 @@ impl Condition {
                 Null(s) => s.evaluate_null(values),
                 Bool(s) => s.evaluate_bool(values),
                 NumericEquals(s) => s.evaluate(i64::eq, false, values),
-                NumericNotEquals(s) => s.evaluate(i64::ne, false, values),
+                NumericNotEquals(s) => s.evaluate(i64::ne, true, values),
                 NumericLessThan(s) => s.evaluate(i64::lt, false, values),
                 NumericLessThanEquals(s) => s.evaluate(i64::le, false, values),
                 NumericGreaterThan(s) => s.evaluate(i64::gt, false, values),
                 NumericGreaterThanIfExists(s) => s.evaluate(i64::ge, true, values),
                 NumericGreaterThanEquals(s) => s.evaluate(i64::ge, false, values),
-                DateEquals(s) => s.evaluate(OffsetDateTime::eq, values),
-                DateNotEquals(s) => s.evaluate(OffsetDateTime::ne, values),
-                DateLessThan(s) => s.evaluate(OffsetDateTime::lt, values),
-                DateLessThanEquals(s) => s.evaluate(OffsetDateTime::le, values),
-                DateGreaterThan(s) => s.evaluate(OffsetDateTime::gt, values),
-                DateGreaterThanEquals(s) => s.evaluate(OffsetDateTime::ge, values),
+                DateEquals(s) => s.evaluate(OffsetDateTime::eq, false, values),
+                DateNotEquals(s) => s.evaluate(OffsetDateTime::ne, true, values),
+                DateLessThan(s) => s.evaluate(OffsetDateTime::lt, false, values),
+                DateLessThanEquals(s) => s.evaluate(OffsetDateTime::le, false, values),
+                DateGreaterThan(s) => s.evaluate(OffsetDateTime::gt, false, values),
+                DateGreaterThanEquals(s) => s.evaluate(OffsetDateTime::ge, false, values),
                 IfExists(inner) => {
                     if !inner.has_any_key_in(values) {
                         return true;
@@ -486,6 +491,31 @@ mod tests {
         assert_eq!(functions, deserialized);
     }
 
+    #[test]
+    fn repeated_if_exists_suffixes_are_rejected_before_recursion() {
+        for count in [3, 10_000] {
+            let key = format!("StringEquals{}", "IfExists".repeat(count));
+            let document = serde_json::json!({key: {"s3:x-amz-server-side-encryption": "aws:kms"}});
+            let result = serde_json::from_slice::<Functions>(&serde_json::to_vec(&document).expect("serialize condition"));
+            assert!(result.is_err(), "{count} suffixes must be rejected without constructing a deep chain");
+        }
+    }
+
+    #[tokio::test]
+    async fn numeric_and_date_not_equals_match_absent_keys() {
+        for (operator, key, value, absent) in [
+            ("NumericNotEquals", "s3:max-keys", "30", true),
+            ("NumericEquals", "s3:max-keys", "30", false),
+            ("DateNotEquals", "s3:object-lock-retain-until-date", "2030-01-01T00:00:00Z", true),
+            ("DateEquals", "s3:object-lock-retain-until-date", "2030-01-01T00:00:00Z", false),
+        ] {
+            let document = serde_json::json!({operator: {key: value}});
+            let functions: Functions =
+                serde_json::from_slice(&serde_json::to_vec(&document).expect("serialize condition")).expect("parse condition");
+            assert_eq!(functions.evaluate(&HashMap::new()).await, absent, "{operator}");
+        }
+    }
+
     #[test]
     fn test_nested_if_exists_serde_round_trip() {
         // Verifies that nested IfExists(IfExists(StringEquals)) serializes with the

crates/policy/src/policy/function/date.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use super::func::InnerFunc;
+use super::key_name::{AwsKeyName, KeyName};
 use serde::{Deserialize, Deserializer, Serialize, de};
 use std::{collections::HashMap, fmt};
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
@@ -20,14 +21,27 @@ use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 pub type DateFunc = InnerFunc<DateFuncValue>;
 
 impl DateFunc {
-    pub fn evaluate(&self, op: impl Fn(&OffsetDateTime, &OffsetDateTime) -> bool, values: &HashMap<String, Vec<String>>) -> bool {
+    pub fn evaluate(
+        &self,
+        op: impl Fn(&OffsetDateTime, &OffsetDateTime) -> bool,
+        absent: bool,
+        values: &HashMap<String, Vec<String>>,
+    ) -> bool {
         for inner in self.0.iter() {
             let v = match values.get(inner.key.name().as_str()).and_then(|x| x.first()) {
                 Some(x) => x,
+                None if absent => continue,
                 None => return false,
             };
 
-            let Ok(rv) = OffsetDateTime::parse(v, &Rfc3339) else {
+            let parsed = if inner.key.is(&KeyName::Aws(AwsKeyName::AWSEpochTime)) {
+                v.parse::<i64>()
+                    .ok()
+                    .and_then(|seconds| OffsetDateTime::from_unix_timestamp(seconds).ok())
+            } else {
+                OffsetDateTime::parse(v, &Rfc3339).ok()
+            };
+            let Some(rv) = parsed else {
                 return false;
             };
 
@@ -88,7 +102,7 @@ impl<'de> Deserialize<'de> for DateFuncValue {
 
 #[cfg(test)]
 mod tests {
-    use super::{DateFunc, DateFuncValue};
+    use super::{AwsKeyName, DateFunc, DateFuncValue};
     use crate::policy::function::func::FuncKeyValue;
     use crate::policy::function::{
         key::Key,
@@ -124,15 +138,50 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn date_conditions_compare_epoch_seconds_only_for_epoch_time() {
+        let function = new_func(KeyName::Aws(AwsKeyName::AWSEpochTime), None, "2020-01-01T00:00:00Z");
+        for (value, equal, later) in [
+            ("1577836799", false, false),
+            ("1577836800", true, false),
+            ("1577836801", false, true),
+        ] {
+            let values = HashMap::from([("EpochTime".to_string(), vec![value.to_string()])]);
+            assert_eq!(function.evaluate(OffsetDateTime::eq, false, &values), equal);
+            assert_eq!(function.evaluate(OffsetDateTime::gt, false, &values), later);
+            assert_eq!(function.evaluate(OffsetDateTime::lt, false, &values), !equal && !later);
+        }
+        for invalid in ["2020-01-01T00:00:00Z", "invalid", "9223372036854775807"] {
+            let values = HashMap::from([("EpochTime".to_string(), vec![invalid.to_string()])]);
+            assert!(!function.evaluate(OffsetDateTime::ne, true, &values));
+        }
+        let header_date = new_func(S3(S3ObjectLockRetainUntilDate), None, "2020-01-01T00:00:00Z");
+        let values = HashMap::from([("object-lock-retain-until-date".to_string(), vec!["1577836800".to_string()])]);
+        assert!(!header_date.evaluate(OffsetDateTime::eq, false, &values));
+    }
+
+    #[test]
+    fn absent_date_key_does_not_skip_other_keys() {
+        let mut function = new_func(S3(S3ObjectLockRetainUntilDate), None, "2030-01-01T00:00:00Z");
+        function
+            .0
+            .extend(new_func(KeyName::Aws(AwsKeyName::AWSCurrentTime), None, "2030-01-01T00:00:00Z").0);
+        let values = HashMap::from([("CurrentTime".to_string(), vec!["2030-01-01T00:00:00Z".to_string()])]);
+        assert!(!function.evaluate(OffsetDateTime::ne, true, &values));
+        assert!(function.evaluate(OffsetDateTime::eq, true, &values));
+        function.0.reverse();
+        assert!(!function.evaluate(OffsetDateTime::ne, true, &values));
+    }
+
     #[test]
     fn evaluate_compares_request_date_to_policy_date() {
         let function = new_func(S3(S3ObjectLockRetainUntilDate), None, "2030-01-01T00:00:00Z");
         let later = HashMap::from([("object-lock-retain-until-date".to_string(), vec!["2099-01-01T00:00:00Z".to_string()])]);
         let earlier = HashMap::from([("object-lock-retain-until-date".to_string(), vec!["2029-01-01T00:00:00Z".to_string()])]);
 
-        assert!(function.evaluate(OffsetDateTime::gt, &later));
-        assert!(!function.evaluate(OffsetDateTime::gt, &earlier));
-        assert!(function.evaluate(OffsetDateTime::lt, &earlier));
-        assert!(!function.evaluate(OffsetDateTime::lt, &later));
+        assert!(function.evaluate(OffsetDateTime::gt, false, &later));
+        assert!(!function.evaluate(OffsetDateTime::gt, false, &earlier));
+        assert!(function.evaluate(OffsetDateTime::lt, false, &earlier));
+        assert!(!function.evaluate(OffsetDateTime::lt, false, &later));
     }
 }

crates/policy/src/policy/function/number.rs

@@ -26,11 +26,12 @@ pub type NumberFunc = InnerFunc<NumberFuncValue>;
 pub struct NumberFuncValue(i64);
 
 impl NumberFunc {
-    pub fn evaluate(&self, op: impl Fn(&i64, &i64) -> bool, if_exists: bool, values: &HashMap<String, Vec<String>>) -> bool {
+    pub fn evaluate(&self, op: impl Fn(&i64, &i64) -> bool, absent: bool, values: &HashMap<String, Vec<String>>) -> bool {
         for inner in self.0.iter() {
             let v = match values.get(inner.key.name().as_str()).and_then(|x| x.first()) {
                 Some(x) => x,
-                None => return if_exists,
+                None if absent => continue,
+                None => return false,
             };
 
             let Ok(rv) = v.parse::<i64>() else {
@@ -146,6 +147,20 @@ mod tests {
         assert_eq!(limit.evaluate(i64::eq, false, &values), equals);
     }
 
+    #[test]
+    fn absent_numeric_key_does_not_skip_other_keys() {
+        let mut function = new_func(S3(S3MaxKeys), None, 30);
+        function
+            .0
+            .extend(new_func(S3(S3ObjectLockRemainingRetentionDays), None, 30).0);
+        let values = HashMap::from([("object-lock-remaining-retention-days".to_string(), vec!["30".to_string()])]);
+        assert!(!function.evaluate(i64::ne, true, &values));
+        assert!(!function.evaluate(i64::gt, true, &values));
+        assert!(function.evaluate(i64::eq, true, &values));
+        function.0.reverse();
+        assert!(!function.evaluate(i64::ne, true, &values));
+    }
+
     #[test]
     fn remaining_retention_days_numeric_condition_without_value_uses_if_exists() {
         let limit = new_func(S3(S3ObjectLockRemainingRetentionDays), None, 30);

crates/policy/src/policy/policy.rs

@@ -3416,6 +3416,277 @@ mod test {
         assert_eq!(round_trip.statements[0].effect, policy.statements[0].effect);
     }
 
+    #[tokio::test]
+    async fn global_wildcard_deny_covers_all_families_without_broadening_allow() {
+        for action in [
+            "s3:GetObject",
+            "s3:ForceDeleteObject",
+            "admin:CreateUser",
+            "sts:AssumeRole",
+            "kms:Configure",
+            "kms:Decrypt",
+        ] {
+            let parsed_action = Action::try_from(action).expect("action parses");
+            let conditions = HashMap::new();
+            let claims = HashMap::new();
+            let args = Args {
+                account: "user",
+                groups: &None,
+                action: parsed_action,
+                bucket: if action.starts_with("s3:") { "bucket" } else { "" },
+                object: if action.starts_with("s3:") { "object" } else { "" },
+                conditions: &conditions,
+                claims: &claims,
+                is_owner: false,
+                deny_only: false,
+            };
+            for (effect, selector, wildcard, expected) in [
+                ("Deny", "Action", "*", false),
+                ("Deny", "Action", "s3:*", !action.starts_with("s3:")),
+                ("Deny", "NotAction", "*", true),
+            ] {
+                let document = serde_json::json!({
+                    "Version": "2012-10-17", "Statement": [
+                        {"Effect": "Allow", "Action": action, "Resource": "*"},
+                        {"Effect": effect, selector: wildcard, "Resource": "*"}
+                    ]
+                });
+                let policy = Policy::parse_config(&serde_json::to_vec(&document).expect("serialize policy"))
+                    .expect("parse wildcard policy");
+                assert_eq!(policy.is_allowed(&args).await, expected, "{effect} {selector} {wildcard} for {action}");
+                let serialized = serde_json::to_value(&policy).expect("serialize policy");
+                assert_eq!(serialized["Statement"][1][selector][0], wildcard);
+            }
+            for selector in ["Action", "NotAction"] {
+                let document = serde_json::json!({
+                    "Version": "2012-10-17", "Statement": [{"Effect": "Allow", selector: "*", "Resource": "*"}]
+                });
+                let policy = Policy::parse_config(&serde_json::to_vec(&document).expect("serialize policy"))
+                    .expect("parse wildcard Allow");
+                assert_eq!(policy.is_allowed(&args).await, selector == "Action" && action == "s3:GetObject");
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn global_wildcard_deny_still_obeys_resource_and_condition_scope() {
+        let document = serde_json::json!({
+            "Version": "2012-10-17", "Statement": [
+                {"Effect": "Allow", "Action": "admin:CreateUser", "Resource": "*"},
+                {"Effect": "Deny", "Action": "*", "Resource": "arn:aws:s3:::bucket/*"}
+            ]
+        });
+        let policy = Policy::parse_config(&serde_json::to_vec(&document).expect("serialize policy")).expect("parse scoped Deny");
+        let claims = HashMap::new();
+        let conditions = HashMap::new();
+        let mut args = Args {
+            account: "user",
+            groups: &None,
+            action: Action::try_from("admin:CreateUser").expect("admin action"),
+            bucket: "",
+            object: "",
+            conditions: &conditions,
+            claims: &claims,
+            is_owner: false,
+            deny_only: false,
+        };
+        assert!(policy.is_allowed(&args).await, "bucket-scoped Deny must not become a global admin Deny");
+        let mut document = document;
+        document["Statement"][1]["Resource"] = serde_json::json!("*");
+        document["Statement"][1]["Condition"] = serde_json::json!({"StringEquals": {"aws:username": "blocked"}});
+        let conditional =
+            Policy::parse_config(&serde_json::to_vec(&document).expect("serialize policy")).expect("parse conditional Deny");
+        assert!(conditional.is_allowed(&args).await);
+        let conditions = HashMap::from([("username".to_string(), vec!["blocked".to_string()])]);
+        args.conditions = &conditions;
+        assert!(!conditional.is_allowed(&args).await);
+    }
+
+    #[tokio::test]
+    async fn security_conditions_enforce_identity_and_bucket_denies() {
+        for (operator, key, policy_value, request_value, denied) in [
+            ("NumericNotEquals", "s3:object-lock-remaining-retention-days", "30", None, true),
+            ("NumericNotEquals", "s3:object-lock-remaining-retention-days", "30", Some("30"), false),
+            ("NumericNotEquals", "s3:object-lock-remaining-retention-days", "30", Some("10"), true),
+            ("DateNotEquals", "s3:object-lock-retain-until-date", "2030-01-01T00:00:00Z", None, true),
+            (
+                "DateNotEquals",
+                "s3:object-lock-retain-until-date",
+                "2030-01-01T00:00:00Z",
+                Some("2030-01-01T00:00:00Z"),
+                false,
+            ),
+            ("DateGreaterThan", "aws:EpochTime", "2020-01-01T00:00:00Z", Some("1577836801"), true),
+            ("DateGreaterThan", "aws:EpochTime", "2020-01-01T00:00:00Z", Some("1577836799"), false),
+            (
+                "NotIpAddress",
+                "aws:SourceIp",
+                "2001:db8:85a3::8a2e:370:7334",
+                Some("2001:db8:ffff::1"),
+                true,
+            ),
+            (
+                "NotIpAddress",
+                "aws:SourceIp",
+                "2001:db8:85a3::8a2e:370:7334",
+                Some("2001:db8:85a3::8a2e:370:7334"),
+                false,
+            ),
+            (
+                "IpAddress",
+                "aws:SourceIp",
+                "2001:db8:85a3::8a2e:370:7334",
+                Some("2001:db8:ffff::1"),
+                false,
+            ),
+            ("IpAddress", "aws:SourceIp", "2001:db8::/32", Some("2001:db8:ffff::1"), true),
+        ] {
+            let mut document = serde_json::json!({
+                "Version": "2012-10-17", "Statement": [
+                    {"Effect": "Allow", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::bucket/*"},
+                    {"Effect": "Deny", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::bucket/*",
+                     "Condition": {operator: {key: policy_value}}}
+                ]
+            });
+            let policy =
+                Policy::parse_config(&serde_json::to_vec(&document).expect("serialize policy")).expect("parse identity policy");
+            let conditions = request_value
+                .map(|value| {
+                    HashMap::from([(key.split_once(':').expect("condition prefix").1.to_string(), vec![value.to_string()])])
+                })
+                .unwrap_or_default();
+            let claims = HashMap::new();
+            let args = Args {
+                account: "user",
+                groups: &None,
+                action: Action::S3Action(S3Action::PutObjectAction),
+                bucket: "bucket",
+                object: "object",
+                conditions: &conditions,
+                claims: &claims,
+                is_owner: false,
+                deny_only: false,
+            };
+            assert_eq!(policy.is_allowed(&args).await, !denied, "identity {operator} {key} {request_value:?}");
+            for statement in document["Statement"].as_array_mut().expect("statements array") {
+                statement["Principal"] = serde_json::json!("*");
+            }
+            let policy: BucketPolicy = serde_json::from_slice(&serde_json::to_vec(&document).expect("serialize bucket policy"))
+                .expect("parse bucket policy");
+            assert_eq!(
+                policy
+                    .is_allowed(&BucketPolicyArgs {
+                        account: "user",
+                        groups: &None,
+                        action: args.action,
+                        bucket: "bucket",
+                        object: "object",
+                        conditions: &conditions,
+                        is_owner: false,
+                    })
+                    .await,
+                !denied,
+                "bucket {operator} {key} {request_value:?}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn service_principal_never_grants_access_to_a_named_access_key() {
+        for principal in [
+            serde_json::json!({"Service": "logging.s3.amazonaws.com"}),
+            serde_json::json!({"Service": "*"}),
+            serde_json::json!({"Service": "logging.s3.amazonaws.com", "AWS": "authorized-user"}),
+        ] {
+            let document = serde_json::json!({
+                "Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Principal": principal,
+                "Action": ["s3:GetObject", "s3:PutObject"], "Resource": "arn:aws:s3:::bucket/*"}]
+            });
+            let policy: BucketPolicy = serde_json::from_slice(&serde_json::to_vec(&document).unwrap()).unwrap();
+            for action in [S3Action::GetObjectAction, S3Action::PutObjectAction] {
+                for account in ["logging.s3.amazonaws.com", "authorized-user"] {
+                    let allowed = policy
+                        .is_allowed(&BucketPolicyArgs {
+                            account,
+                            groups: &None,
+                            action: Action::S3Action(action),
+                            bucket: "bucket",
+                            object: "log",
+                            conditions: &HashMap::new(),
+                            is_owner: false,
+                        })
+                        .await;
+                    assert_eq!(allowed, account == "authorized-user" && principal.get("AWS").is_some());
+                }
+            }
+            assert_eq!(serde_json::to_value(&policy).unwrap()["Statement"][0]["Principal"], principal);
+        }
+    }
+
+    #[tokio::test]
+    async fn directory_objects_obey_prefix_denies_in_identity_and_bucket_policies() {
+        let mut document = serde_json::json!({
+            "Version": "2012-10-17", "Statement": [
+                {"Effect": "Allow", "Action": "s3:*", "Resource": "arn:aws:s3:::bucket/*"},
+                {"Effect": "Deny", "Action": "s3:*", "Resource": "arn:aws:s3:::bucket/private/*"}
+            ]
+        });
+        let identity = Policy::parse_config(&serde_json::to_vec(&document).unwrap()).unwrap();
+        for statement in document["Statement"].as_array_mut().unwrap() {
+            statement["Principal"] = serde_json::json!("*");
+        }
+        let bucket: BucketPolicy = serde_json::from_slice(&serde_json::to_vec(&document).unwrap()).unwrap();
+        for action in [
+            S3Action::GetObjectAction,
+            S3Action::PutObjectAction,
+            S3Action::DeleteObjectAction,
+            S3Action::DeleteObjectVersionAction,
+        ] {
+            for (object, allowed) in [
+                ("private/", false),
+                ("private/child", false),
+                ("public/", true),
+                ("private", true),
+            ] {
+                let conditions = HashMap::new();
+                assert_eq!(
+                    identity
+                        .is_allowed(&Args {
+                            account: "user",
+                            groups: &None,
+                            action: Action::S3Action(action),
+                            bucket: "bucket",
+                            object,
+                            conditions: &conditions,
+                            claims: &HashMap::new(),
+                            is_owner: false,
+                            deny_only: false,
+                        })
+                        .await,
+                    allowed,
+                    "identity {action:?} {object}"
+                );
+                for account in ["user", ""] {
+                    assert_eq!(
+                        bucket
+                            .is_allowed(&BucketPolicyArgs {
+                                account,
+                                groups: &None,
+                                action: Action::S3Action(action),
+                                bucket: "bucket",
+                                object,
+                                conditions: &conditions,
+                                is_owner: false,
+                            })
+                            .await,
+                        allowed,
+                        "bucket {account} {action:?} {object}"
+                    );
+                }
+            }
+        }
+    }
+
     #[tokio::test]
     async fn force_delete_actions_require_an_explicit_grant() {
         let wildcard = Policy::parse_config(

crates/policy/src/policy/principal.rs

@@ -19,7 +19,7 @@ use std::collections::HashSet;
 
 /// Principal that serializes AWS field as single string when containing only "*",
 /// or as an array otherwise (matching AWS S3 API format).
-/// Also supports Service principals (e.g., "logging.s3.amazonaws.com").
+/// Service principals round-trip for compatibility but do not match access keys.
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct Principal {
     aws: HashSet<String>,
@@ -131,11 +131,6 @@ impl Principal {
                 return true;
             }
         }
-        for pattern in self.service.iter() {
-            if wildcard::is_simple_match(pattern, principal) {
-                return true;
-            }
-        }
         false
     }
 }
@@ -225,7 +220,7 @@ mod test {
             service: HashSet::from(["logging.s3.amazonaws.com".to_string()]),
         };
 
-        assert!(principal.is_match("logging.s3.amazonaws.com"));
+        assert!(!principal.is_match("logging.s3.amazonaws.com"));
         assert!(!principal.is_match("replication.s3.amazonaws.com"));
         assert!(!principal.is_match("arn:aws:iam::123456789012:root"));
     }
@@ -238,7 +233,7 @@ mod test {
         };
 
         assert!(principal.is_match("arn:aws:iam::123456789012:root"));
-        assert!(principal.is_match("logging.s3.amazonaws.com"));
+        assert!(!principal.is_match("logging.s3.amazonaws.com"));
         assert!(!principal.is_match("other-principal"));
     }
 }

crates/policy/src/policy/resource.rs

@@ -17,7 +17,11 @@ use serde::{
     Deserialize, Deserializer, Serialize,
     de::{self, Error as DeError, Visitor},
 };
-use std::{collections::HashMap, fmt, ops::Deref};
+use std::{
+    collections::{HashMap, HashSet},
+    fmt,
+    ops::Deref,
+};
 
 use super::{
     Error as IamError, Validator,
@@ -50,12 +54,6 @@ impl ResourceSet {
         self.0.is_empty()
     }
 
-    fn push_unique(&mut self, resource: Resource) {
-        if !self.0.contains(&resource) {
-            self.0.push(resource);
-        }
-    }
-
     pub async fn is_match(&self, resource: &str, conditions: &HashMap<String, Vec<String>>) -> bool {
         self.is_match_with_resolver(resource, conditions, None).await
     }
@@ -130,9 +128,7 @@ impl<'de> Deserialize<'de> for ResourceSet {
                 E: de::Error,
             {
                 let resource = Resource::try_from(value).map_err(|e| E::custom(format!("invalid resource: {}", e)))?;
-                let mut resources = ResourceSet::default();
-                resources.push_unique(resource);
-                Ok(resources)
+                Ok(ResourceSet(vec![resource]))
             }
 
             fn visit_seq<A>(self, mut seq: A) -> std::result::Result<Self::Value, A::Error>
@@ -140,11 +136,14 @@ impl<'de> Deserialize<'de> for ResourceSet {
                 A: de::SeqAccess<'de>,
                 A::Error: DeError,
             {
-                let mut resources = ResourceSet(Vec::with_capacity(seq.size_hint().unwrap_or(0)));
+                let mut resources = ResourceSet::default();
+                let mut seen = HashSet::new();
                 while let Some(value) = seq.next_element::<String>()? {
                     match Resource::try_from(value.as_str()) {
                         Ok(resource) => {
-                            resources.push_unique(resource);
+                            if seen.insert(resource.clone()) {
+                                resources.0.push(resource);
+                            }
                         }
                         Err(e) => {
                             return Err(A::Error::custom(format!("invalid resource: {}", e)));
@@ -224,7 +223,14 @@ impl Resource {
                 return true;
             }
 
-            if cp != "." && wildcard::is_match(resolved_pattern, &cp) {
+            // Keep directory objects distinct from their bare key for prefix
+            // matching, while retaining path cleaning and exact-match compatibility.
+            let wildcard_resource = if matches!(self, Resource::S3(_)) && resource.ends_with('/') && cp.contains('/') {
+                format!("{cp}/")
+            } else {
+                cp.clone()
+            };
+            if cp != "." && wildcard::is_match(resolved_pattern, &wildcard_resource) {
                 return true;
             }
         }
@@ -314,6 +320,20 @@ mod tests {
     use std::collections::HashMap;
     use test_case::test_case;
 
+    #[test]
+    fn large_resource_sets_deduplicate_without_changing_first_seen_order() {
+        let unique = (0..40_000)
+            .map(|index| format!("arn:aws:s3:::bucket/key-{index}"))
+            .collect::<Vec<_>>();
+        let mut repeated = unique.clone();
+        repeated.extend(unique.iter().rev().cloned());
+        let resources: super::ResourceSet = serde_json::from_slice(&serde_json::to_vec(&repeated).expect("encode resources"))
+            .expect("large valid resource list parses");
+        assert_eq!(resources.len(), unique.len());
+        assert_eq!(serde_json::to_value(resources).expect("serialize resources"), serde_json::json!(unique));
+        assert!(serde_json::from_str::<super::ResourceSet>(r#"["arn:aws:s3:::bucket/key","invalid"]"#).is_err());
+    }
+
     #[test_case("*", "mybucket" => true; "all buckets")]
     #[test_case("*", "mybucket/myobject" => true; "all objects")]
     #[test_case("*", "key/mykey" => true; "all keys")]
@@ -335,6 +355,10 @@ mod tests {
     #[test_case("arn:aws:s3:::mybucket","mybucket/myobject" => false; "15")]
     #[test_case("arn:aws:s3:::attacker-bucket/*","attacker-bucket/../victim-bucket/evil.txt" => false; "16")]
     #[test_case("arn:aws:s3:::attacker-bucket/*","attacker-bucket/safe/../../victim-bucket/evil.txt" => false; "17")]
+    #[test_case("arn:aws:s3:::bucket/secret/*","bucket/secret/" => true; "directory prefix")]
+    #[test_case("arn:aws:s3:::bucket/secret","bucket/secret/" => true; "directory exact compatibility")]
+    #[test_case("arn:aws:s3:::bucket/*","bucket/" => false; "bucket is not a directory object")]
+    #[test_case("arn:aws:s3:::attacker-bucket/*","attacker-bucket/../victim-bucket/" => false; "directory traversal")]
     #[test_case("arn:aws:kms:::key/mykey","key/mykey" => true; "kms exact key")]
     #[test_case("arn:aws:kms:::key/mykey","key/otherkey" => false; "kms wrong key")]
     #[test_case("arn:aws:kms:::key/*","key/mykey" => true; "kms key wildcard")]

crates/policy/src/policy/statement.rs

@@ -144,7 +144,7 @@ impl Statement {
 
         for action in self.actions.iter() {
             match action {
-                Action::S3Action(_) => saw_s3 = true,
+                Action::All | Action::S3Action(_) => saw_s3 = true,
                 Action::AdminAction(_) => saw_admin = true,
                 Action::StsAction(_) => saw_sts = true,
                 Action::KmsAction(_) => saw_kms = true,

crates/policy/src/policy/variables.rs

@@ -259,35 +259,11 @@ async fn resolve_aws_variables_with_depth(pattern: &str, resolver: &dyn PolicyVa
     if depth >= 10 {
         return vec![pattern.to_string()];
     }
-    let mut results = vec![pattern.to_string()];
-
-    let mut changed = true;
-    let max_iterations = 10; // Prevent infinite loops
-    let mut iteration = 0;
-
-    while changed && iteration < max_iterations {
-        changed = false;
-        iteration += 1;
-
-        let mut new_results = Vec::new();
-        for result in &results {
-            let resolved = resolve_single_pass(result, resolver, depth).await;
-            if resolved.len() > 1 || (resolved.len() == 1 && &resolved[0] != result) {
-                changed = true;
-            }
-            new_results.extend(resolved);
-        }
-
-        // Remove duplicates while preserving order
-        results.clear();
-        let mut seen = std::collections::HashSet::new();
-        for result in new_results {
-            if seen.insert(result.clone()) {
-                results.push(result);
-            }
-        }
-    }
-
+    // Values from credentials and claims are data, not policy source. Expanding
+    // substituted placeholders again allows attacker-controlled exponential growth.
+    let mut results = resolve_single_pass(pattern, resolver, depth).await;
+    let mut seen = std::collections::HashSet::new();
+    results.retain(|result| seen.insert(result.clone()));
     results
 }
 
@@ -370,8 +346,8 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
                             let suffix = &results[i].0[end_pos + 1..];
 
                             for value in values {
-                                // Resume after the substitution; a generated placeholder
-                                // belongs to the next bounded pass, not this one.
+                                // Resume after the substitution; generated placeholders
+                                // are literal data and must never be expanded.
                                 new_results.push((format!("{prefix}{value}{suffix}"), actual_pos + value.len()));
                             }
 
@@ -445,8 +421,20 @@ mod tests {
             calls: AtomicUsize::new(0),
         };
         let result = resolve_aws_variables("${aws:AccountId}{aws:username}", &resolver).await;
-        assert!(resolver.calls.load(Ordering::Relaxed) <= 10);
-        assert_eq!(result, vec!["${aws:AccountId}{aws:username}".to_string()]);
+        assert_eq!(resolver.calls.load(Ordering::Relaxed), 1);
+        assert_eq!(result, vec!["$${aws:username}".to_string()]);
+    }
+
+    #[tokio::test]
+    async fn substituted_account_id_placeholders_are_literal() {
+        let account = "${aws:AccountId}".repeat(2);
+        let resolver = VariableResolver::new(VariableContext {
+            account_id: Some(account.clone()),
+            username: Some("alice".to_string()),
+            ..Default::default()
+        });
+        let result = resolve_aws_variables("prefix/${aws:AccountId}/${aws:username}", &resolver).await;
+        assert_eq!(result, vec![format!("prefix/{account}/alice")]);
     }
 
     #[tokio::test]

rustfs/src/storage/access.rs

@@ -155,6 +155,19 @@ struct RequestedObjectLockRetainUntil(OffsetDateTime);
 #[derive(Clone, Debug)]
 struct RequestedObjectLockRetention(ObjectLockRetention);
 
+/// Legal hold status a standalone PutObjectLegalHold applies from its XML body.
+#[derive(Clone, Debug)]
+struct RequestedObjectLegalHold(ObjectLockLegalHold);
+
+/// Tag set a standalone PutObjectTagging applies from its XML body.
+#[derive(Clone, Debug)]
+struct RequestedObjectTags(Tagging);
+
+/// Request headers with the object policy inputs a POST upload applies from its
+/// form fields, so header-derived condition keys describe what is stored.
+#[derive(Clone, Debug)]
+struct PostObjectPolicyHeaders(http::HeaderMap);
+
 /// These operations stage upload state without publishing an object. Keep this
 /// separate from the marker controlling multipart publication fences.
 #[derive(Clone, Copy, Debug)]
@@ -925,8 +938,14 @@ fn authorization_conditions<T>(
     client_info: Option<&ClientInfo>,
     action: Action,
 ) -> S3Result<HashMap<String, Vec<String>>> {
+    // A POST upload applies its form fields, not its headers, so the policy
+    // decision must read the values the handler stores.
+    let headers = req
+        .extensions
+        .get::<PostObjectPolicyHeaders>()
+        .map_or(&req.headers, |headers| &headers.0);
     let mut conditions = get_condition_values_with_query_and_client_info(
-        &req.headers,
+        headers,
         cred,
         version_id,
         None,
@@ -951,7 +970,31 @@ fn authorization_conditions<T>(
         conditions.insert("LocationConstraint".to_string(), vec![location]);
     }
     merge_list_bucket_query_conditions(action, req.uri.query(), &mut conditions);
-    merge_request_object_tag_conditions(action, &req.headers, &mut conditions)?;
+    merge_request_object_tag_conditions(action, headers, &mut conditions)?;
+    if let Some(RequestedObjectLegalHold(legal_hold)) = req.extensions.get::<RequestedObjectLegalHold>() {
+        // Standalone legal hold writes apply the XML body, so headers must not
+        // supply a different value to the policy decision.
+        conditions.remove("object-lock-legal-hold");
+        if let Some(status) = &legal_hold.status {
+            conditions.insert("object-lock-legal-hold".to_string(), vec![status.as_str().to_string()]);
+        }
+    }
+    if let Some(RequestedObjectTags(tagging)) = req.extensions.get::<RequestedObjectTags>() {
+        // Standalone tag writes apply the XML body; the x-amz-tagging header is
+        // not stored and must not describe the request to the policy engine.
+        conditions.retain(|key, _| key != "RequestObjectTagKeys" && !key.starts_with("RequestObjectTag/"));
+        let mut keys = Vec::new();
+        for tag in &tagging.tag_set {
+            if let (Some(key), Some(value)) = (&tag.key, &tag.value) {
+                keys.push(key.clone());
+                conditions
+                    .entry(format!("RequestObjectTag/{key}"))
+                    .or_default()
+                    .push(value.clone());
+            }
+        }
+        conditions.insert("RequestObjectTagKeys".to_string(), keys);
+    }
     if let Some(RequestedObjectLockRetention(retention)) = req.extensions.get::<RequestedObjectLockRetention>() {
         // Standalone retention writes apply the XML body, so headers must not
         // supply a different value to the policy decision.
@@ -1007,6 +1050,58 @@ fn set_requested_object_lock_retention(req: &mut S3Request<PutObjectRetentionInp
     req.extensions.insert(RequestedObjectLockRetention(retention));
 }
 
+/// A POST upload stores the object policy inputs from its form fields, while
+/// header-derived condition keys would otherwise read the HTTP headers. Project
+/// the form fields over the headers for the policy decision, removing a header
+/// whose field is absent so a client header cannot stand in for a stored value.
+fn set_post_object_policy_headers(req: &mut S3Request<PutObjectInput>) -> S3Result<()> {
+    if req.method != http::Method::POST {
+        return Ok(());
+    }
+    let input = &req.input;
+    let retain_until = input
+        .object_lock_retain_until_date
+        .as_ref()
+        .map(|date| OffsetDateTime::from(date.clone()).format(&time::format_description::well_known::Rfc3339))
+        .transpose()
+        .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid retain-until date"))?;
+    let mut headers = req.headers.clone();
+    for (key, value) in [
+        ("x-amz-object-lock-mode", input.object_lock_mode.as_ref().map(ObjectLockMode::as_str)),
+        (
+            "x-amz-object-lock-legal-hold",
+            input
+                .object_lock_legal_hold_status
+                .as_ref()
+                .map(ObjectLockLegalHoldStatus::as_str),
+        ),
+        ("x-amz-object-lock-retain-until-date", retain_until.as_deref()),
+        ("x-amz-tagging", input.tagging.as_deref()),
+        ("x-amz-acl", input.acl.as_ref().map(ObjectCannedACL::as_str)),
+        ("x-amz-storage-class", input.storage_class.as_ref().map(StorageClass::as_str)),
+        (
+            "x-amz-server-side-encryption",
+            input.server_side_encryption.as_ref().map(ServerSideEncryption::as_str),
+        ),
+        ("x-amz-server-side-encryption-aws-kms-key-id", input.ssekms_key_id.as_deref()),
+        ("x-amz-server-side-encryption-customer-algorithm", input.sse_customer_algorithm.as_deref()),
+        ("x-amz-grant-full-control", input.grant_full_control.as_deref()),
+        ("x-amz-grant-read", input.grant_read.as_deref()),
+        ("x-amz-grant-read-acp", input.grant_read_acp.as_deref()),
+        ("x-amz-grant-write-acp", input.grant_write_acp.as_deref()),
+    ] {
+        headers.remove(key);
+        if let Some(value) = value {
+            let value = value
+                .parse()
+                .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid form field"))?;
+            headers.insert(key, value);
+        }
+    }
+    req.extensions.insert(PostObjectPolicyHeaders(headers));
+    Ok(())
+}
+
 fn merge_object_lock_remaining_retention_days_condition(
     requested: Option<&RequestedObjectLockRetainUntil>,
     now: OffsetDateTime,
@@ -3465,6 +3560,7 @@ impl S3Access for FS {
             };
             let requested_retain_until = req.input.object_lock_retain_until_date.clone();
             set_requested_object_lock_retain_until(req, requested_retain_until.as_ref());
+            set_post_object_policy_headers(req)?;
             authorize_request(req, Action::S3Action(S3Action::PutObjectAction)).await?;
             req.extensions.insert(bucket_generation?);
 
@@ -3502,6 +3598,8 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn put_object_legal_hold(&self, req: &mut S3Request<PutObjectLegalHoldInput>) -> S3Result<()> {
+        req.extensions
+            .insert(RequestedObjectLegalHold(req.input.legal_hold.clone().unwrap_or_default()));
         let req_info = ext_req_info_mut(&mut req.extensions)?;
         req_info.bucket = Some(req.input.bucket.clone());
         req_info.object = Some(req.input.key.clone());
@@ -3549,6 +3647,7 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn put_object_tagging(&self, req: &mut S3Request<PutObjectTaggingInput>) -> S3Result<()> {
+        req.extensions.insert(RequestedObjectTags(req.input.tagging.clone()));
         let req_info = ext_req_info_mut(&mut req.extensions)?;
         req_info.bucket = Some(req.input.bucket.clone());
         req_info.object = Some(req.input.key.clone());
@@ -3681,8 +3780,8 @@ mod tests {
     };
     use super::{BulkDenialAudit, audit_request_denial};
     use super::{
-        remaining_retention_days, requested_bucket_location_constraint, set_requested_object_lock_retain_until,
-        set_requested_object_lock_retention,
+        RequestedObjectLegalHold, RequestedObjectTags, remaining_retention_days, requested_bucket_location_constraint,
+        set_post_object_policy_headers, set_requested_object_lock_retain_until, set_requested_object_lock_retention,
     };
     use crate::auth::OBJECT_LOCK_REMAINING_RETENTION_DAYS_CONDITION;
     use crate::error::ApiError;
@@ -4812,6 +4911,108 @@ mod tests {
         authorization_conditions(req, &credentials, None, None, None, None, action).expect("conditions should build")
     }
 
+    #[test]
+    fn legal_hold_policy_checks_body_instead_of_conflicting_headers() {
+        let mut req = build_request(
+            PutObjectLegalHoldInput {
+                legal_hold: Some(ObjectLockLegalHold {
+                    status: Some(ObjectLockLegalHoldStatus::from_static(ObjectLockLegalHoldStatus::OFF)),
+                }),
+                ..Default::default()
+            },
+            Method::PUT,
+        );
+        req.headers
+            .insert("x-amz-object-lock-legal-hold", HeaderValue::from_static("ON"));
+        let action = Action::S3Action(S3Action::PutObjectLegalHoldAction);
+        // Without the hook's extension the header leaks into the decision.
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert_eq!(conditions.get("object-lock-legal-hold"), Some(&vec!["ON".to_string()]));
+        req.extensions
+            .insert(RequestedObjectLegalHold(req.input.legal_hold.clone().unwrap_or_default()));
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert_eq!(conditions.get("object-lock-legal-hold"), Some(&vec!["OFF".to_string()]));
+        // A body without a status stores no hold change; the header must not fill it in.
+        req.extensions
+            .insert(RequestedObjectLegalHold(ObjectLockLegalHold::default()));
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert!(!conditions.contains_key("object-lock-legal-hold"));
+    }
+
+    #[test]
+    fn tagging_policy_checks_body_instead_of_conflicting_headers() {
+        let mut req = build_request(
+            PutObjectTaggingInput::builder()
+                .bucket("bucket".to_string())
+                .key("object".to_string())
+                .tagging(Tagging {
+                    tag_set: vec![Tag {
+                        key: Some("classification".to_string()),
+                        value: Some("confidential".to_string()),
+                    }],
+                })
+                .build()
+                .expect("valid tagging request DTO"),
+            Method::PUT,
+        );
+        req.headers
+            .insert("x-amz-tagging", HeaderValue::from_static("classification=public"));
+        let action = Action::S3Action(S3Action::PutObjectTaggingAction);
+        req.extensions.insert(RequestedObjectTags(req.input.tagging.clone()));
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert_eq!(conditions.get("RequestObjectTag/classification"), Some(&vec!["confidential".to_string()]));
+        assert_eq!(conditions.get("RequestObjectTagKeys"), Some(&vec!["classification".to_string()]));
+        req.extensions.insert(RequestedObjectTags(Tagging::default()));
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert!(!conditions.contains_key("RequestObjectTag/classification"));
+        assert_eq!(conditions.get("RequestObjectTagKeys"), Some(&Vec::new()));
+    }
+
+    #[test]
+    fn post_object_policy_checks_form_fields_instead_of_headers() {
+        let mut req = build_request(
+            PutObjectInput {
+                bucket: "bucket".to_string(),
+                key: "object".to_string(),
+                object_lock_legal_hold_status: Some(ObjectLockLegalHoldStatus::from_static(ObjectLockLegalHoldStatus::ON)),
+                object_lock_mode: Some(ObjectLockMode::from_static(ObjectLockMode::COMPLIANCE)),
+                tagging: Some("classification=confidential".to_string()),
+                ..Default::default()
+            },
+            Method::POST,
+        );
+        req.headers
+            .insert("x-amz-object-lock-legal-hold", HeaderValue::from_static("OFF"));
+        req.headers
+            .insert("x-amz-object-lock-mode", HeaderValue::from_static("GOVERNANCE"));
+        req.headers
+            .insert("x-amz-object-lock-retain-until-date", HeaderValue::from_static("2029-01-01T00:00:00Z"));
+        req.headers
+            .insert("x-amz-tagging", HeaderValue::from_static("classification=public"));
+        set_post_object_policy_headers(&mut req).expect("form fields project onto policy headers");
+        let action = Action::S3Action(S3Action::PutObjectAction);
+        let conditions = remaining_retention_days_conditions(&req, action);
+        assert_eq!(conditions.get("object-lock-legal-hold"), Some(&vec!["ON".to_string()]));
+        assert_eq!(conditions.get("object-lock-mode"), Some(&vec!["COMPLIANCE".to_string()]));
+        // The form sets no retain-until date, so the header must not stand in for one.
+        assert!(!conditions.contains_key("object-lock-retain-until-date"));
+        assert_eq!(conditions.get("RequestObjectTag/classification"), Some(&vec!["confidential".to_string()]));
+        // A PUT keeps reading its headers.
+        let mut put = build_request(
+            PutObjectInput {
+                bucket: "bucket".to_string(),
+                key: "object".to_string(),
+                ..Default::default()
+            },
+            Method::PUT,
+        );
+        put.headers
+            .insert("x-amz-object-lock-legal-hold", HeaderValue::from_static("ON"));
+        set_post_object_policy_headers(&mut put).expect("PUT is untouched");
+        let conditions = remaining_retention_days_conditions(&put, action);
+        assert_eq!(conditions.get("object-lock-legal-hold"), Some(&vec!["ON".to_string()]));
+    }
+
     #[tokio::test]
     async fn retention_policy_checks_body_instead_of_conflicting_headers() {
         use rustfs_policy::policy::{Args, Policy};