feat(admin): scope KMS admin authorization to the target key (#5533)

rustfs/src/admin/auth.rs

@@ -46,6 +46,20 @@ impl<'a> AdminResourceScope<'a> {
     pub fn bucket_object(bucket: &'a str, object: &'a str) -> Self {
         Self { bucket, object }
     }
+
+    /// Scope an admin request to a single KMS key.
+    ///
+    /// The policy crate carries the requested key identifier in the object slot
+    /// with an empty bucket (see `Statement::kms_key_scope_matches`); `Args` has
+    /// no dedicated field for it. An empty `key_id` means the caller could not
+    /// name a target key, which keeps the pre-resource-scoping behaviour where a
+    /// KMS statement matches every key.
+    pub fn kms_key(key_id: &'a str) -> Self {
+        Self {
+            bucket: "",
+            object: key_id,
+        }
+    }
 }
 
 pub async fn validate_admin_request(
@@ -174,6 +188,34 @@ pub async fn validate_admin_request_with_bucket_object(
     evaluate_admin_actions(iam_store, &ctx, &actions, resource.bucket, resource.object).await
 }
 
+/// Admin gate for KMS endpoints that act on one key.
+///
+/// `key_id` is the identifier as requested (before any alias resolution), so a
+/// policy scoped to `arn:aws:kms:::key/<id>` only authorizes that key. Endpoints
+/// without a target key pass `""` and stay unscoped, which is also what a
+/// malformed request resolves to: the request is rejected on its own parse error
+/// right after the gate, so no key is ever touched under an unscoped decision.
+pub async fn validate_admin_request_with_kms_key(
+    headers: &HeaderMap,
+    cred: &Credentials,
+    is_owner: bool,
+    deny_only: bool,
+    actions: Vec<Action>,
+    remote_addr: Option<std::net::SocketAddr>,
+    key_id: &str,
+) -> S3Result<()> {
+    validate_admin_request_with_bucket_object(
+        headers,
+        cred,
+        is_owner,
+        deny_only,
+        actions,
+        remote_addr,
+        AdminResourceScope::kms_key(key_id),
+    )
+    .await
+}
+
 /// Unified authentication request handler for both UI and CLI
 ///
 /// This function provides a single entry point for authentication,
@@ -248,12 +290,13 @@ mod tests {
     use rustfs_iam::manager::IamCache;
     use rustfs_iam::store::{GroupInfo, MappedPolicy, UserType};
     use rustfs_policy::auth::UserIdentity;
-    use rustfs_policy::policy::PolicyDoc;
-    use rustfs_policy::policy::action::AdminAction;
+    use rustfs_policy::policy::action::{AdminAction, KmsAction};
+    use rustfs_policy::policy::{Policy, PolicyDoc};
     use serde::Serialize;
     use serde::de::DeserializeOwned;
     use std::collections::HashMap;
     use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU8, AtomicU64};
+    use time::OffsetDateTime;
 
     /// A `Store` whose methods are never invoked by these tests: the gate only
     /// reaches the persistence layer for non-owner principals, and the
@@ -383,14 +426,14 @@ mod tests {
         }
     }
 
-    /// Build an `IamSys` backed by an empty in-memory cache. The cache is
-    /// constructed directly (all fields are public, mirroring the iam crate's
-    /// own `build_test_iam_cache`) so no cluster-backed `ObjectStore` or disk
-    /// load is required. Readiness state is irrelevant: the gate calls
-    /// `is_allowed` directly, which does not gate on `is_ready`.
-    fn iam_sys_with_empty_store() -> Arc<IamSys<EmptyStore>> {
+    /// An `IamCache` over an empty in-memory snapshot. It is constructed
+    /// directly (all fields are public, mirroring the iam crate's own
+    /// `build_test_iam_cache`) so no cluster-backed `ObjectStore` or disk load is
+    /// required. Readiness state is irrelevant: the gate calls `is_allowed`
+    /// directly, which does not gate on `is_ready`.
+    fn iam_cache_with_empty_store() -> IamCache<EmptyStore> {
         let (send_chan, _rx) = tokio::sync::mpsc::channel::<i64>(1);
-        let cache = IamCache {
+        IamCache {
             cache: Cache::default(),
             api: EmptyStore,
             state: Arc::new(AtomicU8::new(0)),
@@ -401,10 +444,47 @@ mod tests {
             sync_failures: AtomicU64::new(0),
             sync_successes: AtomicU64::new(0),
             last_sync_duration_millis: AtomicU64::new(0),
-        };
-        Arc::new(IamSys::new(Arc::new(cache)))
+        }
     }
 
+    fn iam_sys_with_empty_store() -> Arc<IamSys<EmptyStore>> {
+        Arc::new(IamSys::new(Arc::new(iam_cache_with_empty_store())))
+    }
+
+    /// Same in-memory `IamSys`, with one regular (non-owner, non-temp,
+    /// non-service) account carrying `document` as its only attached policy. All
+    /// three cache entries the evaluation path consults are pre-seeded, so no
+    /// `EmptyStore` method is ever reached.
+    fn iam_sys_with_policy(account: &str, document: &str) -> Arc<IamSys<EmptyStore>> {
+        let store = iam_cache_with_empty_store();
+        let now = OffsetDateTime::now_utc();
+        let cache = &store.cache;
+
+        cache.add_or_update_user(
+            account,
+            &UserIdentity::new(Credentials {
+                access_key: account.to_string(),
+                secret_key: "kms-operator-secret".to_string(),
+                status: "on".to_string(),
+                ..Default::default()
+            }),
+            now,
+        );
+        cache.add_or_update_policy_doc(
+            KMS_TEST_POLICY_NAME,
+            &PolicyDoc {
+                policy: Policy::parse_config(document.as_bytes()).expect("test policy should parse"),
+                ..Default::default()
+            },
+            now,
+        );
+        cache.add_or_update_user_policy(account, &MappedPolicy::new(KMS_TEST_POLICY_NAME), now);
+
+        Arc::new(IamSys::new(Arc::new(store)))
+    }
+
+    const KMS_TEST_POLICY_NAME: &str = "kms-key-a-only";
+
     fn ctx_for<'a>(headers: &'a HeaderMap, cred: &'a Credentials, is_owner: bool) -> AuthContext<'a> {
         AuthContext {
             headers,
@@ -471,6 +551,87 @@ mod tests {
         assert_access_denied(res);
     }
 
+    /// KMS scoping rides the object slot with an empty bucket, matching the
+    /// contract the policy crate evaluates KMS statements against.
+    #[test]
+    fn kms_scope_carries_the_key_id_in_the_object_slot() {
+        let scope = AdminResourceScope::kms_key("key-a");
+        assert_eq!(scope.bucket, "");
+        assert_eq!(scope.object, "key-a");
+
+        let unscoped = AdminResourceScope::kms_key("");
+        assert_eq!(unscoped.object, "", "an absent key id must stay unscoped");
+    }
+
+    const KMS_OPERATOR: &str = "kms-operator";
+
+    fn kms_disable_key_actions() -> Vec<Action> {
+        vec![Action::KmsAction(KmsAction::DisableKeyAction)]
+    }
+
+    async fn gate_for_key(iam: Arc<IamSys<EmptyStore>>, key_id: &str, is_owner: bool) -> S3Result<()> {
+        let headers = HeaderMap::new();
+        let cred = Credentials {
+            access_key: KMS_OPERATOR.to_string(),
+            secret_key: "kms-operator-secret".to_string(),
+            status: "on".to_string(),
+            ..Default::default()
+        };
+        let ctx = ctx_for(&headers, &cred, is_owner);
+        let scope = AdminResourceScope::kms_key(key_id);
+
+        evaluate_admin_actions(iam, &ctx, &kms_disable_key_actions(), scope.bucket, scope.object).await
+    }
+
+    /// End-to-end through the gate: a `kms:DisableKey` grant limited to `key/A`
+    /// authorizes key A and denies key B with `AccessDenied`. This is the seam the
+    /// per-handler tests cannot reach — it pins that the scope actually travels
+    /// into the policy verdict rather than being dropped on the way
+    /// (rustfs/backlog#1582).
+    #[tokio::test]
+    async fn kms_scoped_gate_denies_a_key_outside_the_policy_scope() {
+        let document = r#"{"Version":"2012-10-17","Statement":[
+            {"Effect":"Allow","Action":["kms:DisableKey"],"Resource":["arn:aws:kms:::key/key-a"]}
+        ]}"#;
+
+        let allowed = gate_for_key(iam_sys_with_policy(KMS_OPERATOR, document), "key-a", false).await;
+        assert!(allowed.is_ok(), "the key the policy names must pass the gate");
+
+        assert_access_denied(gate_for_key(iam_sys_with_policy(KMS_OPERATOR, document), "key-b", false).await);
+    }
+
+    /// Compatibility pins for the same gate: a resource-less KMS statement keeps
+    /// authorizing every key, and the owner short-circuit is unaffected by scoping.
+    #[tokio::test]
+    async fn kms_scoped_gate_keeps_unscoped_policies_and_owners_unrestricted() {
+        let unscoped = r#"{"Version":"2012-10-17","Statement":[
+            {"Effect":"Allow","Action":["kms:DisableKey"]}
+        ]}"#;
+        let scoped = r#"{"Version":"2012-10-17","Statement":[
+            {"Effect":"Allow","Action":["kms:DisableKey"],"Resource":["arn:aws:kms:::key/key-a"]}
+        ]}"#;
+
+        for key in ["key-a", "key-b"] {
+            let res = gate_for_key(iam_sys_with_policy(KMS_OPERATOR, unscoped), key, false).await;
+            assert!(res.is_ok(), "a resource-less KMS statement must keep authorizing {key}");
+
+            let res = gate_for_key(iam_sys_with_policy(KMS_OPERATOR, scoped), key, true).await;
+            assert!(res.is_ok(), "the owner short-circuit must still authorize {key}");
+        }
+    }
+
+    /// An endpoint with no target key passes an empty scope, which must behave
+    /// exactly like the unscoped gate it replaced.
+    #[tokio::test]
+    async fn kms_gate_without_a_target_key_matches_every_key() {
+        let scoped = r#"{"Version":"2012-10-17","Statement":[
+            {"Effect":"Allow","Action":["kms:DisableKey"],"Resource":["arn:aws:kms:::key/key-a"]}
+        ]}"#;
+
+        let res = gate_for_key(iam_sys_with_policy(KMS_OPERATOR, scoped), "", false).await;
+        assert!(res.is_ok(), "an absent key id must keep the pre-resource-scoping verdict");
+    }
+
     /// The multi-action loop authorizes as soon as one candidate action passes
     /// (owner short-circuits every action), and denies when none pass.
     #[tokio::test]

