feat(policy): parse and match KMS key resources with a kms:Decrypt action

Bring the dormant Resource::Kms variant to life: arn:aws:kms:::key/<key_id>
patterns (empty-account form, wildcards allowed in the id, alias/<name>
reserved as parse-only) now parse, validate, serialize back, and are matched
by KMS statements against the requested key id carried in Args::object.
Statements without KMS resources keep the legacy match-every-key behaviour,
as do call sites that pass no key resource, so nothing changes until the
admin/SSE authorization paths start passing key ids.

Statement validation rejects KMS resources on non-KMS statements, and bucket
policy validation rejects KMS actions and resources outright while stored
policies keep deserializing; evaluation skips pure-KMS bucket policy
statements with a warning. A kms:Decrypt action is added for the upcoming
SSE-KMS read path.

Refs rustfs/backlog#1582 (part of rustfs/backlog#1562)

crates/policy/src/policy.rs

@@ -72,4 +72,10 @@ pub enum Error {
 
     #[error("invalid resource, type: '{0}', pattern: '{1}'")]
     InvalidResource(String, String),
+
+    #[error("KMS resources require a statement whose actions are all KMS actions")]
+    KmsResourceWithNonKmsAction,
+
+    #[error("bucket policies do not support KMS actions or resources")]
+    KmsUnsupportedInBucketPolicy,
 }

crates/policy/src/policy/action.rs

@@ -728,6 +728,8 @@ pub enum KmsAction {
     ListKeysAction,
     #[strum(serialize = "kms:DescribeKey")]
     DescribeKeyAction,
+    #[strum(serialize = "kms:Decrypt")]
+    DecryptAction,
 }
 
 #[cfg(test)]
@@ -764,6 +766,7 @@ mod tests {
             ("kms:RotateKey", KmsAction::RotateKeyAction),
             ("kms:ListKeys", KmsAction::ListKeysAction),
             ("kms:DescribeKey", KmsAction::DescribeKeyAction),
+            ("kms:Decrypt", KmsAction::DecryptAction),
         ] {
             let action = Action::try_from(raw).expect("Should parse KMS action");
             assert_eq!(action, Action::KmsAction(expected));

crates/policy/src/policy/policy.rs

@@ -1760,6 +1760,391 @@ mod test {
         );
     }
 
