feat(admin): add KMS key lifecycle endpoints and deletion reference gate (#5496)

* feat(kms): add key lifecycle operations to the backend contract

Add enable_key/disable_key/rotate_key to KmsBackend with conservative
defaults returning the typed UnsupportedCapability error, mirroring
remove_expired_key. KmsManager gains matching pass-through methods and
drops cached key metadata after every successful state mutation so the
next describe observes backend truth. The local backend overrides
enable/disable, delegating to its state-machine-gated client methods;
rotation stays rejected, matching its advertised capabilities. New
dedicated policy actions kms:EnableKey and kms:DisableKey complete the
KMS action taxonomy alongside the existing kms:RotateKey.

* feat(admin): add KMS key enable/disable/rotate endpoints

POST /v3/kms/keys/enable, /v3/kms/keys/disable and /v3/kms/keys/rotate,
following the existing /v3/kms/keys handler conventions: key_id body
with keyId query fallback, {success, message, key_id, key_metadata}
responses, and 503 JSON while the KMS service is absent. Error mapping
keeps InvalidOperation/ValidationError at 400 like the sibling handlers
and surfaces UnsupportedCapability as 501 so a backend capability gap is
never mistaken for a missing key. Existing /v3/kms/keys handlers are
untouched apart from a visibility change on a private query helper.

* feat(kms): gate scheduled key deletion on bucket encryption references

Implement the DeletionReferenceChecker seam left by the deletion worker:
before any material is destroyed, every bucket's SSE configuration is
checked for a default KMS key reference and a hit blocks the removal.
The gate fails closed - an unpublished object store, a failed bucket
listing or an unreadable per-bucket encryption config all report a
blocking reference - because destroying key material is irreversible
while a blocked removal is simply retried on the next sweep. Registered
during init_kms_system before the service can start, so every worker
spawn observes it. Storage access goes through a new kms section of the
root storage facade.

crates/kms/src/backends/local.rs

@@ -1533,6 +1533,14 @@ impl KmsBackend for LocalKmsBackend {
         })
     }
 
+    async fn enable_key(&self, key_id: &str) -> Result<()> {
+        self.client.enable_key(key_id, None).await
+    }
+
+    async fn disable_key(&self, key_id: &str) -> Result<()> {
+        self.client.disable_key(key_id, None).await
+    }
+
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }

crates/kms/src/backends/mod.rs