rustfs/src/admin/handlers/kms_key_lifecycle.rs

@@ -14,8 +14,8 @@
 
 //! KMS key lifecycle admin API handlers: enable, disable and rotate.
 
-use super::kms_keys::extract_query_params;
-use crate::admin::auth::validate_admin_request;
+use super::kms_keys::{extract_query_params, scoped_key_id};
+use crate::admin::auth::validate_admin_request_with_kms_key;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::current_kms_runtime_service_manager;
 use crate::auth::{check_key_valid, get_session_token};
@@ -252,21 +252,27 @@ async fn handle_lifecycle_request(
 
     let (cred, owner) = check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
-    validate_admin_request(
+    // The body (or the `keyId` query parameter) names the key this operation acts
+    // on, so it has to be resolved before the gate runs. A read failure is
+    // surfaced only afterwards so an unauthorized caller still sees AccessDenied.
+    let body = req.input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await;
+
+    validate_admin_request_with_kms_key(
         &req.headers,
         &cred,
         owner,
         false,
         actions,
         req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+        body.as_ref()
+            .ok()
+            .and_then(|body| scoped_key_id(body, &req.uri))
+            .as_deref()
+            .unwrap_or_default(),
     )
     .await?;
 
-    let body = req
-        .input
-        .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
-        .await
-        .map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))?;
+    let body = body.map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))?;
 
     let request: KmsKeyLifecycleRequest = if body.is_empty() {
         let query_params = extract_query_params(&req.uri);
@@ -335,6 +341,7 @@ impl Operation for RotateKmsKeyHandler {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use http::Uri;
     use rustfs_kms::backends::local::LocalKmsBackend;
     use rustfs_kms::config::KmsConfig;
     use rustfs_kms::types::{CreateKeyRequest, KeyState};
@@ -447,6 +454,47 @@ mod tests {
         }
     }
 
+    /// The lifecycle gate must be scoped to the key the request names, and the
+    /// key must be resolved the same way for authorization and for execution.
+    #[test]
+    fn lifecycle_requests_authorize_against_the_key_they_operate_on() {
+        let src = include_str!("kms_key_lifecycle.rs");
+        let handler = src
+            .split_once("async fn handle_lifecycle_request(")
+            .expect("lifecycle entry point should exist")
+            .1;
+        let handler = &handler[..handler.find("\nfn kms_service_manager_from_context").unwrap_or(handler.len())];
+
+        assert!(
+            handler.contains("validate_admin_request_with_kms_key("),
+            "lifecycle requests must scope authorization to the requested key"
+        );
+        assert!(
+            handler.contains("scoped_key_id(body, &req.uri)"),
+            "authorization must resolve the key exactly as execution does"
+        );
+
+        let uri: Uri = "/rustfs/admin/v3/kms/keys/disable?keyId=query-key"
+            .parse()
+            .expect("uri should parse");
+        let body = br#"{"key_id":"body-key"}"#;
+
+        assert_eq!(scoped_key_id(body, &uri).as_deref(), Some("body-key"));
+        assert_eq!(
+            scoped_key_id(body, &uri).as_deref(),
+            Some(
+                serde_json::from_slice::<KmsKeyLifecycleRequest>(body)
+                    .expect("lifecycle body should parse")
+                    .key_id
+                    .as_str()
+            )
+        );
+        assert_eq!(
+            scoped_key_id(b"", &uri).as_deref(),
+            extract_query_params(&uri).get("keyId").map(String::as_str)
+        );
+    }
+
     #[test]
     fn lifecycle_request_rejects_unknown_fields() {
         let error = serde_json::from_str::<KmsKeyLifecycleRequest>(r#"{"key_id":"key","unexpected_field":true}"#)

rustfs/src/admin/handlers/kms_keys.rs

@@ -14,7 +14,7 @@
 
 //! KMS key management admin API handlers
 
-use crate::admin::auth::validate_admin_request;
+use crate::admin::auth::{validate_admin_request, validate_admin_request_with_kms_key};
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{current_kms_runtime_service_manager, current_or_init_kms_runtime_service_manager};
 use crate::auth::{check_key_valid, get_session_token};
@@ -116,6 +116,29 @@ fn extract_key_id(uri: &hyper::Uri) -> Option<String> {
         .find_map(|name| query_params.get(name).filter(|value| !value.is_empty()).cloned())
 }
 
+/// The `key_id` of a KMS admin request body, read without committing to the
+/// strict schema of the endpoint: the authorization gate needs the target key
+/// before the body is parsed for execution, and a body that fails the strict
+/// parse is rejected right after the gate anyway.
+#[derive(Deserialize)]
+struct KeyIdProbe {
+    #[serde(default)]
+    key_id: String,
+}
+
+/// Target key of an endpoint that accepts either a JSON body or a `keyId` query
+/// parameter, resolved exactly the way the endpoint resolves it for execution so
+/// the authorized key can never differ from the operated one. Returns `None`
+/// when the target cannot be determined, which authorizes unscoped and lets the
+/// request fail on its own parse error afterwards.
+pub(super) fn scoped_key_id(body: &[u8], uri: &hyper::Uri) -> Option<String> {
+    if body.is_empty() {
+        return extract_query_params(uri).get("keyId").cloned();
+    }
+
+    serde_json::from_slice::<KeyIdProbe>(body).ok().map(|probe| probe.key_id)
+}
+
 fn kms_service_manager_from_context() -> Option<std::sync::Arc<rustfs_kms::KmsServiceManager>> {
     current_kms_runtime_service_manager()
 }
@@ -279,17 +302,20 @@ impl Operation for DescribeKeyHandler {
         let (cred, owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
-        validate_admin_request(
+        let requested_key_id = extract_key_id(&req.uri);
+
+        validate_admin_request_with_kms_key(
             &req.headers,
             &cred,
             owner,
             false,
             kms_describe_key_actions(),
             req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            requested_key_id.as_deref().unwrap_or_default(),
         )
         .await?;
 
-        let Some(key_id) = extract_key_id(&req.uri) else {
+        let Some(key_id) = requested_key_id else {
             return Err(s3_error!(InvalidRequest, "missing required parameter: 'keyId'"));
         };
 
@@ -335,10 +361,12 @@ mod tests {
     use super::{
         CancelKmsKeyDeletionRequest, CreateKeyApiRequest, CreateKmsKeyRequest, DeleteKmsKeyRequest, GenerateDataKeyApiRequest,
         extract_key_id, kms_create_key_actions, kms_delete_key_actions, kms_describe_key_actions, kms_generate_data_key_actions,
-        kms_list_keys_actions,
+        kms_list_keys_actions, scoped_key_id,
     };
     use http::Uri;
     use rustfs_policy::policy::action::{Action, AdminAction, KmsAction};
+    use rustfs_policy::policy::{Args, Policy};
+    use std::collections::HashMap;
 
     fn assert_has_action(actions: &[Action], action: Action) {
         assert!(actions.contains(&action), "expected action list to contain {action:?}");
@@ -422,6 +450,248 @@ mod tests {
             assert!(err.to_string().contains("unknown field"));
         }
     }
+
+    #[test]
+    fn scoped_key_id_reads_the_body_first_and_falls_back_to_the_query() {
+        let uri: Uri = "/rustfs/admin/v3/kms/keys/delete?keyId=query-key"
+            .parse()
+            .expect("uri should parse");
+
+        assert_eq!(scoped_key_id(br#"{"key_id":"body-key"}"#, &uri).as_deref(), Some("body-key"));
+        assert_eq!(scoped_key_id(b"", &uri).as_deref(), Some("query-key"));
+
+        // The query aliases accepted by the legacy describe endpoint are not
+        // accepted here, because the handlers only execute on `keyId`.
+        let alias_uri: Uri = "/rustfs/admin/v3/kms/keys/delete?key-id=query-key"
+            .parse()
+            .expect("uri should parse");
+        assert_eq!(scoped_key_id(b"", &alias_uri), None);
+    }
+
+    /// A body the endpoint will reject leaves the request unscoped: the gate then
+    /// answers as it did before resource scoping and the request still fails on
+    /// its own parse error, so no key is operated on under that decision.
+    #[test]
+    fn scoped_key_id_is_absent_for_bodies_that_name_no_key() {
+        let uri: Uri = "/rustfs/admin/v3/kms/keys/delete".parse().expect("uri should parse");
+
+        assert_eq!(scoped_key_id(b"not json", &uri), None);
+        assert_eq!(scoped_key_id(br#"{"key_id":"" }"#, &uri).as_deref(), Some(""));
+        assert_eq!(scoped_key_id(b"{}", &uri).as_deref(), Some(""));
+        assert_eq!(scoped_key_id(b"", &uri), None);
+    }
+
+    /// Identity policy scoped to a single KMS key, or unscoped when `key_arn` is
+    /// `None` (the legacy action-only form that must keep matching every key).
+    fn kms_policy(key_arn: Option<&str>) -> Policy {
+        let resource = match key_arn {
+            Some(arn) => format!(r#","Resource":["{arn}"]"#),
+            None => String::new(),
+        };
+        let document = format!(r#"{{"Version":"2012-10-17","Statement":[{{"Effect":"Allow","Action":["kms:*"]{resource}}}]}}"#);
+
+        Policy::parse_config(document.as_bytes()).expect("kms policy should parse")
+    }
+
+    async fn policy_allows(policy: &Policy, action: Action, key_id: &str, is_owner: bool) -> bool {
+        let groups = None;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+
+        policy
+            .is_allowed(&Args {
+                account: "kms-operator",
+                groups: &groups,
+                action,
+                // The admin gate scopes KMS requests through `AdminResourceScope::kms_key`,
+                // which puts the requested key id in the object slot with an empty bucket.
+                bucket: "",
+                conditions: &conditions,
+                is_owner,
+                object: key_id,
+                claims: &claims,
+                deny_only: false,
+            })
+            .await
+    }
+
+    fn describe_by_query(key_id: &str) -> String {
+        let uri: Uri = format!("/rustfs/admin/v3/kms/key/status?keyId={key_id}")
+            .parse()
+            .expect("uri should parse");
+        extract_key_id(&uri).unwrap_or_default()
+    }
+
+    fn describe_by_path(key_id: &str) -> String {
+        // `DescribeKmsKeyHandler` reads the router path parameter verbatim.
+        key_id.to_string()
+    }
+
+    fn generate_data_key_body(key_id: &str) -> String {
+        let body = format!(r#"{{"key_id":"{key_id}","key_spec":"Aes256"}}"#);
+        serde_json::from_str::<GenerateDataKeyApiRequest>(&body)
+            .expect("generate-data-key body should parse")
+            .key_id
+    }
+
+    fn key_id_body(key_id: &str) -> String {
+        let uri: Uri = "/rustfs/admin/v3/kms/keys".parse().expect("uri should parse");
+        scoped_key_id(format!(r#"{{"key_id":"{key_id}"}}"#).as_bytes(), &uri).unwrap_or_default()
+    }
+
+    fn key_id_query(key_id: &str) -> String {
+        let uri: Uri = format!("/rustfs/admin/v3/kms/keys?keyId={key_id}")
+            .parse()
+            .expect("uri should parse");
+        scoped_key_id(b"", &uri).unwrap_or_default()
+    }
+
+    /// Endpoint label, the action it gates on, and the resolution its handler
+    /// performs to obtain the key the gate is scoped to.
+    type SingleKeyEndpoint = (&'static str, Action, fn(&str) -> String);
+
+    /// Every KMS admin endpoint that acts on one key. Lifecycle endpoints share
+    /// `scoped_key_id` with the delete/cancel endpoints (see `kms_key_lifecycle`).
+    fn single_key_endpoints() -> Vec<SingleKeyEndpoint> {
+        vec![
+            (
+                "GET /v3/kms/key/status",
+                Action::KmsAction(KmsAction::DescribeKeyAction),
+                describe_by_query,
+            ),
+            (
+                "GET /v3/kms/keys/{key_id}",
+                Action::KmsAction(KmsAction::DescribeKeyAction),
+                describe_by_path,
+            ),
+            (
+                "POST /v3/kms/generate-data-key",
+                Action::KmsAction(KmsAction::GenerateDataKeyAction),
+                generate_data_key_body,
+            ),
+            ("DELETE /v3/kms/keys/delete", Action::KmsAction(KmsAction::DeleteKeyAction), key_id_body),
+            (
+                "DELETE /v3/kms/keys/delete?keyId=",
+                Action::KmsAction(KmsAction::DeleteKeyAction),
+                key_id_query,
+            ),
+            (
+                "POST /v3/kms/keys/cancel-deletion",
+                Action::KmsAction(KmsAction::DeleteKeyAction),
+                key_id_body,
+            ),
+            ("POST /v3/kms/keys/enable", Action::KmsAction(KmsAction::EnableKeyAction), key_id_body),
+            ("POST /v3/kms/keys/disable", Action::KmsAction(KmsAction::DisableKeyAction), key_id_body),
+            ("POST /v3/kms/keys/rotate", Action::KmsAction(KmsAction::RotateKeyAction), key_id_body),
+        ]
+    }
+
+    /// A policy limited to `key/A` must not authorize any single-key endpoint
+    /// against another key (rustfs/backlog#1582).
+    #[tokio::test]
+    async fn single_key_endpoints_reject_a_key_outside_the_policy_scope() {
+        let policy = kms_policy(Some("arn:aws:kms:::key/key-a"));
+
+        for (endpoint, action, resolve_key_id) in single_key_endpoints() {
+            assert!(
+                policy_allows(&policy, action, &resolve_key_id("key-a"), false).await,
+                "{endpoint} must stay allowed on the key the policy names"
+            );
+            assert!(
+                !policy_allows(&policy, action, &resolve_key_id("key-b"), false).await,
+                "{endpoint} must be denied on a key outside the policy scope"
+            );
+        }
+    }
+
+    /// Compatibility pin: a KMS statement without resources keeps matching every
+    /// key, and the owner short-circuit is unaffected by scoping.
+    #[tokio::test]
+    async fn unscoped_policies_and_owner_credentials_keep_matching_every_key() {
+        let unscoped = kms_policy(None);
+        let scoped = kms_policy(Some("arn:aws:kms:::key/key-a"));
+
+        for (endpoint, action, resolve_key_id) in single_key_endpoints() {
+            for key in ["key-a", "key-b"] {
+                assert!(
+                    policy_allows(&unscoped, action, &resolve_key_id(key), false).await,
+                    "{endpoint} must stay allowed on {key} for a resource-less KMS statement"
+                );
+                assert!(
+                    policy_allows(&scoped, action, &resolve_key_id(key), true).await,
+                    "{endpoint} must stay allowed on {key} for an owner credential"
+                );
+            }
+        }
+    }
+
+    /// The key the handler resolves also reaches Deny statements, so an explicit
+    /// Deny on one key survives an unscoped Allow.
+    #[tokio::test]
+    async fn explicit_deny_on_one_key_overrides_an_unscoped_allow() {
+        let document = r#"{"Version":"2012-10-17","Statement":[
+            {"Effect":"Allow","Action":["kms:*"]},
+            {"Effect":"Deny","Action":["kms:*"],"Resource":["arn:aws:kms:::key/key-b"]}
+        ]}"#;
+        let policy = Policy::parse_config(document.as_bytes()).expect("kms policy should parse");
+
+        for (endpoint, action, resolve_key_id) in single_key_endpoints() {
+            assert!(
+                policy_allows(&policy, action, &resolve_key_id("key-a"), false).await,
+                "{endpoint} must stay allowed on a key the Deny does not name"
+            );
+            assert!(
+                !policy_allows(&policy, action, &resolve_key_id("key-b"), false).await,
+                "{endpoint} must be denied on the key the Deny names"
+            );
+        }
+    }
+
+    /// The single-key handlers must reach the gate through the KMS-scoped entry
+    /// point; the endpoints that have no target key must not (passing an empty
+    /// scope there would be a no-op, but the unscoped call documents intent).
+    #[test]
+    fn single_key_handlers_authorize_through_the_kms_scoped_gate() {
+        let src = include_str!("kms_keys.rs");
+
+        for handler in [
+            "DescribeKeyHandler",
+            "GenerateDataKeyHandler",
+            "DeleteKmsKeyHandler",
+            "CancelKmsKeyDeletionHandler",
+            "DescribeKmsKeyHandler",
+        ] {
+            let block = operation_block(src, handler);
+            assert!(
+                block.contains("validate_admin_request_with_kms_key("),
+                "{handler} must scope its authorization to the requested key"
+            );
+        }
+
+        for handler in [
+            "CreateKeyHandler",
+            "ListKeysHandler",
+            "CreateKmsKeyHandler",
+            "ListKmsKeysHandler",
+        ] {
+            let block = operation_block(src, handler);
+            assert!(
+                !block.contains("validate_admin_request_with_kms_key("),
+                "{handler} has no target key and must not claim a KMS resource scope"
+            );
+        }
+    }
+
+    fn operation_block<'a>(src: &'a str, handler: &str) -> &'a str {
+        let marker = format!("impl Operation for {handler}");
+        let block = src.split_once(&marker).expect("handler impl should exist").1;
+        let end = ["\n/// ", "\n#[derive(", "\n#[cfg(test)]", "\npub struct "]
+            .into_iter()
+            .filter_map(|item| block.find(item))
+            .min()
+            .unwrap_or(block.len());
+        &block[..end]
+    }
 }
 
 /// List KMS keys (legacy endpoint)
@@ -507,24 +777,31 @@ impl Operation for GenerateDataKeyHandler {
         let (cred, owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
-        validate_admin_request(
+        // The body names the key this endpoint derives a data key from, so it has
+        // to be read before the gate runs. Input failures stay deferred until
+        // after the gate so an unauthorized caller still sees AccessDenied.
+        let parsed = req
+            .input
+            .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
+            .await
+            .map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))
+            .and_then(|body| {
+                serde_json::from_slice::<GenerateDataKeyApiRequest>(&body)
+                    .map_err(|e| s3_error!(InvalidRequest, "invalid JSON: {}", e))
+            });
+
+        validate_admin_request_with_kms_key(
             &req.headers,
             &cred,
             owner,
             false,
             kms_generate_data_key_actions(),
             req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            parsed.as_ref().map(|request| request.key_id.as_str()).unwrap_or_default(),
         )
         .await?;
 
-        let body = req
-            .input
-            .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
-            .await
-            .map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))?;
-
-        let request: GenerateDataKeyApiRequest =
-            serde_json::from_slice(&body).map_err(|e| s3_error!(InvalidRequest, "invalid JSON: {}", e))?;
+        let request = parsed?;
 
         let Some(service) = kms_encryption_service_from_context().await else {
             return Err(s3_error!(InternalError, "KMS service not initialized"));
@@ -732,21 +1009,27 @@ impl Operation for DeleteKmsKeyHandler {
         let (cred, owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
-        validate_admin_request(
+        // Read the body before the gate so the request can be scoped to the key it
+        // targets; the read failure is surfaced afterwards to keep AccessDenied
+        // ahead of input errors for callers that are not authorized at all.
+        let body = req.input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await;
+
+        validate_admin_request_with_kms_key(
             &req.headers,
             &cred,
             owner,
             false,
             kms_delete_key_actions(),
             req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            body.as_ref()
+                .ok()
+                .and_then(|body| scoped_key_id(body, &req.uri))
+                .as_deref()
+                .unwrap_or_default(),
         )
         .await?;
 
-        let body = req
-            .input
-            .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
-            .await
-            .map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))?;
+        let body = body.map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))?;
 
         let request: DeleteKmsKeyRequest = if body.is_empty() {
             let query_params = extract_query_params(&req.uri);
@@ -906,21 +1189,26 @@ impl Operation for CancelKmsKeyDeletionHandler {
         let (cred, owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
-        validate_admin_request(
+        // Same ordering as the delete endpoint: the target key is resolved before
+        // the gate, the read failure only after it.
+        let body = req.input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await;
+
+        validate_admin_request_with_kms_key(
             &req.headers,
             &cred,
             owner,
             false,
             kms_delete_key_actions(),
             req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            body.as_ref()
+                .ok()
+                .and_then(|body| scoped_key_id(body, &req.uri))
+                .as_deref()
+                .unwrap_or_default(),
         )
         .await?;
 
-        let body = req
-            .input
-            .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
-            .await
-            .map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))?;
+        let body = body.map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))?;
 
         let request: CancelKmsKeyDeletionRequest = if body.is_empty() {
             let query_params = extract_query_params(&req.uri);
@@ -1180,13 +1468,14 @@ impl Operation for DescribeKmsKeyHandler {
         let (cred, owner) =
             check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
 
-        validate_admin_request(
+        validate_admin_request_with_kms_key(
             &req.headers,
             &cred,
             owner,
             false,
             kms_describe_key_actions(),
             req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+            params.get("key_id").unwrap_or_default(),
         )
         .await?;