+    fn kms_args<'a>(
+        action: Action,
+        key_id: &'a str,
+        conditions: &'a HashMap<String, Vec<String>>,
+        claims: &'a HashMap<String, Value>,
+    ) -> Args<'a> {
+        Args {
+            account: "testuser",
+            groups: &None,
+            action,
+            bucket: "",
+            conditions,
+            is_owner: false,
+            object: key_id,
+            claims,
+            deny_only: false,
+        }
+    }
+
+    #[tokio::test]
+    async fn test_kms_statement_with_key_resource_scopes_by_key() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:DisableKey"],
+      "Resource": ["arn:aws:kms:::key/key-a"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let action = Action::KmsAction(KmsAction::DisableKeyAction);
+
+        assert!(
+            policy.is_allowed(&kms_args(action, "key-a", &conditions, &claims)).await,
+            "the granted key must be allowed"
+        );
+        assert!(
+            !policy.is_allowed(&kms_args(action, "key-b", &conditions, &claims)).await,
+            "a key outside the granted resource must be denied"
+        );
+        assert!(
+            !policy
+                .is_allowed(&kms_args(Action::KmsAction(KmsAction::EnableKeyAction), "key-a", &conditions, &claims))
+                .await,
+            "an action outside the grant must stay denied even for the granted key"
+        );
+        assert!(
+            policy.is_allowed(&kms_args(action, "", &conditions, &claims)).await,
+            "call sites that do not pass a key resource keep the legacy match-every-key behaviour"
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_kms_statement_with_wildcard_key_resource() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:GenerateDataKey"],
+      "Resource": ["arn:aws:kms:::key/app-*"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let action = Action::KmsAction(KmsAction::GenerateDataKeyAction);
+
+        assert!(
+            policy
+                .is_allowed(&kms_args(action, "app-primary", &conditions, &claims))
+                .await
+        );
+        assert!(
+            !policy
+                .is_allowed(&kms_args(action, "backup-primary", &conditions, &claims))
+                .await
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_kms_statement_without_resource_matches_every_key() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:*"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+
+        for key_id in ["", "key-a", "any-other-key"] {
+            assert!(
+                policy
+                    .is_allowed(&kms_args(Action::KmsAction(KmsAction::DisableKeyAction), key_id, &conditions, &claims))
+                    .await,
+                "resource-less KMS statement must keep matching every key (key_id: {key_id:?})"
+            );
+        }
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_kms_deny_with_wildcard_resource_overrides_allow() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:*"],
+      "Resource": ["arn:aws:kms:::key/key-a"]
+    },
+    {
+      "Effect": "Deny",
+      "Action": ["kms:DisableKey"],
+      "Resource": ["arn:aws:kms:::key/*"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+
+        assert!(
+            !policy
+                .is_allowed(&kms_args(Action::KmsAction(KmsAction::DisableKeyAction), "key-a", &conditions, &claims))
+                .await,
+            "a wildcard Deny must override the narrower Allow"
+        );
+        assert!(
+            policy
+                .is_allowed(&kms_args(Action::KmsAction(KmsAction::RotateKeyAction), "key-a", &conditions, &claims))
+                .await,
+            "actions outside the Deny keep the Allow"
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_kms_statement_with_not_resource_excludes_keys() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:DescribeKey"],
+      "NotResource": ["arn:aws:kms:::key/prod-*"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let action = Action::KmsAction(KmsAction::DescribeKeyAction);
+
+        assert!(policy.is_allowed(&kms_args(action, "dev-key", &conditions, &claims)).await);
+        assert!(!policy.is_allowed(&kms_args(action, "prod-key", &conditions, &claims)).await);
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_kms_statement_with_s3_resource_is_treated_as_unscoped() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        // Statements combining KMS actions with S3 resources predate KMS resource
+        // support and were always evaluated as if unscoped. Pin that they still
+        // match every key (a warning is logged during evaluation).
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:DisableKey"],
+      "Resource": ["arn:aws:s3:::somebucket/*"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let action = Action::KmsAction(KmsAction::DisableKeyAction);
+
+        for key_id in ["", "key-a"] {
+            assert!(
+                policy.is_allowed(&kms_args(action, key_id, &conditions, &claims)).await,
+                "malformed KMS statement with S3 resources must keep matching every key (key_id: {key_id:?})"
+            );
+        }
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_kms_alias_resource_parses_but_matches_no_key() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:DescribeKey"],
+      "Resource": ["arn:aws:kms:::alias/app-alias"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let action = Action::KmsAction(KmsAction::DescribeKeyAction);
+
+        assert!(
+            !policy.is_allowed(&kms_args(action, "app-alias", &conditions, &claims)).await,
+            "alias patterns are parse-only until alias resolution lands and must not match key requests"
+        );
+        assert!(
+            policy.is_allowed(&kms_args(action, "", &conditions, &claims)).await,
+            "call sites without a key resource keep the legacy match-every-key behaviour"
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_kms_resource_policy_round_trips() {
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:GenerateDataKey", "kms:Decrypt"],
+      "Resource": ["arn:aws:kms:::key/app-*", "arn:aws:kms:::alias/app-alias"]
+    }
+  ]
+}"#,
+        )
+        .expect("KMS resource policy should parse");
+
+        let json = serde_json::to_string(&policy).expect("policy should serialize");
+        let round_trip = Policy::parse_config(json.as_bytes()).expect("serialized KMS resource policy should re-parse");
+        assert_eq!(round_trip.statements[0].resources, policy.statements[0].resources);
+        assert_eq!(round_trip.statements[0].actions, policy.statements[0].actions);
+
+        let value: serde_json::Value = serde_json::from_str(&json).expect("JSON valid");
+        let resources: Vec<_> = value["Statement"][0]["Resource"]
+            .as_array()
+            .expect("Resource should serialize as an array")
+            .iter()
+            .map(|resource| resource.as_str().expect("Resource entries should be strings"))
+            .collect();
+        assert_eq!(resources, vec!["arn:aws:kms:::key/app-*", "arn:aws:kms:::alias/app-alias"]);
+    }
+
+    #[test]
+    fn test_kms_resource_with_non_kms_action_is_invalid() {
+        for action in ["s3:GetObject", "admin:ServerInfo", "sts:AssumeRole"] {
+            let data = format!(
+                r#"{{
+  "Version": "2012-10-17",
+  "Statement": [
+    {{
+      "Effect": "Allow",
+      "Action": ["{action}"],
+      "Resource": ["arn:aws:kms:::key/key-a"]
+    }}
+  ]
+}}"#
+            );
+
+            let result = Policy::parse_config(data.as_bytes());
+            assert!(
+                matches!(result.as_ref().unwrap_err(), Error::PolicyError(IamError::KmsResourceWithNonKmsAction)),
+                "{action} with a KMS resource should fail with KmsResourceWithNonKmsAction, got: {result:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn test_bucket_policy_with_kms_statement_is_invalid() {
+        for statement in [
+            r#"{"Effect":"Allow","Principal":{"AWS":"*"},"Action":["kms:GenerateDataKey"],"Resource":["arn:aws:s3:::bucket/*"]}"#,
+            r#"{"Effect":"Allow","Principal":{"AWS":"*"},"Action":["s3:GetObject"],"Resource":["arn:aws:kms:::key/key-a"]}"#,
+            r#"{"Effect":"Allow","Principal":{"AWS":"*"},"NotAction":["kms:*"],"Resource":["arn:aws:s3:::bucket/*"]}"#,
+        ] {
+            let data = format!(r#"{{"Version":"2012-10-17","Statement":[{statement}]}}"#);
+            let policy: BucketPolicy =
+                serde_json::from_str(&data).expect("bucket policy with KMS content should still deserialize");
+            let result = policy.is_valid();
+            assert!(
+                matches!(result.as_ref().unwrap_err(), Error::PolicyError(IamError::KmsUnsupportedInBucketPolicy)),
+                "bucket policy statement {statement} should fail with KmsUnsupportedInBucketPolicy, got: {result:?}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn test_stored_bucket_policy_with_kms_statement_loads_and_is_ignored() -> Result<()> {
+        // Policies stored before KMS statements were rejected at validation must keep
+        // deserializing, and their KMS statements must not affect bucket traffic.
+        let bucket_policy: BucketPolicy = serde_json::from_str(
+            r#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Principal": {"AWS": "*"},
+      "Action": ["kms:GenerateDataKey"],
+      "Resource": ["arn:aws:s3:::bucket/*"]
+    },
+    {
+      "Effect": "Allow",
+      "Principal": {"AWS": "*"},
+      "Action": ["s3:GetObject"],
+      "Resource": ["arn:aws:s3:::bucket/*"]
+    }
+  ]
+}"#,
+        )?;
+
+        let conditions = HashMap::new();
+        let args = BucketPolicyArgs {
+            account: "testuser",
+            groups: &None,
+            action: Action::S3Action(crate::policy::action::S3Action::GetObjectAction),
+            bucket: "bucket",
+            conditions: &conditions,
+            is_owner: false,
+            object: "a.txt",
+        };
+        assert!(
+            bucket_policy.is_allowed(&args).await,
+            "the S3 statement must keep working alongside an ignored KMS statement"
+        );
+
+        let put_args = BucketPolicyArgs {
+            account: "testuser",
+            groups: &None,
+            action: Action::S3Action(crate::policy::action::S3Action::PutObjectAction),
+            bucket: "bucket",
+            conditions: &conditions,
+            is_owner: false,
+            object: "a.txt",
+        };
+        assert!(
+            !bucket_policy.is_allowed(&put_args).await,
+            "the ignored KMS statement must not grant anything"
+        );
+
+        Ok(())
+    }
+
     #[test]
     fn test_mixed_action_families_are_invalid_even_with_resource() {
         let data = r#"

crates/policy/src/policy/resource.rs

@@ -40,7 +40,7 @@ impl Serialize for ResourceSet {
         for resource in &self.0 {
             let resource_str = match resource {
                 Resource::S3(value) => format!("{}{}", Resource::S3_PREFIX, value),
-                Resource::Kms(value) => value.clone(),
+                Resource::Kms(value) => format!("{}{}", Resource::KMS_PREFIX, value),
             };
             seq.serialize_element(&resource_str)?;
         }
@@ -171,6 +171,20 @@ pub enum Resource {
 
 impl Resource {
     pub const S3_PREFIX: &'static str = "arn:aws:s3:::";
+    /// KMS ARNs use the same empty-account form as [`Self::S3_PREFIX`]; the suffix is
+    /// `key/<key_id>` (wildcards allowed in the id), `alias/<name>`, or a bare `*`.
+    pub const KMS_PREFIX: &'static str = "arn:aws:kms:::";
+    /// Resource-type segment for key ids; request-side KMS resource strings are
+    /// `key/<key_id>` so they line up with these patterns.
+    pub const KMS_KEY_SEGMENT: &'static str = "key/";
+    /// Resource-type segment reserved for key aliases. Alias patterns parse and
+    /// validate, but requests are always evaluated against `key/<key_id>` strings,
+    /// so an alias pattern matches nothing until alias resolution lands.
+    pub const KMS_ALIAS_SEGMENT: &'static str = "alias/";
+
+    pub fn is_kms(&self) -> bool {
+        matches!(self, Resource::Kms(_))
+    }
 
     pub async fn is_match(&self, resource: &str, conditions: &HashMap<String, Vec<String>>) -> bool {
         self.is_match_with_resolver(resource, conditions, None).await
@@ -228,10 +242,13 @@ impl Resource {
 impl TryFrom<&str> for Resource {
     type Error = Error;
     fn try_from(value: &str) -> std::result::Result<Self, Self::Error> {
-        let Some(value) = value.strip_prefix(Self::S3_PREFIX) else {
+        let resource = if let Some(suffix) = value.strip_prefix(Self::S3_PREFIX) {
+            Resource::S3(suffix.into())
+        } else if let Some(suffix) = value.strip_prefix(Self::KMS_PREFIX) {
+            Resource::Kms(suffix.into())
+        } else {
             return Err(IamError::InvalidResource("unknown".into(), value.into()).into());
         };
-        let resource = Resource::S3(value.into());
 
         resource.is_valid()?;
         Ok(resource)
@@ -248,13 +265,17 @@ impl Validator for Resource {
                 }
             }
             Self::Kms(pattern) => {
-                if pattern.is_empty()
+                // A bare `*` matches every key resource; anything else must carry a
+                // resource-type segment. Key ids never contain separators (the KMS
+                // backends reject them), while alias names may nest ("alias/aws/s3").
+                let well_formed = pattern == "*"
                     || pattern
-                        .char_indices()
-                        .find(|&(_, c)| c == '/' || c == '\\' || c == '.')
-                        .map(|(i, _)| i)
-                        .is_some()
-                {
+                        .strip_prefix(Self::KMS_KEY_SEGMENT)
+                        .is_some_and(|id| !id.is_empty() && !id.contains('/') && !id.contains('\\'))
+                    || pattern
+                        .strip_prefix(Self::KMS_ALIAS_SEGMENT)
+                        .is_some_and(|name| !name.is_empty() && !name.contains('\\'));
+                if !well_formed {
                     return Err(IamError::InvalidResource("kms".into(), pattern.into()).into());
                 }
             }
@@ -270,7 +291,7 @@ impl Serialize for Resource {
     {
         match self {
             Resource::S3(s) => serializer.serialize_str(&format!("{}{}", Self::S3_PREFIX, s)),
-            Resource::Kms(s) => serializer.serialize_str(s),
+            Resource::Kms(s) => serializer.serialize_str(&format!("{}{}", Self::KMS_PREFIX, s)),
         }
     }
 }
@@ -308,9 +329,71 @@ 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: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")]
+    #[test_case("arn:aws:kms:::key/app-*","key/app-primary" => true; "kms key prefix wildcard")]
+    #[test_case("arn:aws:kms:::key/app-*","key/backup-primary" => false; "kms key prefix mismatch")]
+    #[test_case("arn:aws:kms:::key/mykey?","key/mykey1" => true; "kms key question mark")]
+    #[test_case("arn:aws:kms:::*","key/mykey" => true; "kms bare star")]
+    #[test_case("arn:aws:kms:::key/mykey","key/mykey/../otherkey" => false; "kms traversal cleaned")]
+    #[test_case("arn:aws:kms:::alias/myalias","key/myalias" => false; "kms alias never matches key")]
+    #[test_case("arn:aws:kms:::alias/*","key/mykey" => false; "kms alias wildcard never matches key")]
+    #[test_case("arn:aws:kms:::key/mykey","mykey" => false; "kms bare id lacks key segment")]
     fn test_resource_is_match(resource: &str, object: &str) -> bool {
         let resource: Resource = resource.try_into().unwrap();
 
         pollster::block_on(resource.is_match(object, &HashMap::new()))
     }
+
+    #[test_case("arn:aws:kms:::key/mykey" => true; "key id parses")]
+    #[test_case("arn:aws:kms:::key/*" => true; "key wildcard parses")]
+    #[test_case("arn:aws:kms:::key/app-key.v2" => true; "key id with dot parses")]
+    #[test_case("arn:aws:kms:::alias/myalias" => true; "alias parses")]
+    #[test_case("arn:aws:kms:::alias/aws/s3" => true; "nested alias parses")]
+    #[test_case("arn:aws:kms:::*" => true; "bare star parses")]
+    #[test_case("arn:aws:kms:::" => false; "empty suffix rejected")]
+    #[test_case("arn:aws:kms:::key/" => false; "empty key id rejected")]
+    #[test_case("arn:aws:kms:::alias/" => false; "empty alias name rejected")]
+    #[test_case("arn:aws:kms:::mykey" => false; "missing resource type segment rejected")]
+    #[test_case("arn:aws:kms:::key/a/b" => false; "separator in key id rejected")]
+    #[test_case("arn:aws:kms:::key/a\\b" => false; "backslash in key id rejected")]
+    #[test_case("arn:aws:kms:us-east-1:123456789012:key/mykey" => false; "region and account form rejected")]
+    fn test_kms_resource_parse(resource: &str) -> bool {
+        Resource::try_from(resource).is_ok()
+    }
+
+    #[test]
+    fn test_kms_resource_serialization_round_trip() {
+        for raw in [
+            "arn:aws:kms:::key/mykey",
+            "arn:aws:kms:::key/*",
+            "arn:aws:kms:::alias/myalias",
+            "arn:aws:kms:::*",
+        ] {
+            let resource = Resource::try_from(raw).expect("KMS resource should parse");
+            assert!(resource.is_kms());
+
+            let json = serde_json::to_string(&resource).expect("KMS resource should serialize");
+            assert_eq!(json, format!("\"{raw}\""), "serialization must write back the full ARN");
+
+            let round_trip: Resource = serde_json::from_str(&json).expect("serialized KMS resource should deserialize");
+            assert_eq!(round_trip, resource);
+        }
+    }
+
+    #[test]
+    fn test_kms_resource_set_serialization_round_trip() {
+        use crate::policy::resource::ResourceSet;
+
+        let set: ResourceSet =
+            serde_json::from_str(r#"["arn:aws:kms:::key/app-*","arn:aws:kms:::alias/myalias"]"#).expect("set should parse");
+        assert_eq!(set.len(), 2);
+
+        let json = serde_json::to_string(&set).expect("set should serialize");
+        assert_eq!(json, r#"["arn:aws:kms:::key/app-*","arn:aws:kms:::alias/myalias"]"#);
+
+        let round_trip: ResourceSet = serde_json::from_str(&json).expect("serialized set should deserialize");
+        assert_eq!(round_trip, set);
+    }
 }

crates/policy/src/policy/statement.rs

@@ -16,6 +16,7 @@ use super::{
     ActionSet, Args, BucketPolicyArgs, Effect, Error as IamError, Functions, ID, Principal, ResourceSet, Validator,
     action::{Action, S3Action},
     function::key_name::{KeyName, S3KeyName},
+    resource::Resource,
     variables::{VariableContext, VariableResolver},
 };
 use crate::error::{Error, Result};
@@ -173,13 +174,79 @@ impl Statement {
         Some(ActionFamily::Mixed)
     }
 
+    /// Resource scope check for KMS statements, which match `arn:aws:kms:::key/<key_id>`
+    /// patterns instead of the bucket/object path grammar used by S3 statements.
+    ///
+    /// Call-site contract (wired up by the admin/SSE authorization paths): the requested
+    /// key identifier (pre-alias-resolution) travels in `args.object` with `args.bucket`
+    /// left empty. An empty `args.object` means the caller did not scope the request to a
+    /// key, which preserves the legacy match-every-key behaviour.
+    async fn kms_key_scope_matches(&self, args: &Args<'_>, resolver: &VariableResolver) -> bool {
+        let kms_resources: Vec<&Resource> = self.resources.iter().filter(|resource| resource.is_kms()).collect();
+        let kms_not_resources: Vec<&Resource> = self.not_resources.iter().filter(|resource| resource.is_kms()).collect();
+
+        if kms_resources.len() != self.resources.len() || kms_not_resources.len() != self.not_resources.len() {
+            // Statements combining KMS actions with S3 resources predate KMS resource
+            // support and were always evaluated as if unscoped; keep that behaviour
+            // but surface it, since the S3 patterns never constrain key access.
+            tracing::warn!(
+                sid = %self.sid.0,
+                "KMS statement carries non-KMS resources; they are ignored and the statement matches every key"
+            );
+        }
+
+        if kms_resources.is_empty() && kms_not_resources.is_empty() {
+            // No KMS resources: the statement scopes by action only (legacy form).
+            return true;
+        }
+
+        if args.object.is_empty() {
+            // Call sites that do not pass a key resource keep the pre-resource-scoping
+            // behaviour where any key matches.
+            return true;
+        }
+
+        let requested = format!("{}{}", Resource::KMS_KEY_SEGMENT, args.object);
+
+        if !kms_resources.is_empty() {
+            let mut matched = false;
+            for resource in kms_resources {
+                if resource
+                    .is_match_with_resolver(&requested, args.conditions, Some(resolver))
+                    .await
+                {
+                    matched = true;
+                    break;
+                }
+            }
+            if !matched {
+                return false;
+            }
+        }
+
+        for resource in kms_not_resources {
+            if resource
+                .is_match_with_resolver(&requested, args.conditions, Some(resolver))
+                .await
+            {
+                return false;
+            }
+        }
+
+        true
+    }
+
     /// Returns true when this statement would reach `conditions.evaluate_with_resolver` in
-    /// [`Statement::is_allowed`] (including the KMS shortcut path). Does not evaluate conditions.
+    /// [`Statement::is_allowed`] (including the KMS resource path). Does not evaluate conditions.
     pub(crate) async fn request_reaches_condition_eval(&self, args: &Args<'_>, resolver: &VariableResolver) -> bool {
         if (!self.actions.is_match(&args.action) && !self.actions.is_empty()) || self.not_actions.is_match(&args.action) {
             return false;
         }
 
+        if self.is_kms() {
+            return self.kms_key_scope_matches(args, resolver).await;
+        }
+
         let resource = build_resource(
             &args.action,
             args.bucket,
@@ -187,10 +254,6 @@ impl Statement {
             self.conditions.references_key_name(&KeyName::S3(S3KeyName::S3Prefix)),
         );
 
-        if self.is_kms() && (resource == "/" || self.resources.is_empty()) {
-            return true;
-        }
-
         if self.resources.is_empty() && self.not_resources.is_empty() && !self.is_admin() && !self.is_sts() {
             return false;
         }
@@ -273,6 +336,19 @@ impl Validator for Statement {
             return Err(IamError::BothResourceAndNotResource.into());
         }
 
+        // KMS resources only make sense on pure-KMS statements. The reverse
+        // combination (KMS actions with S3 resources) predates KMS resources,
+        // may already be stored, and stays loadable; evaluation treats it as
+        // unscoped and warns.
+        let has_kms_resource = self
+            .resources
+            .iter()
+            .chain(self.not_resources.iter())
+            .any(|resource| resource.is_kms());
+        if has_kms_resource && !matches!(action_family, Some(ActionFamily::Kms)) {
+            return Err(IamError::KmsResourceWithNonKmsAction.into());
+        }
+
         self.actions.is_valid()?;
         self.not_actions.is_valid()?;
         self.resources.is_valid()?;
@@ -323,6 +399,18 @@ pub struct BPStatement {
 impl BPStatement {
     /// Returns true when this statement would reach `conditions.evaluate` in [`BPStatement::is_allowed`].
     pub(crate) async fn request_reaches_condition_eval(&self, args: &BucketPolicyArgs<'_>) -> bool {
+        if !self.actions.is_empty() && self.actions.iter().all(|action| matches!(action, Action::KmsAction(_))) {
+            // Bucket policies cannot grant or deny KMS access; such statements are
+            // rejected at validation but may exist in policies stored before that
+            // check. Skip them so they never influence bucket traffic. Statements
+            // mixing KMS with S3 actions keep evaluating their S3 actions as before.
+            tracing::warn!(
+                sid = %self.sid.0,
+                "ignoring bucket policy statement with KMS actions during evaluation"
+            );
+            return false;
+        }
+
         if !self.principal.is_match(args.account) {
             return false;
         }
@@ -379,6 +467,24 @@ impl Validator for BPStatement {
             return Err(IamError::BothActionAndNotAction.into());
         }
 
+        // Bucket policies govern S3 access; KMS grants belong in identity policies.
+        // Rejected here (PutBucketPolicy) only: deserialization stays permissive so
+        // stored policies from before this check keep loading, and evaluation skips
+        // pure-KMS statements with a warning.
+        let has_kms_action = self
+            .actions
+            .iter()
+            .chain(self.not_actions.iter())
+            .any(|action| matches!(action, Action::KmsAction(_)));
+        let has_kms_resource = self
+            .resources
+            .iter()
+            .chain(self.not_resources.iter())
+            .any(|resource| resource.is_kms());
+        if has_kms_action || has_kms_resource {
+            return Err(IamError::KmsUnsupportedInBucketPolicy.into());
+        }
+
         if self.resources.is_empty() && self.not_resources.is_empty() {
             return Err(IamError::NonResource.into());
         }

crates/policy/tests/policy_eval_proptest.rs

@@ -26,6 +26,12 @@
 //!       policy is also allowed by the widened policy;
 //!   (c) an empty policy denies every non-owner request (default deny).
 //!
+//! Properties (d)-(g) extend the same invariants to KMS key resources
+//! (`arn:aws:kms:::key/<key_id>`, backlog#1582): Deny-first over key scopes,
+//! wildcard/resource-less supersets implying concrete key grants, exact key
+//! scoping without cross-key leaks, and the legacy resource-less
+//! match-every-key compatibility pin.
+//!
 //! Pure evaluation: no IO, no global state, parallel-safe. Statements are built
 //! from JSON exactly like production policies arriving via PutPolicy. Generated
 //! bucket/key/action pools avoid wildcard metacharacters so resource patterns
@@ -47,10 +53,23 @@ const OBJECT_ACTIONS: &[&str] = &[
     "s3:PutObjectTagging",
 ];
 
+/// Key-scoped KMS actions safe to pair with an `arn:aws:kms:::key/<id>` resource.
+const KMS_KEY_ACTIONS: &[&str] = &[
+    "kms:GenerateDataKey",
+    "kms:Decrypt",
+    "kms:DisableKey",
+    "kms:RotateKey",
+    "kms:DescribeKey",
+];
+
 fn statement_json(effect: &str, action: &str, resource: &str) -> String {
     format!(r#"{{"Effect":"{effect}","Action":["{action}"],"Resource":["{resource}"]}}"#)
 }
 
+fn resourceless_statement_json(effect: &str, action: &str) -> String {
+    format!(r#"{{"Effect":"{effect}","Action":["{action}"]}}"#)
+}
+
 fn policy_from_statements(statements: &[String]) -> Policy {
     let json = format!(r#"{{"Version":"2012-10-17","Statement":[{}]}}"#, statements.join(","));
     serde_json::from_str(&json).expect("generated policy JSON should parse")
@@ -88,6 +107,22 @@ fn action_strategy() -> impl Strategy<Value = &'static str> {
     proptest::sample::select(OBJECT_ACTIONS)
 }
 
+/// Strategy: a KMS key id without wildcard metacharacters or separators.
+fn key_id_strategy() -> impl Strategy<Value = String> {
+    "[a-z][a-z0-9-]{2,11}"
+}
+
+/// Strategy: one action name from the key-scoped KMS action pool.
+fn kms_action_strategy() -> impl Strategy<Value = &'static str> {
+    proptest::sample::select(KMS_KEY_ACTIONS)
+}
+
+/// KMS evaluation contract: the requested key id travels in `args.object` with an
+/// empty bucket (see `Statement::kms_key_scope_matches`).
+fn is_allowed_for_key(policy: &Policy, action: &str, key_id: &str) -> bool {
+    is_allowed(policy, action, "", key_id)
+}
+
 proptest! {
     /// (a) Deny anywhere wins: a Deny statement matching the request denies it,
     /// no matter how many broad Allow statements surround it or at which index
@@ -205,4 +240,122 @@ proptest! {
             "Policy::default() must deny {action} on {bucket}/{key}"
         );
     }
+
+    /// (d) KMS Deny anywhere wins: a Deny scoped to the exact key (or `key/*`)
+    /// denies the request no matter how many broad KMS Allow statements
+    /// (resource-less or `key/*`-scoped) surround it.
+    #[test]
+    fn kms_explicit_deny_anywhere_denies(
+        key_id in key_id_strategy(),
+        action in kms_action_strategy(),
+        allow_count in 0usize..4,
+        deny_pos_seed in 0usize..16,
+        broad in proptest::bool::ANY,
+        wildcard_deny in proptest::bool::ANY,
+    ) {
+        let mut statements: Vec<String> = (0..allow_count)
+            .map(|_| {
+                if broad {
+                    resourceless_statement_json("Allow", "kms:*")
+                } else {
+                    statement_json("Allow", "kms:*", "arn:aws:kms:::key/*")
+                }
+            })
+            .collect();
+
+        let deny_resource = if wildcard_deny {
+            "arn:aws:kms:::key/*".to_string()
+        } else {
+            format!("arn:aws:kms:::key/{key_id}")
+        };
+        let deny = statement_json("Deny", action, &deny_resource);
+        let deny_pos = deny_pos_seed % (statements.len() + 1);
+        statements.insert(deny_pos, deny);
+
+        let policy = policy_from_statements(&statements);
+
+        if allow_count > 0 {
+            let mut allows_only = statements.clone();
+            allows_only.remove(deny_pos);
+            let allow_policy = policy_from_statements(&allows_only);
+            prop_assert!(
+                is_allowed_for_key(&allow_policy, action, &key_id),
+                "sanity: the KMS Allow statements alone should permit {action} on key {key_id}"
+            );
+        }
+
+        prop_assert!(
+            !is_allowed_for_key(&policy, action, &key_id),
+            "explicit KMS Deny at index {deny_pos} of {} statements must deny {action} on key {key_id}",
+            statements.len()
+        );
+    }
+
+    /// (e) KMS wildcard superset implies the concrete key grant: whatever an
+    /// exact `key/<id>` Allow permits is also permitted by `key/*`, by a bare
+    /// `arn:aws:kms:::*`, and by the legacy resource-less statement form.
+    #[test]
+    fn kms_wildcard_superset_implies_concrete_match(
+        key_id in key_id_strategy(),
+        action in kms_action_strategy(),
+    ) {
+        let narrow = policy_from_statements(&[statement_json(
+            "Allow",
+            action,
+            &format!("arn:aws:kms:::key/{key_id}"),
+        )]);
+        let widened = policy_from_statements(&[statement_json("Allow", "kms:*", "arn:aws:kms:::key/*")]);
+        let star = policy_from_statements(&[statement_json("Allow", "kms:*", "arn:aws:kms:::*")]);
+        let resourceless = policy_from_statements(&[resourceless_statement_json("Allow", "kms:*")]);
+
+        prop_assert!(is_allowed_for_key(&narrow, action, &key_id), "narrow KMS policy must allow its own grant");
+        prop_assert!(is_allowed_for_key(&widened, action, &key_id), "kms:* on key/* must imply the concrete grant");
+        prop_assert!(is_allowed_for_key(&star, action, &key_id), "kms:* on arn:aws:kms:::* must imply the concrete grant");
+        prop_assert!(
+            is_allowed_for_key(&resourceless, action, &key_id),
+            "the legacy resource-less KMS statement must imply the concrete grant"
+        );
+    }
+
+    /// (f) Key scoping is exact: an Allow on `key/<a>` never leaks to a
+    /// different key id, while the compatibility contract keeps unscoped
+    /// requests (no key id passed) matching.
+    #[test]
+    fn kms_key_scope_does_not_leak_across_keys(
+        key_a in key_id_strategy(),
+        key_b in key_id_strategy(),
+        action in kms_action_strategy(),
+    ) {
+        prop_assume!(key_a != key_b);
+
+        let policy = policy_from_statements(&[statement_json(
+            "Allow",
+            action,
+            &format!("arn:aws:kms:::key/{key_a}"),
+        )]);
+
+        prop_assert!(is_allowed_for_key(&policy, action, &key_a), "the granted key must be allowed");
+        prop_assert!(
+            !is_allowed_for_key(&policy, action, &key_b),
+            "an Allow scoped to key {key_a} must not leak to key {key_b}"
+        );
+        prop_assert!(
+            is_allowed_for_key(&policy, action, ""),
+            "call sites that pass no key resource keep the legacy match-every-key behaviour"
+        );
+    }
+
+    /// (g) Legacy compatibility pin: resource-less KMS statements match every
+    /// generated key id, exactly as before KMS resources existed.
+    #[test]
+    fn kms_resourceless_statement_matches_every_key(
+        key_id in key_id_strategy(),
+        action in kms_action_strategy(),
+    ) {
+        let policy = policy_from_statements(&[resourceless_statement_json("Allow", action)]);
+        prop_assert!(
+            is_allowed_for_key(&policy, action, &key_id),
+            "resource-less KMS statement must keep matching {action} on key {key_id}"
+        );
+    }
 }