diff --git a/crates/e2e_test/src/anonymous_access_test.rs b/crates/e2e_test/src/anonymous_access_test.rs
index eb10d1063..25c44fe85 100644
--- a/crates/e2e_test/src/anonymous_access_test.rs
+++ b/crates/e2e_test/src/anonymous_access_test.rs
@@ -170,3 +170,81 @@ async fn test_anonymous_access_allowed_when_restrict_public_buckets_disabled()
     info!("Test passed: anonymous access allowed with RestrictPublicBuckets=false");
     Ok(())
 }
+
+/// A policy granting anonymous `s3:ListBucket` also permits ListObjectVersions.
+/// That grant must still be subject to RestrictPublicBuckets: the versions listing
+/// reaches authorization through a fallback branch, and that branch has to apply the
+/// same public-access gate as a direct grant.
+#[tokio::test]
+#[serial]
+async fn test_anonymous_list_object_versions_denied_when_restrict_public_buckets_enabled()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+    info!("Starting test: anonymous ListObjectVersions denied with RestrictPublicBuckets=true...");
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let bucket_name = "anon-test-restrict-versions";
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket_name).send().await?;
+
+    let policy_json = serde_json::json!({
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "AllowAnonymousListBucket",
+                "Effect": "Allow",
+                "Principal": "*",
+                "Action": ["s3:ListBucket"],
+                "Resource": [format!("arn:aws:s3:::{}", bucket_name)]
+            }
+        ]
+    })
+    .to_string();
+
+    admin_client
+        .put_bucket_policy()
+        .bucket(bucket_name)
+        .policy(&policy_json)
+        .send()
+        .await?;
+
+    admin_client
+        .put_object()
+        .bucket(bucket_name)
+        .key("test.txt")
+        .body(aws_sdk_s3::primitives::ByteStream::from_static(b"hello anonymous"))
+        .send()
+        .await?;
+
+    // Without the public-access block the fallback grant is expected to work.
+    let versions_url = format!("{}/{}?versions=", env.url, bucket_name);
+    let resp = local_http_client().get(&versions_url).send().await?;
+    assert_eq!(
+        resp.status().as_u16(),
+        200,
+        "Anonymous ListObjectVersions should succeed via the s3:ListBucket grant"
+    );
+
+    admin_client
+        .put_public_access_block()
+        .bucket(bucket_name)
+        .public_access_block_configuration(
+            PublicAccessBlockConfiguration::builder()
+                .restrict_public_buckets(true)
+                .build(),
+        )
+        .send()
+        .await?;
+
+    let resp = local_http_client().get(&versions_url).send().await?;
+    assert_eq!(
+        resp.status().as_u16(),
+        403,
+        "Anonymous ListObjectVersions must be denied when RestrictPublicBuckets is true"
+    );
+
+    info!("Test passed: anonymous ListObjectVersions denied with RestrictPublicBuckets=true");
+    Ok(())
+}
diff --git a/crates/policy/src/policy.rs b/crates/policy/src/policy.rs
index 43ae5b72a..0c01b9729 100644
--- a/crates/policy/src/policy.rs
+++ b/crates/policy/src/policy.rs
@@ -29,7 +29,7 @@ pub mod variables;
 pub use action::ActionSet;
 pub use doc::PolicyDoc;
 pub use effect::Effect;
-pub use function::Functions;
+pub use function::{Functions, is_server_derived_condition_key};
 pub use id::ID;
 pub use policy::*;
 pub use principal::Principal;
diff --git a/crates/policy/src/policy/function.rs b/crates/policy/src/policy/function.rs
index 2a598c398..b1fa5d1ee 100644
--- a/crates/policy/src/policy/function.rs
+++ b/crates/policy/src/policy/function.rs
@@ -30,6 +30,36 @@ pub mod key_name;
 pub mod number;
 pub mod string;
 
