refactor(kms): fold the KmsClient layer into KmsBackend and complete lifecycle overrides (#5501)

crates/kms/src/backends/contract_tests.rs

@@ -27,11 +27,11 @@
 //! server, so they are `#[ignore]`d in CI. Static is covered by its own
 //! stateless contract below.
 
+use super::KmsBackend;
 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;
@@ -46,6 +46,24 @@ use rand::RngExt as _;
 use std::collections::HashMap;
 use std::sync::Arc;
 
+fn expect_unsupported<T: std::fmt::Debug>(result: Result<T>) {
+    match result {
+        Err(KmsError::UnsupportedCapability { .. }) => {}
+        other => panic!("expected UnsupportedCapability, got {other:?}"),
+    }
+}
+
+/// Rotation while not Enabled: backends with rotation support must reject it
+/// through the state machine; backends without it report the capability gap.
+async fn expect_rotate_rejected(backend: &dyn KmsBackend, key_id: &str) {
+    let result = backend.rotate_key(key_id).await;
+    if backend.capabilities().rotate {
+        expect_invalid_key_state(result, "");
+    } else {
+        expect_unsupported(result);
+    }
+}
+
 fn expect_invalid_key_state<T: std::fmt::Debug>(result: Result<T>, expected_fragment: &str) {
     match result {
         Err(KmsError::InvalidOperation { message }) => assert!(
@@ -117,11 +135,9 @@ async fn assert_key_state(backend: &dyn KmsBackend, key_id: &str, expected: KeyS
     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.
+/// Drives one freshly created (Enabled) key through the full state matrix,
-///
+/// entirely through the `KmsBackend` product surface.
-/// `backend` is the product surface; `client` drives the lifecycle
+async fn assert_state_machine_contract(backend: &dyn KmsBackend, key_id: &str) {
-/// 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
@@ -134,16 +150,13 @@ async fn assert_state_machine_contract(backend: &dyn KmsBackend, client: &dyn Km
         .expect("Enabled key must encrypt");
 
     // Enabled -> Disabled.
-    client
+    backend.disable_key(key_id).await.expect("disable from Enabled must succeed");
-        .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, "");
+    expect_rotate_rejected(backend, key_id).await;
     // ...but decryption of existing data keeps working (explicit AWS deviation)...
     let decrypted = backend
         .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
@@ -151,17 +164,14 @@ async fn assert_state_machine_contract(backend: &dyn KmsBackend, client: &dyn Km
         .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");
+    backend.disable_key(key_id).await.expect("disable must be idempotent");
     expect_invalid_key_state(backend.cancel_key_deletion(cancel_request(key_id)).await, "not pending deletion");
-    client
+    backend.enable_key(key_id).await.expect("enable from Disabled must succeed");
-        .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
+    backend
-        .disable_key(key_id, None)
+        .disable_key(key_id)
         .await
         .expect("disable before scheduling must succeed");
     backend
@@ -173,10 +183,9 @@ async fn assert_state_machine_contract(backend: &dyn KmsBackend, client: &dyn Km
     // 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(backend.enable_key(key_id).await, "pending deletion");
-    expect_invalid_key_state(client.disable_key(key_id, None).await, "pending deletion");
+    expect_invalid_key_state(backend.disable_key(key_id).await, "pending deletion");
-    expect_invalid_key_state(client.rotate_key(key_id, None).await, "");
+    expect_rotate_rejected(backend, key_id).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()))
@@ -215,7 +224,7 @@ async fn local_fixture() -> (tempfile::TempDir, KmsConfig, LocalKmsBackend, Stri
 #[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;
+    assert_state_machine_contract(&backend, &key_id).await;
 }
 
 /// SSE-shaped regression: disabling a key must not break decryption of data
@@ -256,10 +265,7 @@ async fn static_backend_stateless_contract() {
     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))
@@ -275,9 +281,11 @@ async fn static_backend_stateless_contract() {
     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");
+    // Enable/disable and rotation are capability gaps at the product
-    expect_invalid_key_state(client.schedule_key_deletion(key_id, 7, None).await, "read-only");
+    // surface, not state-machine rejections.
-    expect_invalid_key_state(client.rotate_key(key_id, None).await, "read-only");
+    expect_unsupported(backend.enable_key(key_id).await);
+    expect_unsupported(backend.disable_key(key_id).await);
+    expect_unsupported(backend.rotate_key(key_id).await);
 }
 
 fn vault_dev_config(constructor: fn(url::Url, String) -> KmsConfig) -> KmsConfig {
@@ -298,7 +306,15 @@ async fn vault_kv2_backend_state_machine_contract() {
         .await
         .expect("key should be created");
 
-    assert_state_machine_contract(&backend, backend.lifecycle_client(), &created.key_id).await;
+    assert_state_machine_contract(&backend, &created.key_id).await;
+
+    // KV2 additionally supports version-retaining rotation, which must only
+    // work while the key is Enabled (the shared matrix covered the
+    // rejections).
+    backend
+        .rotate_key(&created.key_id)
+        .await
+        .expect("rotation of an Enabled KV2 key must succeed");
 
     // Cleanup: leave the key pending deletion so repeated runs stay tidy.
     let _ = backend.delete_key(schedule_request(&created.key_id)).await;
@@ -316,13 +332,12 @@ async fn vault_transit_backend_state_machine_contract() {
         .await
         .expect("key should be created");
 
-    assert_state_machine_contract(&backend, backend.lifecycle_client(), &created.key_id).await;
+    assert_state_machine_contract(&backend, &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)
-        .rotate_key(&created.key_id, None)
         .await
         .expect("rotation of an Enabled transit key must succeed");
 

crates/kms/src/backends/local.rs

@@ -14,9 +14,7 @@
 
 //! Local file-based KMS backend implementation
 
-use crate::backends::{
+use crate::backends::{BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_status_permits};
-    BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_status_permits,
-};
 use crate::config::KmsConfig;
 use crate::config::LocalConfig;
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -937,9 +935,12 @@ impl LocalKmsClient {
     }
 }
 
-#[async_trait]
+impl LocalKmsClient {
-impl KmsClient for LocalKmsClient {
+    pub(crate) async fn generate_data_key(
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, context: Option<&OperationContext>) -> Result<DataKeyInfo> {
+        &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?;
@@ -980,7 +981,7 @@ impl KmsClient for LocalKmsClient {
         Ok(data_key)
     }
 
-    async fn encrypt(&self, request: &EncryptRequest, context: Option<&OperationContext>) -> Result<EncryptResponse> {
+    pub(crate) async fn encrypt(&self, request: &EncryptRequest, context: Option<&OperationContext>) -> Result<EncryptResponse> {
         debug!("Encrypting data with key: {}", request.key_id);
 
         // Verify key exists and its state allows encryption
@@ -997,7 +998,7 @@ impl KmsClient for LocalKmsClient {
         })
     }
 
-    async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
+    pub(crate) async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
         debug!("Decrypting data");
 
         // Parse the data key envelope from ciphertext
@@ -1031,7 +1032,14 @@ impl KmsClient for LocalKmsClient {
         Ok(plaintext)
     }
 
-    async fn create_key(&self, key_id: &str, algorithm: &str, context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn create_key(
+        &self,
+        key_id: &str,
+        algorithm: &str,
+        context: Option<&OperationContext>,
+    ) -> Result<MasterKeyInfo> {
         debug!("Creating master key: {}", key_id);
 
         // Check if key already exists
@@ -1060,14 +1068,18 @@ impl KmsClient for LocalKmsClient {
         Ok(master_key)
     }
 
-    async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
+    pub(crate) async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
         debug!("Describing key: {}", key_id);
 
         let master_key = self.load_master_key(key_id).await?;
         Ok(master_key.into())
     }
 
-    async fn list_keys(&self, request: &ListKeysRequest, _context: Option<&OperationContext>) -> Result<ListKeysResponse> {
+    pub(crate) async fn list_keys(
+        &self,
+        request: &ListKeysRequest,
+        _context: Option<&OperationContext>,
+    ) -> Result<ListKeysResponse> {
         debug!("Listing keys");
 
         let mut keys = Vec::new();
@@ -1111,7 +1123,7 @@ impl KmsClient for LocalKmsClient {
         })
     }
 
-    async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         debug!("Enabling key: {}", key_id);
 
         let _write_guard = self.lock_key_for_write(key_id).await;
@@ -1129,7 +1141,7 @@ impl KmsClient for LocalKmsClient {
         Ok(())
     }
 
-    async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         debug!("Disabling key: {}", key_id);
 
         let _write_guard = self.lock_key_for_write(key_id).await;
@@ -1146,7 +1158,9 @@ impl KmsClient for LocalKmsClient {
         Ok(())
     }
 
-    async fn schedule_key_deletion(
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn schedule_key_deletion(
         &self,
         key_id: &str,
         pending_window_days: u32,
@@ -1170,7 +1184,9 @@ impl KmsClient for LocalKmsClient {
         Ok(())
     }
 
-    async fn cancel_key_deletion(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn cancel_key_deletion(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         debug!("Canceling deletion for key: {}", key_id);
 
         let _write_guard = self.lock_key_for_write(key_id).await;
@@ -1190,7 +1206,9 @@ impl KmsClient for LocalKmsClient {
         Ok(())
     }
 
-    async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
         if !fs::try_exists(self.master_key_path(key_id)?).await? {
             return Err(KmsError::key_not_found(key_id));
         }
@@ -1199,7 +1217,7 @@ impl KmsClient for LocalKmsClient {
         ))
     }
 
-    async fn health_check(&self) -> Result<()> {
+    pub(crate) async fn health_check(&self) -> Result<()> {
         // Check if key directory is accessible
         if !self.config.key_dir.exists() {
             return Err(KmsError::backend_error("Key directory does not exist"));
@@ -1210,17 +1228,6 @@ impl KmsClient for LocalKmsClient {
 
         Ok(())
     }
-
-    fn backend_info(&self) -> BackendInfo {
-        BackendInfo::new(
-            "local".to_string(),
-            env!("CARGO_PKG_VERSION").to_string(),
-            self.config.key_dir.to_string_lossy().to_string(),
-            true, // We'll assume healthy for now
-        )
-        .with_metadata("key_dir".to_string(), self.config.key_dir.to_string_lossy().to_string())
-        .with_metadata("encrypted_at_rest".to_string(), self.master_cipher.is_some().to_string())
-    }
 }
 
 /// LocalKmsBackend wraps LocalKmsClient and implements the KmsBackend trait

crates/kms/src/backends/mod.rs

@@ -19,7 +19,6 @@ use crate::types::*;
 use async_trait::async_trait;
 use jiff::Zoned;
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
 
 #[cfg(test)]
 mod contract_tests;
@@ -99,136 +98,6 @@ pub(crate) fn ensure_key_status_permits(key_id: &str, status: &KeyStatus, operat
     ensure_key_state_permits(key_id, &state, operation)
 }
 
-/// Abstract KMS client interface that all backends must implement
-#[async_trait]
-pub trait KmsClient: Send + Sync {
-    /// Generate a new data encryption key (DEK)
-    ///
-    /// Creates a new data key using the specified master key. The returned DataKey
-    /// contains both the plaintext and encrypted versions of the key.
-    ///
-    /// # Arguments
-    /// * `request` - The key generation request
-    /// * `context` - Optional operation context for auditing
-    ///
-    /// # Returns
-    /// Returns a DataKey containing both plaintext and encrypted key material
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, context: Option<&OperationContext>) -> Result<DataKeyInfo>;
-
-    /// Encrypt data directly using a master key
-    ///
-    /// Encrypts the provided plaintext using the specified master key.
-    /// This is different from generate_data_key as it encrypts user data directly.
-    ///
-    /// # Arguments
-    /// * `request` - The encryption request containing plaintext and key ID
-    /// * `context` - Optional operation context for auditing
-    async fn encrypt(&self, request: &EncryptRequest, context: Option<&OperationContext>) -> Result<EncryptResponse>;
-
-    /// Decrypt data using a master key
-    ///
-    /// Decrypts the provided ciphertext. The KMS automatically determines
-    /// which key was used for encryption based on the ciphertext metadata.
-    ///
-    /// # Arguments
-    /// * `request` - The decryption request containing ciphertext
-    /// * `context` - Optional operation context for auditing
-    async fn decrypt(&self, request: &DecryptRequest, context: Option<&OperationContext>) -> Result<Vec<u8>>;
-
-    /// Create a new master key
-    ///
-    /// Creates a new master key in the KMS with the specified ID.
-    /// Returns an error if a key with the same ID already exists.
-    ///
-    /// # Arguments
-    /// * `key_id` - Unique identifier for the new key
-    /// * `algorithm` - Key algorithm (e.g., "AES_256")
-    /// * `context` - Optional operation context for auditing
-    async fn create_key(&self, key_id: &str, algorithm: &str, context: Option<&OperationContext>) -> Result<MasterKeyInfo>;
-
-    /// Get information about a specific key
-    ///
-    /// Returns metadata and information about the specified key.
-    ///
-    /// # Arguments
-    /// * `key_id` - The key identifier
-    /// * `context` - Optional operation context for auditing
-    async fn describe_key(&self, key_id: &str, context: Option<&OperationContext>) -> Result<KeyInfo>;
-
-    /// List available keys
-    ///
-    /// Returns a paginated list of keys available in the KMS.
-    ///
-    /// # Arguments
-    /// * `request` - List request parameters (pagination, filters)
-    /// * `context` - Optional operation context for auditing
-    async fn list_keys(&self, request: &ListKeysRequest, context: Option<&OperationContext>) -> Result<ListKeysResponse>;
-
-    /// Enable a key
-    ///
-    /// Enables a previously disabled key, allowing it to be used for cryptographic operations.
-    ///
-    /// # Arguments
-    /// * `key_id` - The key identifier
-    /// * `context` - Optional operation context for auditing
-    async fn enable_key(&self, key_id: &str, context: Option<&OperationContext>) -> Result<()>;
-
-    /// Disable a key
-    ///
-    /// Disables a key, preventing it from being used for new cryptographic operations.
-    /// Existing encrypted data can still be decrypted.
-    ///
-    /// # Arguments
-    /// * `key_id` - The key identifier
-    /// * `context` - Optional operation context for auditing
-    async fn disable_key(&self, key_id: &str, context: Option<&OperationContext>) -> Result<()>;
-
-    /// Schedule key deletion
-    ///
-    /// Schedules a key for deletion after a specified number of days.
-    /// This allows for a grace period to recover the key if needed.
-    ///
-    /// # Arguments
-    /// * `key_id` - The key identifier
-    /// * `pending_window_days` - Number of days before actual deletion
-    /// * `context` - Optional operation context for auditing
-    async fn schedule_key_deletion(
-        &self,
-        key_id: &str,
-        pending_window_days: u32,
-        context: Option<&OperationContext>,
-    ) -> Result<()>;
-
-    /// Cancel key deletion
-    ///
-    /// Cancels a previously scheduled key deletion.
-    ///
-    /// # Arguments
-    /// * `key_id` - The key identifier
-    /// * `context` - Optional operation context for auditing
-    async fn cancel_key_deletion(&self, key_id: &str, context: Option<&OperationContext>) -> Result<()>;
-
-    /// Rotate a key
-    ///
-    /// Creates a new version of the specified key. Previous versions remain
-    /// available for decryption but new operations will use the new version.
-    ///
-    /// # Arguments
-    /// * `key_id` - The key identifier
-    /// * `context` - Optional operation context for auditing
-    async fn rotate_key(&self, key_id: &str, context: Option<&OperationContext>) -> Result<MasterKeyInfo>;
-
-    /// Health check
-    ///
-    /// Performs a health check on the KMS backend to ensure it's operational.
-    async fn health_check(&self) -> Result<()>;
-
-    /// Get backend information
-    ///
-    /// Returns information about the KMS backend (type, version, etc.).
-    fn backend_info(&self) -> BackendInfo;
-}
-
 /// Simplified KMS backend interface for manager
 #[async_trait]
 pub trait KmsBackend: Send + Sync {
@@ -327,58 +196,6 @@ pub enum ExpiredKeyRemoval {
     NotExpired,
 }
 
-/// Information about a KMS backend
-#[derive(Debug, Clone)]
-pub struct BackendInfo {
-    /// Backend type name (e.g., "local", "vault")
-    pub backend_type: String,
-    /// Backend version
-    pub version: String,
-    /// Backend endpoint or location
-    pub endpoint: String,
-    /// Whether the backend is currently healthy
-    pub healthy: bool,
-    /// Additional metadata about the backend
-    pub metadata: HashMap<String, String>,
-}
-
-impl BackendInfo {
-    /// Create a new backend info
-    ///
-    /// # Arguments
-    /// * `backend_type` - The type of the backend
-    /// * `version` - The version of the backend
-    /// * `endpoint` - The endpoint or location of the backend
-    /// * `healthy` - Whether the backend is healthy
-    ///
-    /// # Returns
-    /// A new BackendInfo instance
-    ///
-    pub fn new(backend_type: String, version: String, endpoint: String, healthy: bool) -> Self {
-        Self {
-            backend_type,
-            version,
-            endpoint,
-            healthy,
-            metadata: HashMap::new(),
-        }
-    }
-
-    /// Add metadata to the backend info
-    ///
-    /// # Arguments
-    /// * `key` - Metadata key
-    /// * `value` - Metadata value
-    ///
-    /// # Returns
-    /// Updated BackendInfo instance
-    ///
-    pub fn with_metadata(mut self, key: String, value: String) -> Self {
-        self.metadata.insert(key, value);
-        self
-    }
-}
-
 /// Set of operations a KMS backend supports.
 ///
 /// Reported by [`KmsBackend::capabilities`] so callers (manager, admin API)

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

@@ -8,7 +8,7 @@ expression: capabilities_snapshot(backend.capabilities())
   "encrypt": true,
   "generate_data_key": true,
   "physical_delete": true,
-  "rotate": false,
+  "rotate": true,
   "schedule_deletion": true,
-  "versioning": false
+  "versioning": true
 }

crates/kms/src/backends/static_kms.rs

@@ -21,7 +21,7 @@
 //!
 //! encrypted_data(plaintext_len+16) || nonce (12 bytes)
 
-use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{BackendCapabilities, KmsBackend};
 use crate::config::{BackendConfig, KmsConfig};
 use crate::encryption::DataKeyEnvelope;
 use crate::error::{KmsError, Result};
@@ -98,9 +98,10 @@ impl StaticKmsBackend {
     }
 }
 
-#[async_trait]
+impl StaticKmsBackend {
-impl KmsClient for StaticKmsBackend {
+    /// Generate a fresh data key and wrap it in the standard KMS envelope,
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
+    /// authenticated against the canonical encryption context.
+    pub(crate) fn generate_data_key_envelope(&self, request: &GenerateKeyRequest) -> Result<DataKeyInfo> {
         if request.master_key_id != self.key_id {
             return Err(KmsError::key_not_found(&request.master_key_id));
         }
@@ -151,7 +152,8 @@ impl KmsClient for StaticKmsBackend {
         ))
     }
 
-    async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
+    /// Encrypt caller-provided plaintext into the standard KMS envelope.
+    pub(crate) fn encrypt_to_envelope(&self, request: &EncryptRequest) -> Result<EncryptResponse> {
         if request.key_id != self.key_id {
             return Err(KmsError::key_not_found(&request.key_id));
         }
@@ -196,7 +198,8 @@ impl KmsClient for StaticKmsBackend {
         })
     }
 
-    async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
+    /// Open a KMS envelope produced by this backend.
+    pub(crate) fn decrypt_envelope(&self, request: &DecryptRequest) -> Result<Vec<u8>> {
         let envelope: DataKeyEnvelope = serde_json::from_slice(&request.ciphertext)
             .map_err(|error| KmsError::cryptographic_error("parse", format!("Failed to parse data key envelope: {error}")))?;
         if envelope.master_key_id != self.key_id {
@@ -235,14 +238,8 @@ impl KmsClient for StaticKmsBackend {
         Ok(plaintext)
     }
 
-    async fn create_key(&self, key_id: &str, _algorithm: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    /// Describe the single configured key.
-        if key_id == self.key_id {
+    pub(crate) fn configured_key_info(&self, key_id: &str) -> Result<KeyInfo> {
-            return Err(KmsError::key_already_exists(key_id));
-        }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot create new keys"))
-    }
-
-    async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
         if key_id != self.key_id {
             return Err(KmsError::key_not_found(key_id));
         }
@@ -263,7 +260,8 @@ impl KmsClient for StaticKmsBackend {
         })
     }
 
-    async fn list_keys(&self, request: &ListKeysRequest, _context: Option<&OperationContext>) -> Result<ListKeysResponse> {
+    /// List the single configured key, honouring the pagination marker.
+    pub(crate) fn list_configured_key(&self, request: &ListKeysRequest) -> Result<ListKeysResponse> {
         let key_info = KeyInfo {
             key_id: self.key_id.clone(),
             description: Some("Static single-key KMS backend".to_string()),
@@ -295,57 +293,6 @@ impl KmsClient for StaticKmsBackend {
             truncated: false,
         })
     }
-
-    async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        if key_id != self.key_id {
-            return Err(KmsError::key_not_found(key_id));
-        }
-        // Static KMS key is always enabled
-        Ok(())
-    }
-
-    async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        if key_id != self.key_id {
-            return Err(KmsError::key_not_found(key_id));
-        }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot disable keys"))
-    }
-
-    async fn schedule_key_deletion(
-        &self,
-        key_id: &str,
-        _pending_window_days: u32,
-        _context: Option<&OperationContext>,
-    ) -> Result<()> {
-        if key_id != self.key_id {
-            return Err(KmsError::key_not_found(key_id));
-        }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot schedule key deletion"))
-    }
-
-    async fn cancel_key_deletion(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        if key_id != self.key_id {
-            return Err(KmsError::key_not_found(key_id));
-        }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot cancel key deletion"))
-    }
-
-    async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
-        if key_id != self.key_id {
-            return Err(KmsError::key_not_found(key_id));
-        }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot rotate keys"))
-    }
-
-    async fn health_check(&self) -> Result<()> {
-        // Static KMS is always healthy if it was successfully initialized
-        Ok(())
-    }
-
-    fn backend_info(&self) -> BackendInfo {
-        BackendInfo::new("static".to_string(), env!("CARGO_PKG_VERSION").to_string(), "local".to_string(), true)
-            .with_metadata("key_id".to_string(), self.key_id.clone())
-    }
 }
 
 #[async_trait]
@@ -359,12 +306,12 @@ impl KmsBackend for StaticKmsBackend {
     }
 
     async fn encrypt(&self, request: EncryptRequest) -> Result<EncryptResponse> {
-        <Self as KmsClient>::encrypt(self, &request, None).await
+        self.encrypt_to_envelope(&request)
     }
 
     async fn decrypt(&self, request: DecryptRequest) -> Result<DecryptResponse> {
         let key_id = self.key_id.clone();
-        let plaintext = <Self as KmsClient>::decrypt(self, &request, None).await?;
+        let plaintext = self.decrypt_envelope(&request)?;
         Ok(DecryptResponse {
             plaintext,
             key_id,
@@ -380,7 +327,7 @@ impl KmsBackend for StaticKmsBackend {
             encryption_context: request.encryption_context,
             grant_tokens: Vec::new(),
         };
-        let data_key = <Self as KmsClient>::generate_data_key(self, &gen_req, None).await?;
+        let data_key = self.generate_data_key_envelope(&gen_req)?;
 
         let plaintext_key = data_key
             .plaintext
@@ -395,7 +342,7 @@ impl KmsBackend for StaticKmsBackend {
     }
 
     async fn describe_key(&self, request: DescribeKeyRequest) -> Result<DescribeKeyResponse> {
-        let key_info = <Self as KmsClient>::describe_key(self, &request.key_id, None).await?;
+        let key_info = self.configured_key_info(&request.key_id)?;
         let key_metadata = KeyMetadata {
             key_id: key_info.key_id.clone(),
             key_state: if key_info.status == KeyStatus::Active {
@@ -415,7 +362,7 @@ impl KmsBackend for StaticKmsBackend {
     }
 
     async fn list_keys(&self, request: ListKeysRequest) -> Result<ListKeysResponse> {
-        <Self as KmsClient>::list_keys(self, &request, None).await
+        self.list_configured_key(&request)
     }
 
     async fn delete_key(&self, request: DeleteKeyRequest) -> Result<DeleteKeyResponse> {
@@ -446,7 +393,7 @@ impl KmsBackend for StaticKmsBackend {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::backends::{KmsBackend as KmsBackendTrait, KmsClient};
+    use crate::backends::KmsBackend as KmsBackendTrait;
     use crate::config::{BackendConfig, KmsBackend, StaticConfig};
     use crate::encryption::is_data_key_envelope;
     use base64::Engine as _;
@@ -491,8 +438,8 @@ mod tests {
         // Generate data key
         let request = GenerateKeyRequest::new(key_id.clone(), "AES_256".to_string())
             .with_context("bucket".to_string(), "test-bucket".to_string());
-        let data_key = KmsClient::generate_data_key(&backend, &request, None)
+        let data_key = backend
-            .await
+            .generate_data_key_envelope(&request)
             .expect("Failed to generate data key");
 
         assert_eq!(data_key.key_id, key_id);
@@ -509,9 +456,7 @@ mod tests {
         // Decrypt the data key
         let decrypt_request =
             DecryptRequest::new(data_key.ciphertext.clone()).with_context("bucket".to_string(), "test-bucket".to_string());
-        let decrypted = KmsClient::decrypt(&backend, &decrypt_request, None)
+        let decrypted = backend.decrypt_envelope(&decrypt_request).expect("Failed to decrypt");
-            .await
-            .expect("Failed to decrypt");
 
         assert_eq!(decrypted.as_slice(), data_key.plaintext.as_deref().expect("plaintext should exist"));
     }
@@ -523,8 +468,8 @@ mod tests {
             .with_context("bucket".to_string(), "source-bucket".to_string())
             .with_context("object".to_string(), "source-object".to_string());
 
-        let data_key = KmsClient::generate_data_key(&backend, &request, None)
+        let data_key = backend
-            .await
+            .generate_data_key_envelope(&request)
             .expect("generate static KMS data key");
 
         assert!(
@@ -568,8 +513,8 @@ mod tests {
         let (backend, key_id, _key) = create_test_backend().await;
         let request = GenerateKeyRequest::new(key_id, "AES_256".to_string())
             .with_context("bucket".to_string(), "source-bucket".to_string());
-        let generated = KmsClient::generate_data_key(&backend, &request, None)
+        let generated = backend
-            .await
+            .generate_data_key_envelope(&request)
             .expect("generate context-bound data key");
         let mut envelope: DataKeyEnvelope = serde_json::from_slice(&generated.ciphertext).expect("parse static KMS envelope");
         envelope
@@ -578,8 +523,8 @@ mod tests {
         let decrypt_request = DecryptRequest::new(serde_json::to_vec(&envelope).expect("serialize tampered envelope"))
             .with_context("bucket".to_string(), "different-bucket".to_string());
 
-        let error = KmsClient::decrypt(&backend, &decrypt_request, None)
+        let error = backend
-            .await
+            .decrypt_envelope(&decrypt_request)
             .expect_err("tampering with authenticated envelope context must fail");
 
         assert!(matches!(error, KmsError::CryptographicError { .. }));
@@ -590,7 +535,7 @@ mod tests {
         let (backend, _key_id, _key) = create_test_backend().await;
 
         let request = GenerateKeyRequest::new("wrong-key-id".to_string(), "AES_256".to_string());
-        let result = KmsClient::generate_data_key(&backend, &request, None).await;
+        let result = backend.generate_data_key_envelope(&request);
         assert!(result.is_err());
         assert!(result.expect_err("should be Err").to_string().contains("wrong-key-id"));
     }
@@ -602,7 +547,7 @@ mod tests {
         // Ciphertext too short
         let short = vec![0u8; 10];
         let request = DecryptRequest::new(short);
-        let result = KmsClient::decrypt(&backend, &request, None).await;
+        let result = backend.decrypt_envelope(&request);
         assert!(result.is_err());
     }
 
@@ -612,9 +557,7 @@ mod tests {
 
         // Generate a valid ciphertext first
         let gen_request = GenerateKeyRequest::new(key_id, "AES_256".to_string());
-        let data_key = KmsClient::generate_data_key(&backend, &gen_request, None)
+        let data_key = backend.generate_data_key_envelope(&gen_request).expect("generate");
-            .await
-            .expect("generate");
 
         // Tamper with the ciphertext (flip a bit in the encrypted portion)
         let mut tampered = data_key.ciphertext.clone();
@@ -623,7 +566,7 @@ mod tests {
         }
 
         let request = DecryptRequest::new(tampered);
-        let result = KmsClient::decrypt(&backend, &request, None).await;
+        let result = backend.decrypt_envelope(&request);
         assert!(result.is_err());
     }
 
@@ -632,7 +575,14 @@ mod tests {
         let (backend, key_id, _key) = create_test_backend().await;
 
         // Creating the pre-configured key should return KeyAlreadyExists
-        let result = KmsClient::create_key(&backend, &key_id, "AES_256", None).await;
+        let result = KmsBackendTrait::create_key(
+            &backend,
+            CreateKeyRequest {
+                key_name: Some(key_id.clone()),
+                ..Default::default()
+            },
+        )
+        .await;
         assert!(result.is_err());
         assert!(result.expect_err("should be Err").to_string().contains("already exists"));
     }
@@ -642,7 +592,14 @@ mod tests {
         let (backend, _key_id, _key) = create_test_backend().await;
 
         // Creating any other key should return invalid operation (read-only)
-        let result = KmsClient::create_key(&backend, "other-key", "AES_256", None).await;
+        let result = KmsBackendTrait::create_key(
+            &backend,
+            CreateKeyRequest {
+                key_name: Some("other-key".to_string()),
+                ..Default::default()
+            },
+        )
+        .await;
         assert!(result.is_err());
         let err_msg = result.expect_err("should be Err").to_string();
         assert!(err_msg.contains("read-only") || err_msg.contains("cannot create"));
@@ -652,15 +609,13 @@ mod tests {
     async fn test_describe_key() {
         let (backend, key_id, _key) = create_test_backend().await;
 
-        let key_info = KmsClient::describe_key(&backend, &key_id, None)
+        let key_info = backend.configured_key_info(&key_id).expect("describe_key should succeed");
-            .await
-            .expect("describe_key should succeed");
         assert_eq!(key_info.key_id, key_id);
         assert_eq!(key_info.status, KeyStatus::Active);
         assert_eq!(key_info.algorithm, "AES_256");
 
         // Wrong key ID
-        let result = KmsClient::describe_key(&backend, "nonexistent", None).await;
+        let result = backend.configured_key_info("nonexistent");
         assert!(result.is_err());
     }
 
@@ -668,8 +623,8 @@ mod tests {
     async fn test_list_keys() {
         let (backend, key_id, _key) = create_test_backend().await;
 
-        let response = KmsClient::list_keys(&backend, &ListKeysRequest::default(), None)
+        let response = backend
-            .await
+            .list_configured_key(&ListKeysRequest::default())
             .expect("list_keys should succeed");
         assert_eq!(response.keys.len(), 1);
         assert_eq!(response.keys[0].key_id, key_id);
@@ -677,61 +632,45 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_disable_key_returns_error() {
+    async fn lifecycle_mutations_are_unsupported_at_the_product_surface() {
         let (backend, key_id, _key) = create_test_backend().await;
 
-        let result = KmsClient::disable_key(&backend, &key_id, None).await;
+        // The static backend advertises no enable/disable or rotation
-        assert!(result.is_err());
+        // capability, so the shared KmsBackend defaults reject all three.
-        assert!(result.expect_err("should be Err").to_string().contains("read-only"));
+        for result in [
+            KmsBackendTrait::enable_key(&backend, &key_id).await,
+            KmsBackendTrait::disable_key(&backend, &key_id).await,
+            KmsBackendTrait::rotate_key(&backend, &key_id).await,
+        ] {
+            let error = result.expect_err("static lifecycle mutations must be rejected");
+            assert!(matches!(error, KmsError::UnsupportedCapability { .. }), "got {error:?}");
         }
-
-    #[tokio::test]
-    async fn test_enable_key_is_noop() {
-        let (backend, key_id, _key) = create_test_backend().await;
-
-        // Enable should succeed (no-op for static KMS)
-        KmsClient::enable_key(&backend, &key_id, None)
-            .await
-            .expect("enable_key should be no-op");
-
-        // Wrong key should still fail
-        let result = KmsClient::enable_key(&backend, "wrong", None).await;
-        assert!(result.is_err());
     }
 
     #[tokio::test]
     async fn test_delete_key_returns_error() {
         let (backend, key_id, _key) = create_test_backend().await;
 
-        let result = KmsClient::schedule_key_deletion(&backend, &key_id, 7, None).await;
+        let result = KmsBackendTrait::delete_key(
+            &backend,
+            DeleteKeyRequest {
+                key_id: key_id.clone(),
+                pending_window_in_days: Some(7),
+                force_immediate: None,
+            },
+        )
+        .await;
         assert!(result.is_err());
         assert!(result.expect_err("should be Err").to_string().contains("read-only"));
     }
 
-    #[tokio::test]
-    async fn test_rotate_key_returns_error() {
-        let (backend, key_id, _key) = create_test_backend().await;
-
-        let result = KmsClient::rotate_key(&backend, &key_id, None).await;
-        assert!(result.is_err());
-    }
-
     #[tokio::test]
     async fn test_health_check() {
         let (backend, _key_id, _key) = create_test_backend().await;
 
-        KmsClient::health_check(&backend).await.expect("health_check should succeed");
+        KmsBackendTrait::health_check(&backend)
-    }
+            .await
-
+            .expect("health_check should succeed");
-    #[tokio::test]
-    async fn test_backend_info() {
-        let (backend, key_id, _key) = create_test_backend().await;
-
-        let info = KmsClient::backend_info(&backend);
-        assert_eq!(info.backend_type, "static");
-        assert_eq!(info.endpoint, "local");
-        assert!(info.healthy);
-        assert_eq!(info.metadata.get("key_id"), Some(&key_id));
     }
 
     #[tokio::test]
@@ -740,17 +679,13 @@ mod tests {
 
         let plaintext = b"Hello, static KMS world!";
         let enc_request = EncryptRequest::new(key_id.clone(), plaintext.to_vec());
-        let enc_response = KmsClient::encrypt(&backend, &enc_request, None)
+        let enc_response = backend.encrypt_to_envelope(&enc_request).expect("encrypt should succeed");
-            .await
-            .expect("encrypt should succeed");
 
         assert_eq!(enc_response.key_id, key_id);
         assert!(!enc_response.ciphertext.is_empty());
 
         let dec_request = DecryptRequest::new(enc_response.ciphertext);
-        let decrypted = KmsClient::decrypt(&backend, &dec_request, None)
+        let decrypted = backend.decrypt_envelope(&dec_request).expect("decrypt should succeed");
-            .await
-            .expect("decrypt should succeed");
 
         assert_eq!(decrypted, plaintext);
     }

crates/kms/src/backends/vault.rs

@@ -19,8 +19,7 @@ use crate::backends::vault_credentials::{
     token_source_for,
 };
 use crate::backends::{
-    BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits,
+    BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_state_permits, ensure_key_status_permits,
-    ensure_key_status_permits,
 };
 use crate::config::{KmsConfig, VaultConfig};
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -42,7 +41,6 @@ use vaultrs::{api::kv2::requests::SetSecretRequestOptions, error::ClientError, k
 /// Vault KMS client implementation
 pub struct VaultKmsClient {
     credentials: Arc<VaultCredentialProvider>,
-    config: VaultConfig,
     /// Mount path for the KV engine (typically "kv" or "secret")
     kv_mount: String,
     /// Path prefix for storing keys
@@ -200,7 +198,6 @@ impl VaultKmsClient {
             credentials,
             kv_mount: config.kv_mount.clone(),
             key_path_prefix: config.key_path_prefix.clone(),
-            config,
             dek_crypto: AesDekCrypto::new(),
             retry: RetryPolicy::from_config(kms_config),
             cancel: CancellationToken::new(),
@@ -254,13 +251,6 @@ impl VaultKmsClient {
         Ok(general_purpose::STANDARD.encode(key_material))
     }
 
-    /// Decode key material from KV2 storage (plain Base64, see `encrypt_key_material`).
-    async fn decrypt_key_material(&self, encrypted_material: &str) -> Result<Vec<u8>> {
-        general_purpose::STANDARD
-            .decode(encrypted_material)
-            .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))
-    }
-
     /// Read the immutable material record of one key version.
     ///
     /// A missing record fails closed with [`KmsError::KeyVersionNotFound`]; falling
@@ -589,9 +579,12 @@ impl VaultKmsClient {
     }
 }
 
-#[async_trait]
+impl VaultKmsClient {
-impl KmsClient for VaultKmsClient {
+    pub(crate) async fn generate_data_key(
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
+        &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?;
@@ -632,20 +625,32 @@ impl KmsClient for VaultKmsClient {
         Ok(data_key)
     }
 
-    async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
+    pub(crate) async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
         debug!("Encrypting data with key: {}", request.key_id);
 
-        // Get the master key and verify its state allows encryption
+        // Single read of the key record: the material we wrap with and the
+        // version stamped into the envelope must come from the same snapshot
+        // (see generate_data_key).
         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?;
+        let key_material = decode_stored_key_material(&request.key_id, &key_data.encrypted_key_material)
+            .inspect_err(|error| warn!(key_id = %request.key_id, %error, "Vault KMS key material failed validation"))?;
+        let (encrypted_key, nonce) = self.dek_crypto.encrypt(&key_material, &request.plaintext).await?;
 
-        // For simplicity, we'll use a basic encryption approach
+        // Wrap the ciphertext in the same authenticated envelope that
-        // In practice, you'd use proper AEAD encryption
+        // generate_data_key emits, so decrypt() round-trips it and resolves
-        let mut ciphertext = request.plaintext.clone();
+        // the wrapping master key version after rotations.
-        for (i, byte) in ciphertext.iter_mut().enumerate() {
+        let envelope = DataKeyEnvelope {
-            *byte ^= key_material[i % key_material.len()];
+            key_id: uuid::Uuid::new_v4().to_string(),
-        }
+            master_key_id: request.key_id.clone(),
+            key_spec: "AES_256".to_string(),
+            encrypted_key,
+            nonce,
+            encryption_context: request.encryption_context.clone(),
+            created_at: Zoned::now(),
+            master_key_version: Some(key_data.version),
+        };
+        let ciphertext = serde_json::to_vec(&envelope)?;
 
         Ok(EncryptResponse {
             ciphertext,
@@ -655,7 +660,7 @@ impl KmsClient for VaultKmsClient {
         })
     }
 
-    async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
+    pub(crate) async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
         debug!("Decrypting data");
 
         // Parse the data key envelope from ciphertext
@@ -697,7 +702,12 @@ impl KmsClient for VaultKmsClient {
         Ok(plaintext)
     }
 
-    async fn create_key(&self, key_id: &str, algorithm: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    pub(crate) async fn create_key(
+        &self,
+        key_id: &str,
+        algorithm: &str,
+        _context: Option<&OperationContext>,
+    ) -> Result<MasterKeyInfo> {
         debug!("Creating master key: {} with algorithm: {}", key_id, algorithm);
 
         // Existence pre-check with read-confirm recovery: a create whose
@@ -779,7 +789,7 @@ impl KmsClient for VaultKmsClient {
         Ok(master_key)
     }
 
-    async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
+    pub(crate) async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
         debug!("Describing key: {}", key_id);
 
         let key_data = self.get_key_data(key_id).await?;
@@ -799,7 +809,11 @@ impl KmsClient for VaultKmsClient {
         })
     }
 
-    async fn list_keys(&self, request: &ListKeysRequest, _context: Option<&OperationContext>) -> Result<ListKeysResponse> {
+    pub(crate) async fn list_keys(
+        &self,
+        request: &ListKeysRequest,
+        _context: Option<&OperationContext>,
+    ) -> Result<ListKeysResponse> {
         debug!("Listing keys with limit: {:?}", request.limit);
 
         let all_keys = self.list_vault_keys().await?;
@@ -836,7 +850,7 @@ impl KmsClient for VaultKmsClient {
         })
     }
 
-    async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         debug!("Enabling key: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
@@ -848,7 +862,7 @@ impl KmsClient for VaultKmsClient {
         Ok(())
     }
 
-    async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         debug!("Disabling key: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
@@ -860,39 +874,6 @@ impl KmsClient for VaultKmsClient {
         Ok(())
     }
 
-    async fn schedule_key_deletion(
-        &self,
-        key_id: &str,
-        pending_window_days: u32,
-        _context: Option<&OperationContext>,
-    ) -> Result<()> {
-        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;
-        key_data.deletion_date = Some(Zoned::now() + Duration::from_secs(pending_window_days as u64 * 86400));
-        self.store_key_data(key_id, &key_data).await?;
-
-        debug!(key_id, "Vault KMS key deletion scheduled");
-        Ok(())
-    }
-
-    async fn cancel_key_deletion(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        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;
-        key_data.deletion_date = None;
-        self.store_key_data(key_id, &key_data).await?;
-
-        debug!(key_id, "Vault KMS key deletion canceled");
-        Ok(())
-    }
-
     /// Rotate the master key while keeping every historical version decryptable.
     ///
     /// Commit protocol (all writes check-and-set, in this order):
@@ -908,10 +889,11 @@ impl KmsClient for VaultKmsClient {
     /// or interrupted rotation never exposes half-committed material. Concurrent
     /// rotations are serialized by the check-and-set writes: at most one caller
     /// commits each version and the losers fail without side effects on current.
-    async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    pub(crate) async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
         debug!("Rotating master key: {}", key_id);
 
         let (mut cas, mut key_data) = self.get_key_data_versioned(key_id).await?;
+        ensure_key_status_permits(key_id, &key_data.status, StateGatedOperation::Rotate)?;
 
         // The material about to be frozen must be decodable: freezing poisoned
         // material would give legacy envelopes a permanently broken baseline. This
@@ -992,7 +974,7 @@ impl KmsClient for VaultKmsClient {
         })
     }
 
-    async fn health_check(&self) -> Result<()> {
+    pub(crate) async fn health_check(&self) -> Result<()> {
         debug!("Performing Vault health check");
 
         // Use list_vault_keys but handle the case where no keys exist (which is normal)
@@ -1014,15 +996,6 @@ impl KmsClient for VaultKmsClient {
             }
         }
     }
-
-    fn backend_info(&self) -> BackendInfo {
-        BackendInfo::new("vault-kv2".to_string(), "0.1.0".to_string(), self.config.address.clone(), true)
-            .with_metadata("kv_mount".to_string(), self.kv_mount.clone())
-            .with_metadata("key_prefix".to_string(), self.key_path_prefix.clone())
-            // Master key material is protected only by Vault ACLs and KV2 at-rest
-            // encryption; there is no additional cryptographic wrapping.
-            .with_metadata("at_rest_protection".to_string(), "vault-kv2-acl".to_string())
-    }
 }
 
 /// VaultKmsBackend wraps VaultKmsClient and implements the KmsBackend trait
@@ -1031,12 +1004,6 @@ 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()?;
@@ -1296,17 +1263,32 @@ impl KmsBackend for VaultKmsBackend {
         })
     }
 
+    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 rotate_key(&self, key_id: &str) -> Result<()> {
+        self.client.rotate_key(key_id, None).await.map(|_| ())
+    }
+
     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
+        // Rotation freezes the outgoing material as an immutable version
-        // replacing key material in place, and no historical versions are
+        // record before switching the current pointer, and envelopes resolve
-        // retained, so versioning is unsupported as well.
+        // their wrapping version on decrypt, so every historical version
+        // stays decryptable after a rotation.
         BackendCapabilities::minimal()
+            .with_rotate(true)
             .with_enable_disable(true)
             .with_schedule_deletion(true)
+            .with_versioning(true)
             .with_physical_delete(true)
     }
 
@@ -1720,19 +1702,6 @@ mod tests {
         assert!(!is_cas_conflict(&not_found));
     }
 
-    #[tokio::test]
-    async fn test_vault_kv2_backend_info_reports_at_rest_protection() {
-        let client = VaultKmsClient::new(integration_vault_config(), &KmsConfig::default())
-            .await
-            .expect("client");
-
-        let info = client.backend_info();
-        assert_eq!(info.backend_type, "vault-kv2");
-        assert_eq!(info.metadata.get("at_rest_protection").map(String::as_str), Some("vault-kv2-acl"));
-        // The KV2 backend must not present itself as Transit-backed.
-        assert!(!format!("{info:?}").contains("Transit"));
-    }
-
     fn integration_generate_request(key_id: &str) -> GenerateKeyRequest {
         GenerateKeyRequest {
             master_key_id: key_id.to_string(),
@@ -2081,4 +2050,150 @@ mod tests {
         let legacy: VaultKeyData = serde_json::from_value(value).expect("legacy record must deserialize");
         assert!(legacy.deletion_date.is_none());
     }
+
+    /// KV2 write acknowledgement (`SecretVersionMetadata`) for `kv2::set`.
+    fn kv2_write_ack() -> serde_json::Value {
+        serde_json::json!({
+            "created_time": "2026-01-01T00:00:00Z",
+            "custom_metadata": null,
+            "deletion_time": "",
+            "destroyed": false,
+            "version": 2,
+        })
+    }
+
+    #[tokio::test]
+    async fn wired_kv2_encrypt_round_trips_through_decrypt() {
+        // One key-record read for the encrypt, one for the decrypt.
+        let (_vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(kv2_read_data(&healthy_key_data())),
+            ScriptedResponse::ok(kv2_read_data(&healthy_key_data())),
+        ])
+        .await;
+        let context = HashMap::from([("bucket".to_string(), "kv2".to_string())]);
+
+        let encrypted = client
+            .encrypt(
+                &EncryptRequest {
+                    key_id: "wired-key".to_string(),
+                    plaintext: b"kv2-direct-encrypt".to_vec(),
+                    encryption_context: context.clone(),
+                    grant_tokens: Vec::new(),
+                },
+                None,
+            )
+            .await
+            .expect("encrypt must produce an envelope");
+
+        // The ciphertext is a real KMS envelope wrapping AEAD output that
+        // decrypt() can open, not an XOR of the plaintext with the master key
+        // material.
+        let envelope: DataKeyEnvelope = serde_json::from_slice(&encrypted.ciphertext).expect("envelope must parse");
+        assert_eq!(envelope.master_key_id, "wired-key");
+        assert_eq!(envelope.master_key_version, Some(1));
+
+        let decrypted = client
+            .decrypt(
+                &DecryptRequest {
+                    ciphertext: encrypted.ciphertext.clone(),
+                    encryption_context: context,
+                    grant_tokens: Vec::new(),
+                },
+                None,
+            )
+            .await
+            .expect("decrypt must round-trip the envelope");
+        assert_eq!(decrypted, b"kv2-direct-encrypt".to_vec());
+
+        // A different object context must not decrypt (checked before any
+        // Vault read, so no scripted response is consumed).
+        let error = client
+            .decrypt(
+                &DecryptRequest {
+                    ciphertext: encrypted.ciphertext,
+                    encryption_context: HashMap::from([("bucket".to_string(), "other".to_string())]),
+                    grant_tokens: Vec::new(),
+                },
+                None,
+            )
+            .await
+            .expect_err("a different context must not decrypt");
+        assert!(matches!(error, KmsError::ContextMismatch { .. }), "got {error:?}");
+    }
+
+    /// KV2 secret-metadata read payload (`kv2::read_metadata`) pinning the
+    /// current secret version used as the rotation check-and-set base.
+    fn kv2_metadata_read_data(current_version: u64) -> serde_json::Value {
+        serde_json::json!({
+            "cas_required": false,
+            "created_time": "2026-01-01T00:00:00Z",
+            "current_version": current_version,
+            "delete_version_after": "0s",
+            "max_versions": 0,
+            "oldest_version": 0,
+            "updated_time": "2026-01-01T00:00:00Z",
+            "custom_metadata": null,
+            "versions": {},
+        })
+    }
+
+    #[tokio::test]
+    async fn wired_kv2_rotate_rejected_while_disabled() {
+        let mut key_data = healthy_key_data();
+        key_data.status = KeyStatus::Disabled;
+        let (vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(kv2_metadata_read_data(1)),
+            ScriptedResponse::ok(kv2_read_data(&key_data)),
+        ])
+        .await;
+
+        let error = client
+            .rotate_key("wired-key", None)
+            .await
+            .expect_err("rotation of a disabled key must be rejected");
+        assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}");
+
+        let requests = vault.requests();
+        assert_eq!(
+            requests.len(),
+            2,
+            "the state gate must reject after the versioned read, before any write: {requests:?}"
+        );
+        assert!(requests.iter().all(|line| line.starts_with("GET ")), "{requests:?}");
+    }
+
+    #[tokio::test]
+    async fn wired_backend_lifecycle_overrides_reach_the_client() {
+        let mut disabled = healthy_key_data();
+        disabled.status = KeyStatus::Disabled;
+        let vault = ScriptedVault::serve(vec![
+            // disable: read the Active record, persist it Disabled.
+            ScriptedResponse::ok(kv2_read_data(&healthy_key_data())),
+            ScriptedResponse::ok(kv2_write_ack()),
+            // enable: read the Disabled record, persist it Active.
+            ScriptedResponse::ok(kv2_read_data(&disabled)),
+            ScriptedResponse::ok(kv2_write_ack()),
+        ])
+        .await;
+        let config = KmsConfig::vault(
+            url::Url::parse(&vault.address).expect("scripted vault address should parse"),
+            "scripted-token".to_string(),
+        )
+        .with_insecure_development_defaults();
+        let backend = VaultKmsBackend::new(config).await.expect("vault kv2 backend should build");
+
+        backend
+            .disable_key("wired-key")
+            .await
+            .expect("KmsBackend::disable_key must persist through the client");
+        backend
+            .enable_key("wired-key")
+            .await
+            .expect("KmsBackend::enable_key must persist through the client");
+
+        let requests = vault.requests();
+        assert_eq!(requests.len(), 4, "each transition is one read plus one write: {requests:?}");
+        assert!(requests[0].starts_with("GET ") && requests[2].starts_with("GET "), "{requests:?}");
+        assert!(requests[1].starts_with("POST ") && requests[3].starts_with("POST "), "{requests:?}");
+    }
 }

crates/kms/src/backends/vault_transit.rs

@@ -18,9 +18,7 @@ use crate::backends::vault_credentials::{
     CredentialTaskHandle, VaultClientHandle, VaultConnectionSettings, VaultCredentialPolicy, VaultCredentialProvider,
     token_source_for,
 };
-use crate::backends::{
+use crate::backends::{BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_state_permits};
-    BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits,
-};
 use crate::config::{KmsConfig, VaultTransitConfig};
 use crate::encryption::{DataKeyEnvelope, generate_key_material};
 use crate::error::{KmsError, Result};
@@ -506,9 +504,12 @@ impl VaultTransitKmsClient {
     }
 }
 
-#[async_trait]
+impl VaultTransitKmsClient {
-impl KmsClient for VaultTransitKmsClient {
+    pub(crate) async fn generate_data_key(
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
+        &self,
+        request: &GenerateKeyRequest,
+        _context: Option<&OperationContext>,
+    ) -> Result<DataKeyInfo> {
         self.ensure_key_state_allows(&request.master_key_id, StateGatedOperation::GenerateDataKey)
             .await?;
 
@@ -540,7 +541,7 @@ impl KmsClient for VaultTransitKmsClient {
         ))
     }
 
-    async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
+    pub(crate) async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
         let metadata = self
             .ensure_key_state_allows(&request.key_id, StateGatedOperation::Encrypt)
             .await?;
@@ -556,7 +557,7 @@ impl KmsClient for VaultTransitKmsClient {
         })
     }
 
-    async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
+    pub(crate) async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
         let envelope: DataKeyEnvelope = serde_json::from_slice(&request.ciphertext)
             .map_err(|e| KmsError::cryptographic_error("parse", format!("Failed to parse data key envelope: {e}")))?;
 
@@ -578,7 +579,14 @@ impl KmsClient for VaultTransitKmsClient {
             .await
     }
 
-    async fn create_key(&self, key_id: &str, algorithm: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn create_key(
+        &self,
+        key_id: &str,
+        algorithm: &str,
+        _context: Option<&OperationContext>,
+    ) -> Result<MasterKeyInfo> {
         if algorithm != "AES_256" {
             return Err(KmsError::unsupported_algorithm(algorithm));
         }
@@ -645,11 +653,17 @@ impl KmsClient for VaultTransitKmsClient {
         })
     }
 
-    async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
         self.key_info(key_id).await
     }
 
-    async fn list_keys(&self, request: &ListKeysRequest, _context: Option<&OperationContext>) -> Result<ListKeysResponse> {
+    pub(crate) async fn list_keys(
+        &self,
+        request: &ListKeysRequest,
+        _context: Option<&OperationContext>,
+    ) -> Result<ListKeysResponse> {
         let all_keys = self
             .run("vault_transit_list_keys", OpClass::ReadIdempotent, move || async move {
                 let vault = self.vault().map_err(AttemptError::fatal)?;
@@ -692,7 +706,7 @@ impl KmsClient for VaultTransitKmsClient {
         })
     }
 
-    async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         // 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?;
@@ -701,13 +715,15 @@ impl KmsClient for VaultTransitKmsClient {
         self.store_key_metadata(key_id, &metadata).await
     }
 
-    async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         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
     }
 
-    async fn schedule_key_deletion(
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn schedule_key_deletion(
         &self,
         key_id: &str,
         pending_window_days: u32,
@@ -721,17 +737,7 @@ impl KmsClient for VaultTransitKmsClient {
         self.store_key_metadata(key_id, &metadata).await
     }
 
-    async fn cancel_key_deletion(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
-        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?;
 
         // Single attempt, never retried: replaying a rotate whose response was
@@ -768,7 +774,7 @@ impl KmsClient for VaultTransitKmsClient {
         })
     }
 
-    async fn health_check(&self) -> Result<()> {
+    pub(crate) async fn health_check(&self) -> Result<()> {
         self.run("vault_transit_health_check", OpClass::ReadIdempotent, move || async move {
             let vault = self.vault().map_err(AttemptError::fatal)?;
             key::list(&vault.client, &self.config.mount_path)
@@ -780,11 +786,6 @@ impl KmsClient for VaultTransitKmsClient {
         })
         .await
     }
-
-    fn backend_info(&self) -> BackendInfo {
-        BackendInfo::new("vault-transit".to_string(), "0.1.0".to_string(), self.config.address.clone(), true)
-            .with_metadata("mount_path".to_string(), self.config.mount_path.clone())
-    }
 }
 
 pub struct VaultTransitKmsBackend {
@@ -792,14 +793,6 @@ 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()?;
 
@@ -1000,6 +993,18 @@ impl KmsBackend for VaultTransitKmsBackend {
         })
     }
 
+    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 rotate_key(&self, key_id: &str) -> Result<()> {
+        self.client.rotate_key(key_id, None).await.map(|_| ())
+    }
+
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
@@ -1437,4 +1442,58 @@ mod tests {
         assert_eq!(metadata.key_state, KeyState::Enabled);
         assert!(metadata.deletion_date.is_none());
     }
+
+    /// KV2 write acknowledgement (`SecretVersionMetadata`) for `kv2::set`.
+    fn kv2_write_ack() -> serde_json::Value {
+        serde_json::json!({
+            "created_time": "2026-01-01T00:00:00Z",
+            "custom_metadata": null,
+            "deletion_time": "",
+            "destroyed": false,
+            "version": 2,
+        })
+    }
+
+    #[tokio::test]
+    async fn wired_backend_lifecycle_overrides_reach_the_client() {
+        let metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default());
+        let vault = ScriptedVault::serve(vec![
+            // disable: metadata cache miss reads KV, then persists Disabled.
+            ScriptedResponse::ok(metadata_read_data(&metadata)),
+            ScriptedResponse::ok(kv2_write_ack()),
+            // enable: the state gate hits the metadata cache, so only the
+            // persisting write goes out.
+            ScriptedResponse::ok(kv2_write_ack()),
+            // rotate: the gate hits the cache again; the single rotate
+            // attempt fails and must not be retried.
+            ScriptedResponse::error(503, "standby"),
+        ])
+        .await;
+        let config = KmsConfig::vault_transit(
+            url::Url::parse(&vault.address).expect("scripted vault address should parse"),
+            "scripted-token".to_string(),
+        )
+        .with_insecure_development_defaults();
+        let backend = VaultTransitKmsBackend::new(config)
+            .await
+            .expect("vault transit backend should build");
+
+        backend
+            .disable_key("wired-key")
+            .await
+            .expect("KmsBackend::disable_key must persist through the client");
+        backend
+            .enable_key("wired-key")
+            .await
+            .expect("KmsBackend::enable_key must persist through the client");
+        let error = backend
+            .rotate_key("wired-key")
+            .await
+            .expect_err("the scripted 503 must fail the rotation");
+        assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
+
+        let requests = vault.requests();
+        assert_eq!(requests.len(), 4, "gated reads, two writes and one rotate attempt: {requests:?}");
+        assert_eq!(requests[3], "POST /v1/transit/keys/wired-key/rotate", "{requests:?}");
+    }
 }

crates/kms/src/deletion_worker.rs

@@ -182,7 +182,6 @@ impl DeletionWorker {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::backends::KmsClient as _;
     use crate::backends::local::LocalKmsBackend;
     use crate::config::KmsConfig;
     use crate::error::KmsError;