@@ -256,6 +256,34 @@ pub trait KmsBackend: Send + Sync {
     /// Cancel key deletion
     async fn cancel_key_deletion(&self, request: CancelKeyDeletionRequest) -> Result<CancelKeyDeletionResponse>;
 
+    /// Enable a disabled key so it can be used for cryptographic operations
+    /// again.
+    ///
+    /// Backends that advertise [`BackendCapabilities::enable_disable`] must
+    /// override this method; the default rejects the operation.
+    async fn enable_key(&self, _key_id: &str) -> Result<()> {
+        Err(KmsError::unsupported_capability("backend without enable/disable support", "enable_key"))
+    }
+
+    /// Disable a key, rejecting new cryptographic use while existing data
+    /// remains decryptable.
+    ///
+    /// Backends that advertise [`BackendCapabilities::enable_disable`] must
+    /// override this method; the default rejects the operation.
+    async fn disable_key(&self, _key_id: &str) -> Result<()> {
+        Err(KmsError::unsupported_capability("backend without enable/disable support", "disable_key"))
+    }
+
+    /// Rotate a key to a new version while prior versions remain available
+    /// for decryption.
+    ///
+    /// Only backends that advertise [`BackendCapabilities::rotate`] (that is,
+    /// backends with retained version history) may override this method; the
+    /// default rejects the operation.
+    async fn rotate_key(&self, _key_id: &str) -> Result<()> {
+        Err(KmsError::unsupported_capability("backend without rotation support", "rotate_key"))
+    }
+
     /// Health check
     async fn health_check(&self) -> Result<bool>;
 
@@ -523,6 +551,21 @@ mod tests {
         assert!(!capabilities.physical_delete);
     }
 
+    #[tokio::test]
+    async fn default_lifecycle_operations_are_unsupported() {
+        for (operation, result) in [
+            ("enable_key", MinimalBackend.enable_key("any-key").await),
+            ("disable_key", MinimalBackend.disable_key("any-key").await),
+            ("rotate_key", MinimalBackend.rotate_key("any-key").await),
+        ] {
+            let error = result.expect_err("backends must opt in to lifecycle operations by overriding them");
+            assert!(
+                matches!(error, KmsError::UnsupportedCapability { .. }),
+                "expected UnsupportedCapability for {operation}, got {error:?}"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn default_remove_expired_key_is_unsupported() {
         let error = MinimalBackend

crates/kms/src/manager.rs

@@ -155,6 +155,36 @@ impl KmsManager {
         Ok(response)
     }
 
+    /// Enable a disabled key
+    pub async fn enable_key(&self, key_id: &str) -> Result<()> {
+        self.backend.enable_key(key_id).await?;
+        self.invalidate_cached_metadata(key_id).await;
+        Ok(())
+    }
+
+    /// Disable a key; existing data remains decryptable
+    pub async fn disable_key(&self, key_id: &str) -> Result<()> {
+        self.backend.disable_key(key_id).await?;
+        self.invalidate_cached_metadata(key_id).await;
+        Ok(())
+    }
+
+    /// Rotate a key to a new version
+    pub async fn rotate_key(&self, key_id: &str) -> Result<()> {
+        self.backend.rotate_key(key_id).await?;
+        self.invalidate_cached_metadata(key_id).await;
+        Ok(())
+    }
+
+    /// Drop cached metadata after a state mutation so the next describe
+    /// observes backend truth instead of the pre-mutation snapshot.
+    async fn invalidate_cached_metadata(&self, key_id: &str) {
+        if self.enable_cache {
+            let mut cache = self.cache.write().await;
+            cache.remove_key_metadata(key_id).await;
+        }
+    }
+
     /// Perform health check on the KMS backend
     pub async fn health_check(&self) -> Result<bool> {
         self.backend.health_check().await
@@ -230,6 +260,52 @@ mod tests {
         assert!(health);
     }
 
+    #[tokio::test]
+    async fn lifecycle_round_trip_invalidates_cached_metadata() {
+        let temp_dir = tempdir().expect("Failed to create temp dir");
+        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+
+        let backend = Arc::new(LocalKmsBackend::new(config.clone()).await.expect("Failed to create backend"));
+        let manager = KmsManager::new(backend, config);
+
+        let key_id = manager
+            .create_key(CreateKeyRequest {
+                key_name: Some("lifecycle-round-trip".to_string()),
+                ..Default::default()
+            })
+            .await
+            .expect("Failed to create key")
+            .key_id;
+
+        let describe = |key_id: String| {
+            let manager = manager.clone();
+            async move {
+                manager
+                    .describe_key(DescribeKeyRequest { key_id })
+                    .await
+                    .expect("describe should succeed")
+                    .key_metadata
+                    .key_state
+            }
+        };
+
+        // Warm the metadata cache, then flip states; each describe must see
+        // the post-mutation state, proving the cache entry was dropped.
+        assert_eq!(describe(key_id.clone()).await, KeyState::Enabled);
+        manager.disable_key(&key_id).await.expect("disable should succeed");
+        assert_eq!(describe(key_id.clone()).await, KeyState::Disabled);
+        manager.enable_key(&key_id).await.expect("enable should succeed");
+        assert_eq!(describe(key_id.clone()).await, KeyState::Enabled);
+
+        // The local backend does not retain version history, so rotation is
+        // reported as a capability gap rather than a missing key.
+        let error = manager.rotate_key(&key_id).await.expect_err("local rotate must be rejected");
+        assert!(
+            matches!(error, crate::error::KmsError::UnsupportedCapability { .. }),
+            "expected UnsupportedCapability, got {error:?}"
+        );
+    }
+
     #[tokio::test]
     async fn generate_data_key_does_not_reuse_context_bound_ciphertext() {
         let temp_dir = tempdir().expect("Failed to create temp dir");

crates/policy/src/policy/action.rs

@@ -718,6 +718,10 @@ pub enum KmsAction {
     GenerateDataKeyAction,
     #[strum(serialize = "kms:DeleteKey")]
     DeleteKeyAction,
+    #[strum(serialize = "kms:EnableKey")]
+    EnableKeyAction,
+    #[strum(serialize = "kms:DisableKey")]
+    DisableKeyAction,
     #[strum(serialize = "kms:RotateKey")]
     RotateKeyAction,
     #[strum(serialize = "kms:ListKeys")]
@@ -755,6 +759,8 @@ mod tests {
             ("kms:ClearCache", KmsAction::ClearCacheAction),
             ("kms:GenerateDataKey", KmsAction::GenerateDataKeyAction),
             ("kms:DeleteKey", KmsAction::DeleteKeyAction),
+            ("kms:EnableKey", KmsAction::EnableKeyAction),
+            ("kms:DisableKey", KmsAction::DisableKeyAction),
             ("kms:RotateKey", KmsAction::RotateKeyAction),
             ("kms:ListKeys", KmsAction::ListKeysAction),
             ("kms:DescribeKey", KmsAction::DescribeKeyAction),