+/// Set qualifier applied to a condition key that carries multiple request values.
+///
+/// The distinction matters for negated operators (`StringNotEquals`, `StringNotLike`,
+/// `ArnNotEquals`, `ArnNotLike`, ...). For those, negation must be applied to each
+/// request value *before* the quantifier aggregates them; negating the aggregate
+/// instead turns `ForAllValues` into `ForAnyValue` and vice versa.
+#[derive(Clone, Copy, Default, Debug, PartialEq, Eq)]
+pub enum Quantifier {
+    /// No `ForAllValues:`/`ForAnyValue:` prefix. The operator is applied to the value
+    /// set as a whole, matching AWS single-valued-key semantics.
+    #[default]
+    None,
+    /// `ForAnyValue:` — satisfied when at least one request value satisfies the operator.
+    /// Unsatisfiable when the key is absent from the request.
+    ForAnyValue,
+    /// `ForAllValues:` — satisfied when every request value satisfies the operator.
+    /// Vacuously satisfied when the key is absent from the request.
+    ForAllValues,
+}
+
+/// Whether `name` — a condition key with its namespace prefix already stripped, as it
+/// appears in the condition-value map — is one of the well-known keys whose value must
+/// come from server-verified state rather than from request input.
+///
+/// Code that assembles the condition map uses this to refuse request-supplied values
+/// for identity and connection keys. See [`KeyName::is_server_derived`].
+pub fn is_server_derived_condition_key(name: &str) -> bool {
+    KeyName::server_derived_key_names().any(|reserved| reserved.eq_ignore_ascii_case(name))
+}
+
 #[derive(Clone, Default, Debug)]
 pub struct Functions {
     for_any_value: Vec<Condition>,
@@ -48,19 +78,19 @@ impl Functions {
         resolver: Option<&dyn PolicyVariableResolver>,
     ) -> bool {
         for c in self.for_any_value.iter() {
-            if !c.evaluate_with_resolver(false, values, resolver).await {
+            if !c.evaluate_with_resolver(Quantifier::ForAnyValue, values, resolver).await {
                 return false;
             }
         }
 
         for c in self.for_all_values.iter() {
-            if !c.evaluate_with_resolver(true, values, resolver).await {
+            if !c.evaluate_with_resolver(Quantifier::ForAllValues, values, resolver).await {
                 return false;
             }
         }
 
         for c in self.for_normal.iter() {
-            if !c.evaluate_with_resolver(false, values, resolver).await {
+            if !c.evaluate_with_resolver(Quantifier::None, values, resolver).await {
                 return false;
             }
         }
diff --git a/crates/policy/src/policy/function/condition.rs b/crates/policy/src/policy/function/condition.rs
index 53244944d..d80cb9e16 100644
--- a/crates/policy/src/policy/function/condition.rs
+++ b/crates/policy/src/policy/function/condition.rs
@@ -21,6 +21,7 @@ use time::OffsetDateTime;
 
 use super::key_name::KeyName;
 use super::{addr::AddrFunc, binary::BinaryFunc, bool_null::BoolFunc, date::DateFunc, number::NumberFunc, string::StringFunc};
+use crate::policy::function::Quantifier;
 
 #[derive(Clone, Deserialize, Debug)]
 pub enum Condition {
@@ -207,7 +208,7 @@ impl Condition {
 
     pub fn evaluate_with_resolver<'a>(
         &'a self,
-        for_all: bool,
+        quantifier: Quantifier,
         values: &'a HashMap<String, Vec<String>>,
         resolver: Option<&'a dyn PolicyVariableResolver>,
     ) -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send + 'a>> {
@@ -215,16 +216,46 @@ impl Condition {
             use Condition::*;
 
             let r = match self {
-                StringEquals(s) => s.evaluate_with_resolver(for_all, false, false, false, values, resolver).await,
-                StringNotEquals(s) => s.evaluate_with_resolver(for_all, false, false, true, values, resolver).await,
-                StringEqualsIgnoreCase(s) => s.evaluate_with_resolver(for_all, true, false, false, values, resolver).await,
-                StringNotEqualsIgnoreCase(s) => s.evaluate_with_resolver(for_all, true, false, true, values, resolver).await,
-                StringLike(s) => s.evaluate_with_resolver(for_all, false, true, false, values, resolver).await,
-                StringNotLike(s) => s.evaluate_with_resolver(for_all, false, true, true, values, resolver).await,
-                ArnLike(s) => s.evaluate_with_resolver(for_all, false, true, false, values, resolver).await,
-                ArnNotLike(s) => s.evaluate_with_resolver(for_all, false, true, true, values, resolver).await,
-                ArnEquals(s) => s.evaluate_with_resolver(for_all, false, false, false, values, resolver).await,
-                ArnNotEquals(s) => s.evaluate_with_resolver(for_all, false, false, true, values, resolver).await,
+                StringEquals(s) => {
+                    s.evaluate_with_resolver(quantifier, false, false, false, values, resolver)
+                        .await
+                }
+                StringNotEquals(s) => {
+                    s.evaluate_with_resolver(quantifier, false, false, true, values, resolver)
+                        .await
+                }
+                StringEqualsIgnoreCase(s) => {
+                    s.evaluate_with_resolver(quantifier, true, false, false, values, resolver)
+                        .await
+                }
+                StringNotEqualsIgnoreCase(s) => {
+                    s.evaluate_with_resolver(quantifier, true, false, true, values, resolver)
+                        .await
+                }
+                StringLike(s) => {
+                    s.evaluate_with_resolver(quantifier, false, true, false, values, resolver)
+                        .await
+                }
+                StringNotLike(s) => {
+                    s.evaluate_with_resolver(quantifier, false, true, true, values, resolver)
+                        .await
+                }
+                ArnLike(s) => {
+                    s.evaluate_with_resolver(quantifier, false, true, false, values, resolver)
+                        .await
+                }
+                ArnNotLike(s) => {
+                    s.evaluate_with_resolver(quantifier, false, true, true, values, resolver)
+                        .await
+                }
+                ArnEquals(s) => {
+                    s.evaluate_with_resolver(quantifier, false, false, false, values, resolver)
+                        .await
+                }
+                ArnNotEquals(s) => {
+                    s.evaluate_with_resolver(quantifier, false, false, true, values, resolver)
+                        .await
+                }
                 BinaryEquals(s) => s.evaluate(values),
                 IpAddress(s) => s.evaluate(values),
                 NotIpAddress(s) => s.evaluate(values),
@@ -247,7 +278,7 @@ impl Condition {
                     if !inner.has_any_key_in(values) {
                         return true;
                     }
-                    return inner.evaluate_with_resolver(for_all, values, resolver).await;
+                    return inner.evaluate_with_resolver(quantifier, values, resolver).await;
                 }
             };
 
@@ -375,7 +406,7 @@ mod tests {
 
         // "AES256" != "aws:kms" is true, so StringNotEquals should evaluate to true
         assert!(
-            cond.evaluate_with_resolver(false, &values, None).await,
+            cond.evaluate_with_resolver(Quantifier::None, &values, None).await,
             "StringNotEquals should be true when values differ"
         );
 
@@ -383,7 +414,7 @@ mod tests {
 
         // "aws:kms" != "aws:kms" is false, so StringNotEquals should evaluate to false
         assert!(
-            !cond.evaluate_with_resolver(false, &values, None).await,
+            !cond.evaluate_with_resolver(Quantifier::None, &values, None).await,
             "StringNotEquals should be false when values match"
         );
     }
@@ -396,14 +427,14 @@ mod tests {
         values.insert("x-amz-server-side-encryption".to_string(), vec!["aws:kms".to_string()]);
 
         assert!(
-            cond.evaluate_with_resolver(false, &values, None).await,
+            cond.evaluate_with_resolver(Quantifier::None, &values, None).await,
             "StringEquals should be true when values match"
         );
 
         values.insert("x-amz-server-side-encryption".to_string(), vec!["AES256".to_string()]);
 
         assert!(
-            !cond.evaluate_with_resolver(false, &values, None).await,
+            !cond.evaluate_with_resolver(Quantifier::None, &values, None).await,
             "StringEquals should be false when values differ"
         );
     }
@@ -417,7 +448,7 @@ mod tests {
         // Key absent: rvalues is empty, intersection is empty.
         // for_all=false: ivalues.count() > 0 → false. Negated → true.
         assert!(
-            cond.evaluate_with_resolver(false, &values, None).await,
+            cond.evaluate_with_resolver(Quantifier::None, &values, None).await,
             "StringNotEquals should be true when key is absent"
         );
     }
@@ -482,7 +513,7 @@ mod tests {
 
         let values = HashMap::new();
         assert!(
-            cond.evaluate_with_resolver(false, &values, None).await,
+            cond.evaluate_with_resolver(Quantifier::None, &values, None).await,
             "IfExists should return true when the key is absent"
         );
     }
@@ -496,14 +527,14 @@ mod tests {
         values.insert("x-amz-server-side-encryption".to_string(), vec!["aws:kms".to_string()]);
 
         assert!(
-            cond.evaluate_with_resolver(false, &values, None).await,
+            cond.evaluate_with_resolver(Quantifier::None, &values, None).await,
             "IfExists should delegate to inner and return true when values match"
         );
 
         values.insert("x-amz-server-side-encryption".to_string(), vec!["AES256".to_string()]);
 
         assert!(
-            !cond.evaluate_with_resolver(false, &values, None).await,
+            !cond.evaluate_with_resolver(Quantifier::None, &values, None).await,
             "IfExists should delegate to inner and return false when values differ"
         );
     }
diff --git a/crates/policy/src/policy/function/key_name.rs b/crates/policy/src/policy/function/key_name.rs
index ffc16d242..7ad64a7c5 100644
--- a/crates/policy/src/policy/function/key_name.rs
+++ b/crates/policy/src/policy/function/key_name.rs
@@ -112,6 +112,30 @@ impl KeyName {
         &Into::<&str>::into(self)[self.prefix()..]
     }
 
+    /// True when the value of this condition key must come from server-verified
+    /// state -- the authenticated identity, its validated claims, or the connection
+    /// itself -- rather than from raw request input.
+    ///
+    /// Callers assembling the policy condition map use this to refuse request-supplied
+    /// values for these keys. Without that check a caller could satisfy conditions such
+    /// as `aws:userid` or `jwt:groups` simply by sending a header of the same name.
+    ///
+    /// The `s3:` namespace is mixed: `s3:x-amz-*` keys mirror request headers by
+    /// design, while the rest (`s3:authType`, `s3:LocationConstraint`, ...) are
+    /// computed by the server.
+    pub fn is_server_derived(&self) -> bool {
+        match self {
+            KeyName::Aws(_) | KeyName::Jwt(_) | KeyName::Ldap(_) | KeyName::Sts(_) | KeyName::Svc(_) => true,
+            KeyName::S3(_) => !self.name().starts_with("x-amz-"),
+        }
+    }
+
+    /// Names (prefix stripped) of the well-known condition keys that request input
+    /// must never populate. See [`KeyName::is_server_derived`].
+    pub fn server_derived_key_names() -> impl Iterator<Item = &'static str> {
+        Self::COMMON_KEYS.iter().filter(|k| k.is_server_derived()).map(|k| k.name())
+    }
+
     pub fn var_name(&self) -> String {
         match self {
             KeyName::Aws(s) => format!("${{aws:{}}}", Into::<&str>::into(s)),
diff --git a/crates/policy/src/policy/function/string.rs b/crates/policy/src/policy/function/string.rs
index 3cd0ec601..a1ef1a7b6 100644
--- a/crates/policy/src/policy/function/string.rs
+++ b/crates/policy/src/policy/function/string.rs
@@ -25,6 +25,7 @@ use futures::future;
 use serde::{Deserialize, Deserializer, Serialize, de, ser::SerializeSeq};
 
 use super::{func::InnerFunc, key_name::KeyName};
+use crate::policy::function::Quantifier;
 use crate::policy::variables::PolicyVariableResolver;
 
 pub type StringFunc = InnerFunc<StringFuncValue>;
@@ -33,7 +34,7 @@ impl StringFunc {
     #[allow(clippy::too_many_arguments)]
     pub(crate) async fn evaluate_with_resolver(
         &self,
-        for_all: bool,
+        quantifier: Quantifier,
         ignore_case: bool,
         like: bool,
         negate: bool,
@@ -41,10 +42,26 @@ impl StringFunc {
         resolver: Option<&dyn PolicyVariableResolver>,
     ) -> bool {
         for inner in self.0.iter() {
-            let result = if like {
-                inner.eval_like(for_all, values, resolver).await ^ negate
-            } else {
-                inner.eval(for_all, ignore_case, values, resolver).await ^ negate
+            // For `ForAllValues:`/`ForAnyValue:` the negation belongs to the per-value
+            // predicate, so it is pushed down into eval/eval_like. Without a qualifier
+            // the operator applies to the value set as a whole and the aggregate is
+            // negated here, which preserves AWS single-valued-key semantics.
+            let result = match quantifier {
+                Quantifier::None => {
+                    let matched = if like {
+                        inner.eval_like(quantifier, false, values, resolver).await
+                    } else {
+                        inner.eval(quantifier, ignore_case, false, values, resolver).await
+                    };
+                    matched ^ negate
+                }
+                Quantifier::ForAnyValue | Quantifier::ForAllValues => {
+                    if like {
+                        inner.eval_like(quantifier, negate, values, resolver).await
+                    } else {
+                        inner.eval(quantifier, ignore_case, negate, values, resolver).await
+                    }
+                }
             };
 
             if !result {
@@ -59,8 +76,9 @@ impl StringFunc {
 impl FuncKeyValue<StringFuncValue> {
     async fn eval(
         &self,
-        for_all: bool,
+        quantifier: Quantifier,
         ignore_case: bool,
+        negate: bool,
         values: &HashMap<String, Vec<String>>,
         resolver: Option<&dyn PolicyVariableResolver>,
     ) -> bool {
@@ -106,18 +124,20 @@ impl FuncKeyValue<StringFuncValue> {
             .map(|x| if ignore_case { Cow::Owned(x.to_lowercase()) } else { x })
             .collect::<Set<_>>();
 
-        let ivalues = rvalues.intersection(&fvalues);
-
-        if for_all {
-            rvalues.is_empty() || rvalues.len() == ivalues.count()
-        } else {
-            ivalues.count() > 0
+        match quantifier {
+            // Unqualified: the operator applies to the value set as a whole. The caller
+            // negates the aggregate, so report the plain "some value matched" result.
+            Quantifier::None => rvalues.intersection(&fvalues).count() > 0,
+            // Qualified: quantify over the per-value predicate, negation included.
+            Quantifier::ForAllValues => rvalues.iter().all(|v| fvalues.contains(v) ^ negate),
+            Quantifier::ForAnyValue => rvalues.iter().any(|v| fvalues.contains(v) ^ negate),
         }
     }
 
     async fn eval_like(
         &self,
-        for_all: bool,
+        quantifier: Quantifier,
+        negate: bool,
         values: &HashMap<String, Vec<String>>,
         resolver: Option<&dyn PolicyVariableResolver>,
     ) -> bool {
@@ -152,17 +172,32 @@ impl FuncKeyValue<StringFuncValue> {
                     })
                     .any(|x| wildcard::is_match(x, v));
 
-                if for_all {
-                    if !matched {
-                        return false;
+                // Unqualified evaluation reports the plain aggregate match and lets the
+                // caller negate it; the qualifiers fold negation into the per-value
+                // predicate so that ForAllValues/ForAnyValue keep their own meaning.
+                let holds = match quantifier {
+                    Quantifier::None => matched,
+                    Quantifier::ForAllValues | Quantifier::ForAnyValue => matched ^ negate,
+                };
+
+                match quantifier {
+                    Quantifier::ForAllValues => {
+                        if !holds {
+                            return false;
+                        }
+                    }
+                    Quantifier::None | Quantifier::ForAnyValue => {
+                        if holds {
+                            return true;
+                        }
                     }
-                } else if matched {
-                    return true;
                 }
             }
         }
 
-        for_all
+        // No request value settled it: ForAllValues is vacuously satisfied, the other
+        // two are not.
+        quantifier == Quantifier::ForAllValues
     }
 }
 
@@ -243,6 +278,7 @@ impl<'d> Deserialize<'d> for StringFuncValue {
 #[cfg(test)]
 mod tests {
     use super::{StringFunc, StringFuncValue};
+    use crate::policy::function::Quantifier;
     use crate::policy::function::func::FuncKeyValue;
     use crate::policy::function::{
         key::Key,
@@ -304,6 +340,9 @@ mod tests {
         }
     }
 
+    /// `for_all` selects `ForAllValues:`; otherwise the operator is unqualified.
+    /// Mirrors the dispatch in `StringFunc::evaluate_with_resolver` so the tests
+    /// exercise the same negation placement as production.
     fn test_eval(
         s: FuncKeyValue<StringFuncValue>,
         for_all: bool,
@@ -315,9 +354,12 @@ mod tests {
             .into_iter()
             .map(|(k, v)| (k.to_owned(), v.into_iter().map(ToOwned::to_owned).collect::<Vec<String>>()))
             .collect();
-        let result = s.eval(for_all, ignore_case, &map, None);
 
-        pollster::block_on(result) ^ negate
+        if for_all {
+            pollster::block_on(s.eval(Quantifier::ForAllValues, ignore_case, negate, &map, None))
+        } else {
+            pollster::block_on(s.eval(Quantifier::None, ignore_case, false, &map, None)) ^ negate
+        }
     }
 
     #[test]
@@ -381,8 +423,9 @@ mod tests {
     #[test_case(new_fkv("s3:LocationConstraint", vec!["eu-west-1", "ap-southeast-1"]), false, vec![("delimiter", vec!["/"])] => true ; "9")]
     #[test_case(new_fkv("jwt:groups", vec!["prod", "art"]), true, vec![("groups", vec!["prod", "art"])] => false ; "10")]
     #[test_case(new_fkv("jwt:groups", vec!["prod", "art"]), true, vec![("groups", vec!["art"])] => false ; "11")]
-    #[test_case(new_fkv("jwt:groups", vec!["prod", "art"]), true, vec![] => false ; "12")]
-    #[test_case(new_fkv("jwt:groups", vec!["prod", "art"]), true, vec![("delimiter", vec!["/"])] => false ; "13")]
+    // ForAllValues is vacuously satisfied when the key is absent from the request.
+    #[test_case(new_fkv("jwt:groups", vec!["prod", "art"]), true, vec![] => true ; "12")]
+    #[test_case(new_fkv("jwt:groups", vec!["prod", "art"]), true, vec![("delimiter", vec!["/"])] => true ; "13")]
     #[test_case(new_fkv("jwt:groups", vec!["prod", "art"]), false, vec![("groups", vec!["prod", "art"])] => false ; "14")]
     #[test_case(new_fkv("jwt:groups", vec!["prod", "art"]), false, vec![("groups", vec!["art"])] => false ; "15")]
     #[test_case(new_fkv("jwt:groups", vec!["prod", "art"]), false, vec![] => true ; "16")]
@@ -423,8 +466,9 @@ mod tests {
     #[test_case(new_fkv("s3:LocationConstraint", vec!["EU-WEST-1", "AP-southeast-1"]), false, vec![("delimiter", vec!["/"])] => true ; "9")]
     #[test_case(new_fkv("jwt:groups", vec!["Prod", "Art"]), true, vec![("groups", vec!["prod", "art"])] => false ; "10")]
     #[test_case(new_fkv("jwt:groups", vec!["Prod", "Art"]), true, vec![("groups", vec!["art"])] => false ; "11")]
-    #[test_case(new_fkv("jwt:groups", vec!["Prod", "Art"]), true, vec![] => false ; "12")]
-    #[test_case(new_fkv("jwt:groups", vec!["Prod", "Art"]), true, vec![("delimiter", vec!["/"])] => false ; "13")]
+    // ForAllValues is vacuously satisfied when the key is absent from the request.
+    #[test_case(new_fkv("jwt:groups", vec!["Prod", "Art"]), true, vec![] => true ; "12")]
+    #[test_case(new_fkv("jwt:groups", vec!["Prod", "Art"]), true, vec![("delimiter", vec!["/"])] => true ; "13")]
     #[test_case(new_fkv("jwt:groups", vec!["Prod", "Art"]), false, vec![("groups", vec!["prod", "art"])] => false ; "14")]
     #[test_case(new_fkv("jwt:groups", vec!["Prod", "Art"]), false, vec![("groups", vec!["art"])] => false ; "15")]
     #[test_case(new_fkv("jwt:groups", vec!["Prod", "Art"]), false, vec![] => true ; "16")]
@@ -442,9 +486,12 @@ mod tests {
             .into_iter()
             .map(|(k, v)| (k.to_owned(), v.into_iter().map(ToOwned::to_owned).collect::<Vec<String>>()))
             .collect();
-        let result = s.eval_like(for_all, &map, None);
 
-        pollster::block_on(result) ^ negate
+        if for_all {
+            pollster::block_on(s.eval_like(Quantifier::ForAllValues, negate, &map, None))
+        } else {
+            pollster::block_on(s.eval_like(Quantifier::None, false, &map, None)) ^ negate
+        }
     }
 
     #[test_case(new_fkv("s3:x-amz-copy-source", vec!["mybucket/myobject"]), false, vec![("x-amz-copy-source", vec!["mybucket/myobject"])] => true ; "1")]
@@ -481,8 +528,9 @@ mod tests {
     #[test_case(new_fkv("s3:LocationConstraint", vec!["eu-west-*", "ap-southeast-1"]), false, vec![("LocationConstraint", vec!["eu-west-2"])] => false ; "10")]
     #[test_case(new_fkv("jwt:groups", vec!["prod", "art*"]), true, vec![("groups", vec!["prod", "art"])] => false ; "11")]
     #[test_case(new_fkv("jwt:groups", vec!["prod", "art*"]), true, vec![("groups", vec!["art"])] => false ; "12")]
-    #[test_case(new_fkv("jwt:groups", vec!["prod", "art*"]), true, vec![] => false ; "13")]
-    #[test_case(new_fkv("jwt:groups", vec!["prod", "art*"]), true, vec![("delimiter", vec!["/"])] => false ; "14")]
+    // ForAllValues is vacuously satisfied when the key is absent from the request.
+    #[test_case(new_fkv("jwt:groups", vec!["prod", "art*"]), true, vec![] => true ; "13")]
+    #[test_case(new_fkv("jwt:groups", vec!["prod", "art*"]), true, vec![("delimiter", vec!["/"])] => true ; "14")]
     #[test_case(new_fkv("jwt:groups", vec!["prod*", "art"]), false, vec![("groups", vec!["prod", "art"])] => false ; "15")]
     #[test_case(new_fkv("jwt:groups", vec!["prod*", "art"]), false, vec![("groups", vec!["art"])] => false ; "16")]
     #[test_case(new_fkv("jwt:groups", vec!["prod*", "art"]), false, vec![] => true ; "17")]
diff --git a/crates/policy/tests/quantified_negation.rs b/crates/policy/tests/quantified_negation.rs
new file mode 100644
index 000000000..0e54f350a
--- /dev/null
+++ b/crates/policy/tests/quantified_negation.rs
@@ -0,0 +1,100 @@
+//! Regression tests for quantified negated string conditions.
+//!
+//! AWS semantics:
+//!   ForAllValues:StringNotEquals -> true iff EVERY request value is absent from the policy set.
+//!   ForAnyValue:StringNotEquals  -> true iff AT LEAST ONE request value is absent from the policy set.
+//!
+//! Negating the aggregate result of the positive quantified match instead of the
+//! per-value predicate yields NOT(all match) and NOT(any match) respectively,
+//! which transposes the two quantifiers. The discriminating case is a request
+//! whose values partially overlap the policy set; a set that is fully contained
+//! or fully disjoint cannot tell the two apart.
+
+use std::collections::HashMap;
+
+use rustfs_policy::policy::Functions;
+
+fn ctx(key: &str, vals: &[&str]) -> HashMap<String, Vec<String>> {
+    let mut m = HashMap::new();
+    m.insert(key.to_owned(), vals.iter().map(|s| (*s).to_owned()).collect());
+    m
+}
+
+fn eval(json: &str, values: &HashMap<String, Vec<String>>) -> bool {
+    let f: Functions = serde_json::from_str(json).expect("parse condition");
+    pollster::block_on(f.evaluate(values))
+}
+
+#[test]
+fn for_all_values_string_not_equals_partial_overlap() {
+    // groups = {art, hr}; policy set = {prod, art}. "art" IS in the set,
+    // so not every value is "not equal" -> AWS says false.
+    let v = ctx("groups", &["art", "hr"]);
+    let got = eval(r#"{"ForAllValues:StringNotEquals":{"jwt:groups":["prod","art"]}}"#, &v);
+    assert!(
+        !got,
+        "ForAllValues:StringNotEquals must be false when a request value matches the policy set"
+    );
+}
+
+#[test]
+fn for_any_value_string_not_equals_partial_overlap() {
+    // groups = {art, hr}; policy set = {prod, art}. "hr" is NOT in the set,
+    // so at least one value is "not equal" -> AWS says true.
+    let v = ctx("groups", &["art", "hr"]);
+    let got = eval(r#"{"ForAnyValue:StringNotEquals":{"jwt:groups":["prod","art"]}}"#, &v);
+    assert!(
+        got,
+        "ForAnyValue:StringNotEquals must be true when some request value is outside the policy set"
+    );
+}
+
+#[test]
+fn for_all_values_string_not_equals_missing_key() {
+    // ForAllValues is vacuously true when the key is absent from the request.
+    let v = HashMap::new();
+    let got = eval(r#"{"ForAllValues:StringNotEquals":{"jwt:groups":["prod","art"]}}"#, &v);
+    assert!(got, "ForAllValues:* is vacuously true when the context key is absent");
+}
+
+#[test]
+fn for_any_value_string_not_equals_missing_key() {
+    // ForAnyValue is false when the key is absent -- there is no value to satisfy it.
+    let v = HashMap::new();
+    let got = eval(r#"{"ForAnyValue:StringNotEquals":{"jwt:groups":["prod","art"]}}"#, &v);
+    assert!(!got, "ForAnyValue:* is false when the context key is absent");
+}
+
+#[test]
+fn for_all_values_string_not_like_partial_overlap() {
+    // Same inversion for the wildcard family (R03-CAN-042).
+    let v = ctx("groups", &["art-team", "hr"]);
+    let got = eval(r#"{"ForAllValues:StringNotLike":{"jwt:groups":["art-*"]}}"#, &v);
+    assert!(!got, "ForAllValues:StringNotLike must be false when a request value matches the pattern");
+}
+
+#[test]
+fn for_any_value_string_not_like_partial_overlap() {
+    let v = ctx("groups", &["art-team", "hr"]);
+    let got = eval(r#"{"ForAnyValue:StringNotLike":{"jwt:groups":["art-*"]}}"#, &v);
+    assert!(got, "ForAnyValue:StringNotLike must be true when some request value does not match");
+}
+
+/// Positive (non-negated) quantifiers must keep working -- guards against a fix
+/// that flips these instead.
+#[test]
+fn positive_quantifiers_unchanged() {
+    let partial = ctx("groups", &["art", "hr"]);
+    let contained = ctx("groups", &["art"]);
+    let empty = HashMap::new();
+
+    let all_eq = r#"{"ForAllValues:StringEquals":{"jwt:groups":["prod","art"]}}"#;
+    let any_eq = r#"{"ForAnyValue:StringEquals":{"jwt:groups":["prod","art"]}}"#;
+
+    assert!(!eval(all_eq, &partial), "ForAllValues:StringEquals false when a value is outside the set");
+    assert!(eval(all_eq, &contained), "ForAllValues:StringEquals true when all values are in the set");
+    assert!(eval(all_eq, &empty), "ForAllValues:StringEquals vacuously true on absent key");
+
+    assert!(eval(any_eq, &partial), "ForAnyValue:StringEquals true when some value is in the set");
+    assert!(!eval(any_eq, &empty), "ForAnyValue:StringEquals false on absent key");
+}
diff --git a/crates/protocols/src/ftps/driver.rs b/crates/protocols/src/ftps/driver.rs
index 2e00c58d7..ca42cfabd 100644
--- a/crates/protocols/src/ftps/driver.rs
+++ b/crates/protocols/src/ftps/driver.rs
@@ -732,6 +732,12 @@ where
         let (bucket, _key) = parse_s3_path(&path_str)
             .map_err(|e| Error::new(ErrorKind::PermanentFileNotAvailable, format!("{}: {}", "Invalid path", e)))?;
 
+        // MKD creates a bucket, so it has to clear the same authorization boundary as
+        // an S3 CreateBucket call.
+        authorize_operation(session_context, &S3Action::CreateBucket, &bucket, None)
+            .await
+            .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?;
+
         // Create bucket for directory
         match self
             .storage
diff --git a/rustfs/src/auth.rs b/rustfs/src/auth.rs
index ae4cd4a73..2009e4305 100644
--- a/rustfs/src/auth.rs
+++ b/rustfs/src/auth.rs
@@ -20,7 +20,7 @@ use rustfs_iam::error::Error as IamError;
 use rustfs_iam::sys::{
     SESSION_POLICY_NAME, get_claims_from_token_with_secret, get_claims_from_token_with_secret_allow_missing_exp,
 };
-use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive};
+use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive, is_server_derived_condition_key};
 use rustfs_trusted_proxies::ClientInfo;
 use rustfs_utils::MaskedAccessKey;
 use rustfs_utils::http::{AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER};
@@ -744,24 +744,8 @@ pub fn get_condition_values_with_query_and_client_info(
         clone_header.remove(*grant_header);
     }
 
-    for (key, _values) in clone_header.iter() {
-        if key.as_str().eq_ignore_ascii_case("x-amz-tagging") {
-            continue;
-        }
-        if let Some(existing_values) = args.get_mut(key.as_str()) {
-            existing_values.extend(clone_header.get_all(key).iter().map(|v| v.to_str().unwrap_or("").to_string()));
-        } else {
-            args.insert(
-                key.as_str().to_string(),
-                header
-                    .get_all(key)
-                    .iter()
-                    .map(|v| v.to_str().unwrap_or("").to_string())
-                    .collect(),
-            );
-        }
-    }
-
+    // Claims and group membership are part of the verified identity, so they are
+    // resolved before request headers are merged in below.
     if let Some(claims) = &cred.claims {
         for (k, v) in claims {
             if let Some(v_str) = v.as_str() {
@@ -786,9 +770,40 @@ pub fn get_condition_values_with_query_and_client_info(
         args.insert("groups".to_string(), groups.clone());
     }
 
+    // Every remaining header is attacker-controlled. A header must never contribute
+    // to a condition key that describes the caller's own identity or the connection,
+    // otherwise sending `userid: admin` (or any `jwt:`/`ldap:` claim name) would let a
+    // request satisfy a policy condition about itself. Reject those names outright --
+    // both the ones already populated above and the well-known identity keys that are
+    // absent for this credential, since an absent key is exactly what a spoofed header
+    // would fill in.
+    for key in clone_header.keys() {
+        if key.as_str().eq_ignore_ascii_case("x-amz-tagging") {
+            continue;
+        }
+        if is_reserved_condition_key(key.as_str(), &args) {
+            continue;
+        }
+        args.insert(
+            key.as_str().to_string(),
+            header
+                .get_all(key)
+                .iter()
+                .map(|v| v.to_str().unwrap_or("").to_string())
+                .collect(),
+        );
+    }
+
     args
 }
 
+/// Whether a request header is forbidden from contributing to policy condition key
+/// `key`, either because the server already derived that key from verified state or
+/// because it is a well-known identity/context key that only the server may populate.
+fn is_reserved_condition_key(key: &str, server_derived: &HashMap<String, Vec<String>>) -> bool {
+    server_derived.contains_key(key) || is_server_derived_condition_key(key)
+}
+
 /// Get request authentication type
 ///
 /// # Arguments
@@ -1329,6 +1344,62 @@ mod tests {
         assert_eq!(conditions.get("principaltype"), Some(&vec!["AssumedRole".to_string()]));
     }
 
+    #[test]
+    fn test_identity_condition_keys_ignore_spoofed_headers() {
+        let cred = create_test_credentials();
+        let mut headers = HeaderMap::new();
+        // A caller naming its headers after identity condition keys must not be able
+        // to add or replace values the server derives from the credential.
+        headers.insert("userid", "admin".parse().unwrap());
+        headers.insert("username", "admin".parse().unwrap());
+        headers.insert("principaltype", "Account".parse().unwrap());
+        headers.insert("signatureversion", "AWS4-HMAC-SHA256".parse().unwrap());
+
+        let conditions = get_condition_values(&headers, &cred, None, None, None);
+
+        assert_eq!(conditions.get("userid"), Some(&vec!["test-access-key".to_string()]));
+        assert_eq!(conditions.get("username"), Some(&vec!["test-access-key".to_string()]));
+        assert_eq!(conditions.get("principaltype"), Some(&vec!["User".to_string()]));
+        assert!(
+            !conditions
+                .get("signatureversion")
+                .is_some_and(|v| v.iter().any(|s| s == "AWS4-HMAC-SHA256")),
+            "an unsigned request must not gain a signatureversion from a header"
+        );
+    }
+
+    #[test]
+    fn test_claim_condition_keys_ignore_spoofed_headers() {
+        // The credential carries no groups/roles claims, so these keys are absent --
+        // precisely the case a spoofed header would otherwise fill in.
+        let cred = create_test_credentials();
+        let mut headers = HeaderMap::new();
+        headers.insert("groups", "admins".parse().unwrap());
+        headers.insert("roles", "RustFS.ConsoleAdmin".parse().unwrap());
+        headers.insert("sub", "someone-else".parse().unwrap());
+
+        let conditions = get_condition_values(&headers, &cred, None, None, None);
+
+        assert_eq!(conditions.get("groups"), None, "groups must come from the credential only");
+        assert_eq!(conditions.get("roles"), None, "roles must come from claims only");
+        assert_eq!(conditions.get("sub"), None, "jwt claim keys must come from claims only");
+    }
+
+    #[test]
+    fn test_request_headers_still_reach_conditions() {
+        // The reserved list must stay narrow: ordinary request headers, including the
+        // `s3:x-amz-*` condition keys, are still expected to be available to policies.
+        let cred = create_test_credentials();
+        let mut headers = HeaderMap::new();
+        headers.insert("x-amz-content-sha256", "UNSIGNED-PAYLOAD".parse().unwrap());
+        headers.insert("x-amz-server-side-encryption", "AES256".parse().unwrap());
+
+        let conditions = get_condition_values(&headers, &cred, None, None, None);
+
+        assert_eq!(conditions.get("x-amz-content-sha256"), Some(&vec!["UNSIGNED-PAYLOAD".to_string()]));
+        assert_eq!(conditions.get("x-amz-server-side-encryption"), Some(&vec!["AES256".to_string()]));
+    }
+
     #[test]
     fn test_get_condition_values_with_object_lock_headers() {
         let cred = create_test_credentials();
diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs
index 8689bb105..35360763f 100644
--- a/rustfs/src/storage/access.rs
+++ b/rustfs/src/storage/access.rs
@@ -731,6 +731,23 @@ pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3R
             })
             .await;
 
+            // A bucket policy granting s3:ListBucket also covers listing versions. This
+            // fallback has to feed the same post-authorization gates as the direct grant
+            // below, otherwise a public bucket keeps serving anonymous
+            // ListObjectVersions after RestrictPublicBuckets is turned on.
+            let policy_allowed = policy_allowed
+                || (action == Action::S3Action(S3Action::ListBucketVersionsAction)
+                    && PolicySys::is_allowed(&BucketPolicyArgs {
+                        bucket: bucket.as_str(),
+                        action: Action::S3Action(S3Action::ListBucketAction),
+                        is_owner: false,
+                        account: "",
+                        groups: &None,
+                        conditions: &conditions,
+                        object: "",
+                    })
+                    .await);
+
             if policy_allowed {
                 deny_anonymous_table_data_plane_if_needed(action, bucket.as_str(), object.as_str()).await?;
                 // RestrictPublicBuckets: when true, deny public access even if bucket policy allows it.
@@ -747,21 +764,6 @@ pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3R
                 }
                 return Ok(());
             }
-
-            if action == Action::S3Action(S3Action::ListBucketVersionsAction)
-                && PolicySys::is_allowed(&BucketPolicyArgs {
-                    bucket: bucket.as_str(),
-                    action: Action::S3Action(S3Action::ListBucketAction),
-                    is_owner: false,
-                    account: "",
-                    groups: &None,
-                    conditions: &conditions,
-                    object: "",
-                })
-                .await
-            {
-                return Ok(());
-            }
         }
     }