feat(kms): enforce shared key state machine across backends

Unify the key state x operation matrix behind a single gate in
backends/mod.rs and wire it into the Local, Vault KV2 and Vault Transit
backends: Disabled keys reject encryption, data key generation and
rotation while still allowing decryption and lifecycle recovery;
PendingDeletion keys reject everything except decryption and
cancellation (including repeated deletion scheduling); cancellation now
requires an actual pending deletion everywhere. This closes the missing
gates on KV2 encrypt/generate and Local generate_data_key, and stops
enable_key from silently reverting a pending deletion.

Decryption is deliberately left ungated in Disabled/PendingDeletion — an
explicit, documented and tested deviation from AWS KMS, since gating it
would break reads of existing objects the moment a key is disabled.

Add shared contract tests driving the full matrix offline for Local (and
via ignored tests against a live Vault for KV2/Transit), a stateless
contract for Static, an SSE-shaped regression proving existing envelopes
stay decryptable after disable, and a pin on the known-risk Enabled
default of Transit's synthesized metadata fallback.

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

crates/kms/src/backends/contract_tests.rs

@@ -0,0 +1,330 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Shared key state × operation contract tests for KMS backends.
+//!
+//! Every stateful backend must satisfy the same lifecycle matrix (see
+//! `ensure_key_state_permits`): Enabled permits everything, Disabled permits
+//! decryption and lifecycle recovery but rejects new cryptographic use, and
+//! PendingDeletion rejects everything except decryption and cancellation.
+//! Decryption staying available in Disabled/PendingDeletion is an explicit,
+//! tested deviation from AWS KMS: disabling a key must not break reads of
+//! objects already encrypted under it.
+//!
+//! The full matrix runs offline against the Local backend. The Vault KV2 and
+//! Vault Transit runs exercise the same helper but need a live Vault dev
+//! server, so they are `#[ignore]`d in CI. Static is covered by its own
+//! stateless contract below.
+
+use super::local::LocalKmsBackend;
+use super::static_kms::StaticKmsBackend;
+use super::vault::VaultKmsBackend;
+use super::vault_transit::VaultTransitKmsBackend;
+use super::{KmsBackend, KmsClient};
+use crate::config::KmsConfig;
+use crate::error::{KmsError, Result};
+use crate::manager::KmsManager;
+use crate::service::ObjectEncryptionService;
+use crate::types::{
+    CancelKeyDeletionRequest, CreateKeyRequest, DecryptRequest, DeleteKeyRequest, DescribeKeyRequest, EncryptRequest,
+    GenerateDataKeyRequest, KeySpec, KeyState, KeyUsage, ObjectEncryptionContext,
+};
+use base64::Engine as _;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use rand::RngExt as _;
+use std::collections::HashMap;
+use std::sync::Arc;
+
+fn expect_invalid_key_state<T: std::fmt::Debug>(result: Result<T>, expected_fragment: &str) {
+    match result {
+        Err(KmsError::InvalidOperation { message }) => assert!(
+            message.contains(expected_fragment),
+            "expected invalid-key-state message containing {expected_fragment:?}, got {message:?}"
+        ),
+        other => panic!("expected InvalidOperation (invalid key state), got {other:?}"),
+    }
+}
+
+fn context() -> HashMap<String, String> {
+    HashMap::from([("bucket".to_string(), "contract".to_string())])
+}
+
+fn generate_request(key_id: &str) -> GenerateDataKeyRequest {
+    GenerateDataKeyRequest {
+        key_id: key_id.to_string(),
+        key_spec: KeySpec::Aes256,
+        encryption_context: context(),
+    }
+}
+
+fn encrypt_request(key_id: &str) -> EncryptRequest {
+    EncryptRequest {
+        key_id: key_id.to_string(),
+        plaintext: b"contract-plaintext".to_vec(),
+        encryption_context: context(),
+        grant_tokens: Vec::new(),
+    }
+}
+
+fn decrypt_request(ciphertext: Vec<u8>) -> DecryptRequest {
+    DecryptRequest {
+        ciphertext,
+        encryption_context: context(),
+        grant_tokens: Vec::new(),
+    }
+}
+
+fn schedule_request(key_id: &str) -> DeleteKeyRequest {
+    DeleteKeyRequest {
+        key_id: key_id.to_string(),
+        pending_window_in_days: Some(7),
+        force_immediate: None,
+    }
+}
+
+fn cancel_request(key_id: &str) -> CancelKeyDeletionRequest {
+    CancelKeyDeletionRequest {
+        key_id: key_id.to_string(),
+    }
+}
+
+fn create_request(key_name: String) -> CreateKeyRequest {
+    CreateKeyRequest {
+        key_name: Some(key_name),
+        key_usage: KeyUsage::EncryptDecrypt,
+        ..Default::default()
+    }
+}
+
+async fn assert_key_state(backend: &dyn KmsBackend, key_id: &str, expected: KeyState) {
+    let described = backend
+        .describe_key(DescribeKeyRequest {
+            key_id: key_id.to_string(),
+        })
+        .await
+        .expect("describe_key must succeed for an existing key");
+    assert_eq!(described.key_metadata.key_state, expected, "unexpected state for key {key_id}");
+}
+
+/// Drives one freshly created (Enabled) key through the full state matrix.
+///
+/// `backend` is the product surface; `client` drives the lifecycle
+/// transitions not yet exposed through `KmsBackend`.
+async fn assert_state_machine_contract(backend: &dyn KmsBackend, client: &dyn KmsClient, key_id: &str) {
+    // Enabled: cryptographic use is allowed. Keep an envelope around to prove
+    // decryption keeps working in later states.
+    let data_key = backend
+        .generate_data_key(generate_request(key_id))
+        .await
+        .expect("Enabled key must generate data keys");
+    backend
+        .encrypt(encrypt_request(key_id))
+        .await
+        .expect("Enabled key must encrypt");
+
+    // Enabled -> Disabled.
+    client
+        .disable_key(key_id, None)
+        .await
+        .expect("disable from Enabled must succeed");
+    assert_key_state(backend, key_id, KeyState::Disabled).await;
+
+    // Disabled: new cryptographic use and rotation are rejected...
+    expect_invalid_key_state(backend.encrypt(encrypt_request(key_id)).await, "disabled");
+    expect_invalid_key_state(backend.generate_data_key(generate_request(key_id)).await, "disabled");
+    expect_invalid_key_state(client.rotate_key(key_id, None).await, "");
+    // ...but decryption of existing data keeps working (explicit AWS deviation)...
+    let decrypted = backend
+        .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
+        .await
+        .expect("decrypt with a disabled key must keep working");
+    assert_eq!(decrypted.plaintext, data_key.plaintext_key, "decrypt must recover the original data key");
+    // ...disable stays idempotent, cancel has nothing to cancel, and enable recovers.
+    client.disable_key(key_id, None).await.expect("disable must be idempotent");
+    expect_invalid_key_state(backend.cancel_key_deletion(cancel_request(key_id)).await, "not pending deletion");
+    client
+        .enable_key(key_id, None)
+        .await
+        .expect("enable from Disabled must succeed");
+    assert_key_state(backend, key_id, KeyState::Enabled).await;
+
+    // Disabled keys may still be scheduled for deletion.
+    client
+        .disable_key(key_id, None)
+        .await
+        .expect("disable before scheduling must succeed");
+    backend
+        .delete_key(schedule_request(key_id))
+        .await
+        .expect("scheduling deletion of a disabled key must succeed");
+    assert_key_state(backend, key_id, KeyState::PendingDeletion).await;
+
+    // PendingDeletion: everything except decryption and cancellation is rejected.
+    expect_invalid_key_state(backend.encrypt(encrypt_request(key_id)).await, "pending deletion");
+    expect_invalid_key_state(backend.generate_data_key(generate_request(key_id)).await, "pending deletion");
+    expect_invalid_key_state(client.enable_key(key_id, None).await, "pending deletion");
+    expect_invalid_key_state(client.disable_key(key_id, None).await, "pending deletion");
+    expect_invalid_key_state(client.rotate_key(key_id, None).await, "");
+    expect_invalid_key_state(client.schedule_key_deletion(key_id, 7, None).await, "pending deletion");
+    expect_invalid_key_state(backend.delete_key(schedule_request(key_id)).await, "pending deletion");
+    let decrypted = backend
+        .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
+        .await
+        .expect("decrypt with a pending-deletion key must keep working");
+    assert_eq!(decrypted.plaintext, data_key.plaintext_key);
+
+    // PendingDeletion -> Enabled through cancellation.
+    backend
+        .cancel_key_deletion(cancel_request(key_id))
+        .await
+        .expect("cancel from PendingDeletion must succeed");
+    assert_key_state(backend, key_id, KeyState::Enabled).await;
+    backend
+        .generate_data_key(generate_request(key_id))
+        .await
+        .expect("cancelled key must be usable again");
+
+    // Cancel without a pending deletion is an invalid state transition.
+    expect_invalid_key_state(backend.cancel_key_deletion(cancel_request(key_id)).await, "not pending deletion");
+}
+
+async fn local_fixture() -> (tempfile::TempDir, KmsConfig, LocalKmsBackend, String) {
+    let temp_dir = tempfile::tempdir().expect("temp dir should be created");
+    let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+    let backend = LocalKmsBackend::new(config.clone())
+        .await
+        .expect("local backend should build");
+    let created = backend
+        .create_key(create_request("contract-key".to_string()))
+        .await
+        .expect("key should be created");
+    (temp_dir, config, backend, created.key_id)
+}
+
+#[tokio::test]
+async fn local_backend_state_machine_contract() {
+    let (_temp_dir, _config, backend, key_id) = local_fixture().await;
+    assert_state_machine_contract(&backend, backend.lifecycle_client(), &key_id).await;
+}
+
+/// SSE-shaped regression: disabling a key must not break decryption of data
+/// keys created while it was enabled, while new data key creation must fail.
+#[tokio::test]
+async fn local_disabled_key_keeps_decrypting_existing_envelopes() {
+    let (_temp_dir, config, backend, key_id) = local_fixture().await;
+    let backend = Arc::new(backend);
+    let service = ObjectEncryptionService::new(KmsManager::new(backend.clone(), config));
+
+    let object_context = ObjectEncryptionContext::new("sse-bucket".to_string(), "dir/object.bin".to_string());
+    let kms_key = Some(key_id.clone());
+    let (_data_key, encrypted_blob) = service
+        .create_data_key(&kms_key, &object_context)
+        .await
+        .expect("data key creation must succeed while the key is enabled");
+
+    backend
+        .lifecycle_client()
+        .disable_key(&key_id, None)
+        .await
+        .expect("disable must succeed");
+
+    service
+        .decrypt_data_key(&encrypted_blob, &object_context)
+        .await
+        .expect("existing objects must stay readable after their KMS key is disabled");
+    expect_invalid_key_state(service.create_data_key(&kms_key, &object_context).await, "disabled");
+}
+
+/// Static is a stateless read-only backend: cryptographic operations always
+/// work against the single configured key and every lifecycle mutation is
+/// rejected as an invalid operation.
+#[tokio::test]
+async fn static_backend_stateless_contract() {
+    let key_id = "static-contract-key";
+    let mut raw_key = [0u8; 32];
+    rand::rng().fill(&mut raw_key[..]);
+    let config = KmsConfig::static_kms(key_id.to_string(), BASE64.encode(raw_key));
+    let static_backend = StaticKmsBackend::new(config).await.expect("static backend should build");
+    // StaticKmsBackend implements both traits with overlapping method names,
+    // so pin each surface once instead of qualifying every call.
+    let backend: &dyn KmsBackend = &static_backend;
+    let client: &dyn KmsClient = &static_backend;
+
+    let data_key = backend
+        .generate_data_key(generate_request(key_id))
+        .await
+        .expect("static backend must generate data keys");
+    let decrypted = backend
+        .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
+        .await
+        .expect("static backend must decrypt its own envelopes");
+    assert_eq!(decrypted.plaintext, data_key.plaintext_key);
+    assert_key_state(backend, key_id, KeyState::Enabled).await;
+
+    expect_invalid_key_state(backend.create_key(create_request("another-key".to_string())).await, "read-only");
+    expect_invalid_key_state(backend.delete_key(schedule_request(key_id)).await, "read-only");
+    expect_invalid_key_state(backend.cancel_key_deletion(cancel_request(key_id)).await, "read-only");
+    expect_invalid_key_state(client.disable_key(key_id, None).await, "read-only");
+    expect_invalid_key_state(client.schedule_key_deletion(key_id, 7, None).await, "read-only");
+    expect_invalid_key_state(client.rotate_key(key_id, None).await, "read-only");
+}
+
+fn vault_dev_config(constructor: fn(url::Url, String) -> KmsConfig) -> KmsConfig {
+    let address = std::env::var("RUSTFS_KMS_VAULT_ADDR").unwrap_or_else(|_| "http://127.0.0.1:8200".to_string());
+    let token = std::env::var("RUSTFS_KMS_VAULT_TOKEN").unwrap_or_else(|_| "dev-token".to_string());
+    let mut config = constructor(url::Url::parse(&address).expect("vault address should parse"), token);
+    config.allow_insecure_dev_defaults = true;
+    config
+}
+
+#[tokio::test]
+#[ignore] // Requires a running Vault instance (dev mode) with a KV2 mount
+async fn vault_kv2_backend_state_machine_contract() {
+    let config = vault_dev_config(KmsConfig::vault);
+    let backend = VaultKmsBackend::new(config).await.expect("vault kv2 backend should build");
+    let created = backend
+        .create_key(create_request(format!("contract-{}", uuid::Uuid::new_v4())))
+        .await
+        .expect("key should be created");
+
+    assert_state_machine_contract(&backend, backend.lifecycle_client(), &created.key_id).await;
+
+    // Cleanup: leave the key pending deletion so repeated runs stay tidy.
+    let _ = backend.delete_key(schedule_request(&created.key_id)).await;
+}
+
+#[tokio::test]
+#[ignore] // Requires a running Vault instance (dev mode) with the transit engine enabled
+async fn vault_transit_backend_state_machine_contract() {
+    let config = vault_dev_config(KmsConfig::vault_transit);
+    let backend = VaultTransitKmsBackend::new(config)
+        .await
+        .expect("vault transit backend should build");
+    let created = backend
+        .create_key(create_request(format!("contract-{}", uuid::Uuid::new_v4())))
+        .await
+        .expect("key should be created");
+
+    assert_state_machine_contract(&backend, backend.lifecycle_client(), &created.key_id).await;
+
+    // Transit additionally supports rotation, which must only work while the
+    // key is Enabled (the shared matrix already covered the rejections).
+    backend
+        .lifecycle_client()
+        .rotate_key(&created.key_id, None)
+        .await
+        .expect("rotation of an Enabled transit key must succeed");
+
+    let _ = backend.delete_key(schedule_request(&created.key_id)).await;
+}

crates/kms/src/backends/local.rs

@@ -14,7 +14,7 @@
 
 //! Local file-based KMS backend implementation
 
-use crate::backends::{BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient, StateGatedOperation, ensure_key_status_permits};
 use crate::config::KmsConfig;
 use crate::config::LocalConfig;
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -931,9 +931,12 @@ impl LocalKmsClient {
 
 #[async_trait]
 impl KmsClient for LocalKmsClient {
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
+    async fn generate_data_key(&self, request: &GenerateKeyRequest, context: Option<&OperationContext>) -> Result<DataKeyInfo> {
         debug!("Generating data key for master key: {}", request.master_key_id);
 
+        let key_info = self.describe_key(&request.master_key_id, context).await?;
+        ensure_key_status_permits(&request.master_key_id, &key_info.status, StateGatedOperation::GenerateDataKey)?;
+
         // Generate random data key material
         let key_length = match request.key_spec.as_str() {
             "AES_256" => 32,
@@ -972,14 +975,9 @@ impl KmsClient for LocalKmsClient {
     async fn encrypt(&self, request: &EncryptRequest, context: Option<&OperationContext>) -> Result<EncryptResponse> {
         debug!("Encrypting data with key: {}", request.key_id);
 
-        // Verify key exists and is active
+        // Verify key exists and its state allows encryption
         let key_info = self.describe_key(&request.key_id, context).await?;
-        if key_info.status != KeyStatus::Active {
-            return Err(KmsError::invalid_operation(format!(
-                "Key {} is not active (status: {:?})",
-                request.key_id, key_info.status
-            )));
-        }
+        ensure_key_status_permits(&request.key_id, &key_info.status, StateGatedOperation::Encrypt)?;
 
         let (ciphertext, _nonce) = self.encrypt_with_master_key(&request.key_id, &request.plaintext).await?;
 
@@ -1110,6 +1108,7 @@ impl KmsClient for LocalKmsClient {
 
         let _write_guard = self.lock_key_for_write(key_id).await;
         let mut master_key = self.load_master_key(key_id).await?;
+        ensure_key_status_permits(key_id, &master_key.status, StateGatedOperation::Enable)?;
         master_key.status = KeyStatus::Active;
 
         // Preserve the existing key material. Regenerating it on a pure status change would
@@ -1127,6 +1126,7 @@ impl KmsClient for LocalKmsClient {
 
         let _write_guard = self.lock_key_for_write(key_id).await;
         let mut master_key = self.load_master_key(key_id).await?;
+        ensure_key_status_permits(key_id, &master_key.status, StateGatedOperation::Disable)?;
         master_key.status = KeyStatus::Disabled;
 
         // Preserve the existing key material (see enable_key): a status change must never
@@ -1148,6 +1148,7 @@ impl KmsClient for LocalKmsClient {
 
         let _write_guard = self.lock_key_for_write(key_id).await;
         let mut master_key = self.load_master_key(key_id).await?;
+        ensure_key_status_permits(key_id, &master_key.status, StateGatedOperation::ScheduleDeletion)?;
         master_key.status = KeyStatus::PendingDeletion;
 
         // Preserve the existing key material (see enable_key): scheduling deletion must not
@@ -1165,6 +1166,9 @@ impl KmsClient for LocalKmsClient {
 
         let _write_guard = self.lock_key_for_write(key_id).await;
         let mut master_key = self.load_master_key(key_id).await?;
+        if master_key.status != KeyStatus::PendingDeletion {
+            return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion")));
+        }
         master_key.status = KeyStatus::Active;
 
         // Preserve the existing key material (see enable_key): cancelling deletion must recover
@@ -1215,6 +1219,12 @@ pub struct LocalKmsBackend {
 }
 
 impl LocalKmsBackend {
+    /// Lifecycle driver for the shared state-machine contract tests.
+    #[cfg(test)]
+    pub(crate) fn lifecycle_client(&self) -> &LocalKmsClient {
+        &self.client
+    }
+
     /// Create a new LocalKmsBackend
     pub async fn new(config: KmsConfig) -> Result<Self> {
         config.validate()?;
@@ -1399,6 +1409,8 @@ impl KmsBackend for LocalKmsBackend {
             });
         } else {
             // Schedule for deletion (default 30 days)
+            ensure_key_status_permits(key_id, &master_key.status, StateGatedOperation::ScheduleDeletion)?;
+
             let days = request.pending_window_in_days.unwrap_or(30);
             if !(7..=30).contains(&days) {
                 return Err(KmsError::invalid_parameter("pending_window_in_days must be between 7 and 30".to_string()));
@@ -1492,6 +1504,17 @@ impl KmsBackend for LocalKmsBackend {
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
+
+    fn capabilities(&self) -> BackendCapabilities {
+        // Rotation stays unadvertised until historical key versions can be
+        // retained (see LocalKmsClient::rotate_key); without version history
+        // there is also no versioning capability. Deletion deadlines are not
+        // yet persisted across restarts, but scheduling itself is supported.
+        BackendCapabilities::minimal()
+            .with_enable_disable(true)
+            .with_schedule_deletion(true)
+            .with_physical_delete(true)
+    }
 }
 
 #[cfg(test)]
@@ -2606,9 +2629,16 @@ mod tests {
             client.schedule_key_deletion(key_id, 7, None),
             client.enable_key(key_id, None),
         );
-        disable.expect("disable");
-        schedule.expect("schedule deletion");
-        enable.expect("enable");
+        // The per-key lock serializes the three transitions in an arbitrary
+        // order, and the state gate may legitimately reject a transition that
+        // lost the race (e.g. enable after deletion was scheduled). Any other
+        // error kind would still mean corrupted storage.
+        for result in [disable, schedule, enable] {
+            match result {
+                Ok(()) | Err(KmsError::InvalidOperation { .. }) => {}
+                Err(other) => panic!("concurrent transition must only fail with a state rejection, got {other:?}"),
+            }
+        }
 
         // Whatever the serialization order, the file must be one writer's
         // complete output with the original material intact.

crates/kms/src/backends/mod.rs

@@ -14,17 +14,88 @@
 
 //! KMS backend implementations
 
-use crate::error::Result;
+use crate::error::{KmsError, Result};
 use crate::types::*;
 use async_trait::async_trait;
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
+#[cfg(test)]
+mod contract_tests;
 pub mod local;
 pub mod static_kms;
 pub mod vault;
 pub(crate) mod vault_credentials;
 pub mod vault_transit;
 
+/// Operations whose availability depends on the key's lifecycle state.
+///
+/// Decryption is deliberately absent: RustFS allows decryption with
+/// `Disabled` and `PendingDeletion` keys — an explicit deviation from AWS
+/// KMS — because rejecting it would break reads of every object encrypted
+/// under a key the moment it is disabled. Deletion cancellation is also
+/// absent: it is valid exactly when the key is `PendingDeletion`, which call
+/// sites enforce directly.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum StateGatedOperation {
+    Encrypt,
+    GenerateDataKey,
+    Rotate,
+    Enable,
+    Disable,
+    ScheduleDeletion,
+}
+
+impl StateGatedOperation {
+    fn describe(self) -> &'static str {
+        match self {
+            Self::Encrypt => "encryption",
+            Self::GenerateDataKey => "data key generation",
+            Self::Rotate => "rotation",
+            Self::Enable => "enabling",
+            Self::Disable => "disabling",
+            Self::ScheduleDeletion => "deletion scheduling",
+        }
+    }
+}
+
+/// Enforce the shared key state × operation matrix.
+///
+/// - `Enabled`: every operation is allowed.
+/// - `Disabled`: enabling, disabling (idempotent) and deletion scheduling are
+///   allowed; encryption, data key generation and rotation are rejected.
+/// - `PendingDeletion`: every state-gated operation is rejected, including a
+///   repeated deletion schedule; only cancellation and decryption proceed.
+/// - `PendingImport`/`Unavailable`: the key is not usable and is reported as
+///   not found.
+pub(crate) fn ensure_key_state_permits(key_id: &str, state: &KeyState, operation: StateGatedOperation) -> Result<()> {
+    match state {
+        KeyState::Enabled => Ok(()),
+        KeyState::Disabled => match operation {
+            StateGatedOperation::Enable | StateGatedOperation::Disable | StateGatedOperation::ScheduleDeletion => Ok(()),
+            StateGatedOperation::Encrypt | StateGatedOperation::GenerateDataKey | StateGatedOperation::Rotate => Err(
+                KmsError::invalid_key_state(format!("Key {key_id} is disabled: {} is not allowed", operation.describe())),
+            ),
+        },
+        KeyState::PendingDeletion => Err(KmsError::invalid_key_state(format!(
+            "Key {key_id} is pending deletion: {} is not allowed",
+            operation.describe()
+        ))),
+        KeyState::PendingImport | KeyState::Unavailable => Err(KmsError::key_not_found(key_id)),
+    }
+}
+
+/// [`ensure_key_state_permits`] for backends that persist [`KeyStatus`].
+pub(crate) fn ensure_key_status_permits(key_id: &str, status: &KeyStatus, operation: StateGatedOperation) -> Result<()> {
+    let state = match status {
+        KeyStatus::Active => KeyState::Enabled,
+        KeyStatus::Disabled => KeyState::Disabled,
+        KeyStatus::PendingDeletion => KeyState::PendingDeletion,
+        KeyStatus::Deleted => KeyState::Unavailable,
+    };
+    ensure_key_state_permits(key_id, &state, operation)
+}
+
 /// Abstract KMS client interface that all backends must implement
 #[async_trait]
 pub trait KmsClient: Send + Sync {
@@ -184,6 +255,16 @@ pub trait KmsBackend: Send + Sync {
 
     /// Health check
     async fn health_check(&self) -> Result<bool>;
+
+    /// Report which operations this backend actually supports.
+    ///
+    /// The default is conservative: only the operations every backend is
+    /// required to implement by this trait are advertised. Optional lifecycle
+    /// operations (rotation, enable/disable, deletion scheduling, ...) must be
+    /// opted in by overriding this method.
+    fn capabilities(&self) -> BackendCapabilities {
+        BackendCapabilities::minimal()
+    }
 }
 
 /// Information about a KMS backend
@@ -237,3 +318,225 @@ impl BackendInfo {
         self
     }
 }
+
+/// Set of operations a KMS backend supports.
+///
+/// Reported by [`KmsBackend::capabilities`] so callers (manager, admin API)
+/// can discover what the active backend can do without probing individual
+/// operations. Marked `#[non_exhaustive]` so new capability flags can be
+/// added without breaking downstream code; construct values through
+/// [`BackendCapabilities::minimal`] and the `with_*` builders.
+#[non_exhaustive]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub struct BackendCapabilities {
+    /// Direct encryption of caller-provided plaintext with a master key
+    pub encrypt: bool,
+    /// Decryption of previously produced ciphertext
+    pub decrypt: bool,
+    /// Data encryption key (DEK) generation
+    pub generate_data_key: bool,
+    /// Key rotation that retains prior versions for decryption
+    pub rotate: bool,
+    /// Enabling and disabling keys
+    pub enable_disable: bool,
+    /// Scheduling key deletion with a pending window
+    pub schedule_deletion: bool,
+    /// Multiple key versions addressable after rotation
+    pub versioning: bool,
+    /// Irreversible physical deletion of key material
+    pub physical_delete: bool,
+}
+
+impl BackendCapabilities {
+    /// Conservative baseline: only the operations that every [`KmsBackend`]
+    /// implementation is required to provide by the trait. All optional
+    /// lifecycle capabilities default to unsupported.
+    pub const fn minimal() -> Self {
+        Self {
+            encrypt: true,
+            decrypt: true,
+            generate_data_key: true,
+            rotate: false,
+            enable_disable: false,
+            schedule_deletion: false,
+            versioning: false,
+            physical_delete: false,
+        }
+    }
+
+    /// Set whether direct encryption is supported
+    pub const fn with_encrypt(mut self, encrypt: bool) -> Self {
+        self.encrypt = encrypt;
+        self
+    }
+
+    /// Set whether decryption is supported
+    pub const fn with_decrypt(mut self, decrypt: bool) -> Self {
+        self.decrypt = decrypt;
+        self
+    }
+
+    /// Set whether data key generation is supported
+    pub const fn with_generate_data_key(mut self, generate_data_key: bool) -> Self {
+        self.generate_data_key = generate_data_key;
+        self
+    }
+
+    /// Set whether version-retaining key rotation is supported
+    pub const fn with_rotate(mut self, rotate: bool) -> Self {
+        self.rotate = rotate;
+        self
+    }
+
+    /// Set whether enabling/disabling keys is supported
+    pub const fn with_enable_disable(mut self, enable_disable: bool) -> Self {
+        self.enable_disable = enable_disable;
+        self
+    }
+
+    /// Set whether scheduled deletion with a pending window is supported
+    pub const fn with_schedule_deletion(mut self, schedule_deletion: bool) -> Self {
+        self.schedule_deletion = schedule_deletion;
+        self
+    }
+
+    /// Set whether multiple key versions are supported
+    pub const fn with_versioning(mut self, versioning: bool) -> Self {
+        self.versioning = versioning;
+        self
+    }
+
+    /// Set whether physical deletion of key material is supported
+    pub const fn with_physical_delete(mut self, physical_delete: bool) -> Self {
+        self.physical_delete = physical_delete;
+        self
+    }
+}
+
+impl Default for BackendCapabilities {
+    fn default() -> Self {
+        Self::minimal()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::KmsConfig;
+    use base64::Engine as _;
+    use base64::engine::general_purpose::STANDARD as BASE64;
+
+    /// Backend that implements only the trait-mandated operations and relies
+    /// on the default `capabilities` implementation.
+    struct MinimalBackend;
+
+    #[async_trait]
+    impl KmsBackend for MinimalBackend {
+        async fn create_key(&self, _request: CreateKeyRequest) -> Result<CreateKeyResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn encrypt(&self, _request: EncryptRequest) -> Result<EncryptResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn decrypt(&self, _request: DecryptRequest) -> Result<DecryptResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn generate_data_key(&self, _request: GenerateDataKeyRequest) -> Result<GenerateDataKeyResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn describe_key(&self, _request: DescribeKeyRequest) -> Result<DescribeKeyResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn list_keys(&self, _request: ListKeysRequest) -> Result<ListKeysResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn delete_key(&self, _request: DeleteKeyRequest) -> Result<DeleteKeyResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn cancel_key_deletion(&self, _request: CancelKeyDeletionRequest) -> Result<CancelKeyDeletionResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn health_check(&self) -> Result<bool> {
+            Ok(true)
+        }
+    }
+
+    fn capabilities_snapshot(capabilities: BackendCapabilities) -> std::collections::BTreeMap<String, bool> {
+        serde_json::from_value(serde_json::to_value(capabilities).expect("capabilities should serialize"))
+            .expect("capabilities should deserialize into a flat bool map")
+    }
+
+    #[test]
+    fn default_capabilities_are_conservative() {
+        let capabilities = MinimalBackend.capabilities();
+        assert_eq!(capabilities, BackendCapabilities::minimal());
+        assert_eq!(capabilities, BackendCapabilities::default());
+
+        // The conservative baseline advertises only trait-mandated operations.
+        assert!(capabilities.encrypt);
+        assert!(capabilities.decrypt);
+        assert!(capabilities.generate_data_key);
+        assert!(!capabilities.rotate);
+        assert!(!capabilities.enable_disable);
+        assert!(!capabilities.schedule_deletion);
+        assert!(!capabilities.versioning);
+        assert!(!capabilities.physical_delete);
+    }
+
+    #[tokio::test]
+    async fn local_backend_capabilities_golden() {
+        let temp_dir = tempfile::tempdir().expect("temp dir should be created");
+        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+        let backend = local::LocalKmsBackend::new(config).await.expect("local backend should build");
+
+        insta::assert_json_snapshot!("local_backend_capabilities", capabilities_snapshot(backend.capabilities()));
+    }
+
+    #[tokio::test]
+    async fn vault_kv2_backend_capabilities_golden() {
+        let config = KmsConfig::vault(
+            url::Url::parse("http://127.0.0.1:8200").expect("vault URL should parse"),
+            "dev-token".to_string(),
+        )
+        .with_insecure_development_defaults();
+        // Constructing the client performs no network I/O with token auth.
+        let backend = vault::VaultKmsBackend::new(config)
+            .await
+            .expect("vault kv2 backend should build");
+
+        insta::assert_json_snapshot!("vault_kv2_backend_capabilities", capabilities_snapshot(backend.capabilities()));
+    }
+
+    #[tokio::test]
+    async fn vault_transit_backend_capabilities_golden() {
+        let config = KmsConfig::vault_transit(
+            url::Url::parse("http://127.0.0.1:8200").expect("vault URL should parse"),
+            "dev-token".to_string(),
+        )
+        .with_insecure_development_defaults();
+        // Constructing the client performs no network I/O with token auth.
+        let backend = vault_transit::VaultTransitKmsBackend::new(config)
+            .await
+            .expect("vault transit backend should build");
+
+        insta::assert_json_snapshot!("vault_transit_backend_capabilities", capabilities_snapshot(backend.capabilities()));
+    }
+
+    #[tokio::test]
+    async fn static_backend_capabilities_golden() {
+        let config = KmsConfig::static_kms("static-key".to_string(), BASE64.encode([0u8; 32]));
+        let backend = static_kms::StaticKmsBackend::new(config)
+            .await
+            .expect("static backend should build");
+
+        insta::assert_json_snapshot!("static_backend_capabilities", capabilities_snapshot(backend.capabilities()));
+    }
+}

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__local_backend_capabilities.snap

@@ -0,0 +1,14 @@
+---
+source: crates/kms/src/backends/mod.rs
+expression: capabilities_snapshot(backend.capabilities())
+---
+{
+  "decrypt": true,
+  "enable_disable": true,
+  "encrypt": true,
+  "generate_data_key": true,
+  "physical_delete": true,
+  "rotate": false,
+  "schedule_deletion": true,
+  "versioning": false
+}

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__static_backend_capabilities.snap

@@ -0,0 +1,14 @@
+---
+source: crates/kms/src/backends/mod.rs
+expression: capabilities_snapshot(backend.capabilities())
+---
+{
+  "decrypt": true,
+  "enable_disable": false,
+  "encrypt": true,
+  "generate_data_key": true,
+  "physical_delete": false,
+  "rotate": false,
+  "schedule_deletion": false,
+  "versioning": false
+}

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__vault_kv2_backend_capabilities.snap

@@ -0,0 +1,14 @@
+---
+source: crates/kms/src/backends/mod.rs
+expression: capabilities_snapshot(backend.capabilities())
+---
+{
+  "decrypt": true,
+  "enable_disable": true,
+  "encrypt": true,
+  "generate_data_key": true,
+  "physical_delete": true,
+  "rotate": false,
+  "schedule_deletion": true,
+  "versioning": false
+}

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__vault_transit_backend_capabilities.snap

@@ -0,0 +1,14 @@
+---
+source: crates/kms/src/backends/mod.rs
+expression: capabilities_snapshot(backend.capabilities())
+---
+{
+  "decrypt": true,
+  "enable_disable": true,
+  "encrypt": true,
+  "generate_data_key": true,
+  "physical_delete": true,
+  "rotate": true,
+  "schedule_deletion": true,
+  "versioning": true
+}

crates/kms/src/backends/static_kms.rs

@@ -21,7 +21,7 @@
 //!
 //! encrypted_data(plaintext_len+16) || nonce (12 bytes)
 
-use crate::backends::{BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient};
 use crate::config::{BackendConfig, KmsConfig};
 use crate::encryption::DataKeyEnvelope;
 use crate::error::{KmsError, Result};
@@ -435,6 +435,12 @@ impl KmsBackend for StaticKmsBackend {
     async fn health_check(&self) -> Result<bool> {
         Ok(true)
     }
+
+    fn capabilities(&self) -> BackendCapabilities {
+        // Static KMS is a read-only single-key backend: it only performs
+        // cryptographic operations and rejects every lifecycle mutation.
+        BackendCapabilities::minimal()
+    }
 }
 
 #[cfg(test)]

crates/kms/src/backends/vault.rs

@@ -15,7 +15,10 @@
 //! Vault-based KMS backend implementation using vaultrs
 
 use crate::backends::vault_credentials::{VaultClientHandle, VaultConnectionSettings, VaultCredentialProvider, token_source_for};
-use crate::backends::{BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{
+    BackendCapabilities, BackendInfo, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits,
+    ensure_key_status_permits,
+};
 use crate::config::{KmsConfig, VaultConfig};
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
 use crate::error::{KmsError, Result};
@@ -464,6 +467,9 @@ impl KmsClient for VaultKmsClient {
     async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
         debug!("Generating data key for master key: {}", request.master_key_id);
 
+        let key_data = self.get_key_data(&request.master_key_id).await?;
+        ensure_key_status_permits(&request.master_key_id, &key_data.status, StateGatedOperation::GenerateDataKey)?;
+
         // Generate random data key material using the existing method
         let plaintext_key = generate_key_material(&request.key_spec)?;
 
@@ -502,8 +508,9 @@ impl KmsClient for VaultKmsClient {
     async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
         debug!("Encrypting data with key: {}", request.key_id);
 
-        // Get the master key
+        // Get the master key and verify its state allows encryption
         let key_data = self.get_key_data(&request.key_id).await?;
+        ensure_key_status_permits(&request.key_id, &key_data.status, StateGatedOperation::Encrypt)?;
         let key_material = self.decrypt_key_material(&key_data.encrypted_key_material).await?;
 
         // For simplicity, we'll use a basic encryption approach
@@ -670,6 +677,7 @@ impl KmsClient for VaultKmsClient {
         debug!("Enabling key: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
+        ensure_key_status_permits(key_id, &key_data.status, StateGatedOperation::Enable)?;
         key_data.status = KeyStatus::Active;
         self.store_key_data(key_id, &key_data).await?;
 
@@ -681,6 +689,7 @@ impl KmsClient for VaultKmsClient {
         debug!("Disabling key: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
+        ensure_key_status_permits(key_id, &key_data.status, StateGatedOperation::Disable)?;
         key_data.status = KeyStatus::Disabled;
         self.store_key_data(key_id, &key_data).await?;
 
@@ -697,6 +706,7 @@ impl KmsClient for VaultKmsClient {
         debug!("Scheduling key deletion: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
+        ensure_key_status_permits(key_id, &key_data.status, StateGatedOperation::ScheduleDeletion)?;
         key_data.status = KeyStatus::PendingDeletion;
         self.store_key_data(key_id, &key_data).await?;
 
@@ -708,6 +718,9 @@ impl KmsClient for VaultKmsClient {
         debug!("Canceling key deletion: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
+        if key_data.status != KeyStatus::PendingDeletion {
+            return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion")));
+        }
         key_data.status = KeyStatus::Active;
         self.store_key_data(key_id, &key_data).await?;
 
@@ -852,6 +865,12 @@ pub struct VaultKmsBackend {
 }
 
 impl VaultKmsBackend {
+    /// Lifecycle driver for the shared state-machine contract tests.
+    #[cfg(test)]
+    pub(crate) fn lifecycle_client(&self) -> &VaultKmsClient {
+        &self.client
+    }
+
     /// Create a new VaultKmsBackend
     pub async fn new(config: KmsConfig) -> Result<Self> {
         config.validate()?;
@@ -1037,6 +1056,8 @@ impl KmsBackend for VaultKmsBackend {
             }
         } else {
             // Schedule for deletion (default 30 days)
+            ensure_key_state_permits(key_id, &key_metadata.key_state, StateGatedOperation::ScheduleDeletion)?;
+
             let days = request.pending_window_in_days.unwrap_or(30);
             if !(7..=30).contains(&days) {
                 return Err(crate::error::KmsError::invalid_parameter(
@@ -1095,6 +1116,16 @@ impl KmsBackend for VaultKmsBackend {
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
+
+    fn capabilities(&self) -> BackendCapabilities {
+        // Rotation is unadvertised: the KV2 backend cannot rotate without
+        // replacing key material in place, and no historical versions are
+        // retained, so versioning is unsupported as well.
+        BackendCapabilities::minimal()
+            .with_enable_disable(true)
+            .with_schedule_deletion(true)
+            .with_physical_delete(true)
+    }
 }
 
 #[cfg(test)]

crates/kms/src/backends/vault_transit.rs

@@ -15,7 +15,7 @@
 //! Vault Transit-based KMS backend.
 
 use crate::backends::vault_credentials::{VaultClientHandle, VaultConnectionSettings, VaultCredentialProvider, token_source_for};
-use crate::backends::{BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits};
 use crate::config::{KmsConfig, VaultTransitConfig};
 use crate::encryption::{DataKeyEnvelope, generate_key_material};
 use crate::error::{KmsError, Result};
@@ -81,6 +81,12 @@ impl TransitKeyMetadata {
         }
     }
 
+    // KNOWN RISK (rustfs/backlog#1571, residual of rustfs/backlog#808): this
+    // fallback defaults to Enabled, so a key whose KV metadata read fails is
+    // treated as usable — a disabled or pending-deletion key can transiently
+    // "revive" on that path. State gates therefore only hold as strongly as
+    // metadata reads do. Changing the fallback is out of scope here; the
+    // synthesized_metadata_defaults_to_enabled test pins the current behavior.
     fn synthesized() -> Self {
         Self {
             key_usage: KeyUsage::EncryptDecrypt,
@@ -363,14 +369,9 @@ impl VaultTransitKmsClient {
         })
     }
 
-    async fn ensure_key_active(&self, key_id: &str) -> Result<TransitKeyMetadata> {
+    async fn ensure_key_state_allows(&self, key_id: &str, operation: StateGatedOperation) -> Result<TransitKeyMetadata> {
         let metadata = self.get_key_metadata(key_id).await?;
-        if metadata.key_state != KeyState::Enabled {
-            return Err(KmsError::invalid_operation(format!(
-                "Key {key_id} is not active (state: {:?})",
-                metadata.key_state
-            )));
-        }
+        ensure_key_state_permits(key_id, &metadata.key_state, operation)?;
         Ok(metadata)
     }
 }
@@ -378,7 +379,8 @@ impl VaultTransitKmsClient {
 #[async_trait]
 impl KmsClient for VaultTransitKmsClient {
     async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
-        self.ensure_key_active(&request.master_key_id).await?;
+        self.ensure_key_state_allows(&request.master_key_id, StateGatedOperation::GenerateDataKey)
+            .await?;
 
         let plaintext_key = generate_key_material(&request.key_spec)?;
         let encrypted_key = self
@@ -409,7 +411,9 @@ impl KmsClient for VaultTransitKmsClient {
     }
 
     async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
-        let metadata = self.ensure_key_active(&request.key_id).await?;
+        let metadata = self
+            .ensure_key_state_allows(&request.key_id, StateGatedOperation::Encrypt)
+            .await?;
         let ciphertext = self
             .transit_encrypt(&request.key_id, &request.plaintext, &request.encryption_context)
             .await?;
@@ -521,14 +525,16 @@ impl KmsClient for VaultTransitKmsClient {
     }
 
     async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        let mut metadata = self.get_key_metadata(key_id).await?;
+        // A pending deletion must be reverted through cancel_key_deletion, not
+        // silently by enabling, so the gate rejects PendingDeletion here.
+        let mut metadata = self.ensure_key_state_allows(key_id, StateGatedOperation::Enable).await?;
         metadata.key_state = KeyState::Enabled;
         metadata.deletion_date = None;
         self.store_key_metadata(key_id, &metadata).await
     }
 
     async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        let mut metadata = self.get_key_metadata(key_id).await?;
+        let mut metadata = self.ensure_key_state_allows(key_id, StateGatedOperation::Disable).await?;
         metadata.key_state = KeyState::Disabled;
         self.store_key_metadata(key_id, &metadata).await
     }
@@ -539,7 +545,9 @@ impl KmsClient for VaultTransitKmsClient {
         pending_window_days: u32,
         _context: Option<&OperationContext>,
     ) -> Result<()> {
-        let mut metadata = self.get_key_metadata(key_id).await?;
+        let mut metadata = self
+            .ensure_key_state_allows(key_id, StateGatedOperation::ScheduleDeletion)
+            .await?;
         metadata.key_state = KeyState::PendingDeletion;
         metadata.deletion_date = Some(Zoned::now() + Duration::from_secs(pending_window_days as u64 * 86400));
         self.store_key_metadata(key_id, &metadata).await
@@ -547,12 +555,17 @@ impl KmsClient for VaultTransitKmsClient {
 
     async fn cancel_key_deletion(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         let mut metadata = self.get_key_metadata(key_id).await?;
+        if metadata.key_state != KeyState::PendingDeletion {
+            return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion")));
+        }
         metadata.key_state = KeyState::Enabled;
         metadata.deletion_date = None;
         self.store_key_metadata(key_id, &metadata).await
     }
 
     async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+        self.ensure_key_state_allows(key_id, StateGatedOperation::Rotate).await?;
+
         key::rotate(&self.vault().client, &self.config.mount_path, key_id)
             .await
             .map_err(|e| KmsError::backend_error(format!("Failed to rotate Vault Transit key {key_id}: {e}")))?;
@@ -593,6 +606,14 @@ pub struct VaultTransitKmsBackend {
 }
 
 impl VaultTransitKmsBackend {
+    /// Lifecycle driver for the shared state-machine contract tests. Using the
+    /// backend's own client keeps its in-process metadata cache coherent with
+    /// the transitions the tests perform.
+    #[cfg(test)]
+    pub(crate) fn lifecycle_client(&self) -> &VaultTransitKmsClient {
+        &self.client
+    }
+
     pub async fn new(config: KmsConfig) -> Result<Self> {
         config.validate()?;
 
@@ -722,6 +743,8 @@ impl KmsBackend for VaultTransitKmsBackend {
                 None
             }
         } else {
+            ensure_key_state_permits(&key_id, &key_metadata.key_state, StateGatedOperation::ScheduleDeletion)?;
+
             let days = request.pending_window_in_days.unwrap_or(30);
             if !(7..=30).contains(&days) {
                 return Err(KmsError::invalid_parameter("pending_window_in_days must be between 7 and 30"));
@@ -762,6 +785,18 @@ impl KmsBackend for VaultTransitKmsBackend {
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
+
+    fn capabilities(&self) -> BackendCapabilities {
+        // Vault Transit natively supports version-retaining rotation, keeps
+        // prior versions addressable for decryption, and allows physical
+        // deletion once a key is pending deletion.
+        BackendCapabilities::minimal()
+            .with_rotate(true)
+            .with_enable_disable(true)
+            .with_schedule_deletion(true)
+            .with_versioning(true)
+            .with_physical_delete(true)
+    }
 }
 
 #[cfg(test)]
@@ -973,4 +1008,15 @@ mod tests {
         // Cleanup so repeated runs against the same Vault do not accumulate keys.
         let _ = client.schedule_key_deletion(&key_id, 7, None).await;
     }
+
+    /// Pins the known-risk fallback documented on `TransitKeyMetadata::synthesized`:
+    /// when KV metadata cannot be read, the synthesized record defaults to Enabled,
+    /// which weakens every state gate on that path. If this test turns red the
+    /// fallback semantics changed on purpose — update the comment there as well.
+    #[test]
+    fn synthesized_metadata_defaults_to_enabled() {
+        let metadata = TransitKeyMetadata::synthesized();
+        assert_eq!(metadata.key_state, KeyState::Enabled);
+        assert!(metadata.deletion_date.is_none());
+    }
 }

crates/kms/src/error.rs

@@ -128,6 +128,10 @@ pub enum KmsError {
     /// Backup/restore bundle contract violation; see [`crate::backup::BackupError`]
     #[error(transparent)]
     Backup(#[from] crate::backup::BackupError),
+
+    /// Operation is not supported by the active KMS backend
+    #[error("Operation '{operation}' is not supported by KMS backend '{backend}'")]
+    UnsupportedCapability { backend: String, operation: String },
 }
 
 impl KmsError {
@@ -269,6 +273,14 @@ impl KmsError {
             version,
         }
     }
+
+    /// Create an unsupported capability error
+    pub fn unsupported_capability<S1: Into<String>, S2: Into<String>>(backend: S1, operation: S2) -> Self {
+        Self::UnsupportedCapability {
+            backend: backend.into(),
+            operation: operation.into(),
+        }
+    }
 }
 
 /// Convert from standard library errors

crates/kms/src/manager.rs

@@ -159,6 +159,11 @@ impl KmsManager {
     pub async fn health_check(&self) -> Result<bool> {
         self.backend.health_check().await
     }
+
+    /// Report the capabilities of the configured backend
+    pub fn backend_capabilities(&self) -> crate::backends::BackendCapabilities {
+        self.backend.capabilities()
+    }
 }
 
 #[cfg(test)]

crates/kms/src/service.rs

@@ -184,6 +184,15 @@ impl ObjectEncryptionService {
         self.kms_manager.health_check().await
     }
 
+    /// Report the capabilities of the configured backend
+    ///
+    /// # Returns
+    /// The capability matrix advertised by the active KMS backend
+    ///
+    pub fn backend_capabilities(&self) -> crate::backends::BackendCapabilities {
+        self.kms_manager.backend_capabilities()
+    }
+
     /// Create a data encryption key for object encryption
     ///
     /// # Arguments

rustfs/src/admin/handlers/kms_management.rs

@@ -72,6 +72,10 @@ pub struct KmsStatusResponse {
     pub cache_enabled: bool,
     pub cache_stats: Option<CacheStatsResponse>,
     pub default_key_id: Option<String>,
+    /// Capability matrix of the active backend. Additive field: omitted by
+    /// older servers, so it must stay optional for consumers.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub capabilities: Option<rustfs_kms::backends::BackendCapabilities>,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -204,6 +208,7 @@ impl Operation for KmsStatusHandler {
             cache_enabled: config.as_ref().is_some_and(|cfg| cfg.enable_cache),
             cache_stats,
             default_key_id: service.get_default_key_id().cloned(),
+            capabilities: Some(service.backend_capabilities()),
         };
 
         let data = serde_json::to_vec(&response).map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?;
@@ -339,4 +344,34 @@ mod tests {
     fn kms_clear_cache_rejects_server_info_fallback() {
         assert_lacks_action(&kms_clear_cache_actions(), Action::AdminAction(AdminAction::ServerInfoAdminAction));
     }
+
+    /// The `capabilities` field is additive: payloads produced by older
+    /// servers (without the field) must keep deserializing, and the field
+    /// must be omitted from JSON when unset so existing consumers see an
+    /// unchanged response shape.
+    #[test]
+    fn kms_status_response_capabilities_field_is_additive() {
+        let legacy_json = serde_json::json!({
+            "backend_type": "local",
+            "backend_status": "healthy",
+            "cache_enabled": true,
+            "cache_stats": null,
+            "default_key_id": null,
+        });
+        let legacy: super::KmsStatusResponse =
+            serde_json::from_value(legacy_json).expect("legacy status payload should deserialize");
+        assert!(legacy.capabilities.is_none());
+
+        let serialized = serde_json::to_value(&legacy).expect("status response should serialize");
+        assert!(serialized.get("capabilities").is_none(), "unset capabilities must be omitted");
+
+        let with_capabilities = super::KmsStatusResponse {
+            capabilities: Some(rustfs_kms::backends::BackendCapabilities::minimal()),
+            ..legacy
+        };
+        let serialized = serde_json::to_value(&with_capabilities).expect("status response should serialize");
+        let capabilities = serialized.get("capabilities").expect("capabilities must be present when set");
+        assert_eq!(capabilities.get("encrypt"), Some(&serde_json::Value::Bool(true)));
+        assert_eq!(capabilities.get("rotate"), Some(&serde_json::Value::Bool(false)));
+    }
 }