feat(kms): export local backend key material as sealed backup bundles

Adds the producer side of the Local backup series on top of the #5483
contract: a directory-wide export fence gives the snapshot a single
consistent generation, every artifact is AEAD-wrapped under a
caller-supplied backup KEK that is separate from the business trust
hierarchy, and the sealed manifest with completeness marker is written
last so an interrupted export can never be mistaken for a restorable
bundle. Restore and the admin API land in follow-up changes.

crates/kms/Cargo.toml

@@ -75,7 +75,8 @@ anyhow = { workspace = true }
 insta = { workspace = true, features = ["yaml", "json"] }
 tempfile = { workspace = true }
 temp-env = { workspace = true }
-tokio = { workspace = true, features = ["test-util"] }
+# "net" backs the scripted loopback Vault used by the policy wiring tests.
+tokio = { workspace = true, features = ["net", "test-util"] }
 
 [features]
 default = []

crates/kms/src/backends/local.rs

@@ -399,6 +399,20 @@ pub struct LocalKmsClient {
     /// Per-key write locks serializing read-modify-write updates within this
     /// process (see [`Self::lock_key_for_write`]).
     key_write_locks: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
+    /// Directory-wide writer fence for backup export (see
+    /// [`Self::acquire_export_fence`]). Writers hold the read side; an export
+    /// snapshot holds the write side so it observes a single-generation view.
+    export_fence: Arc<tokio::sync::RwLock<()>>,
+}
+
+/// Guard pairing the export-fence read lock with a per-key write mutex.
+///
+/// Dropping it releases both, so every existing `lock_key_for_write` call
+/// site participates in the export fence without changes.
+#[must_use]
+struct KeyWriteGuard {
+    _fence: tokio::sync::OwnedRwLockReadGuard<()>,
+    _key: tokio::sync::OwnedMutexGuard<()>,
 }
 
 // pub(crate) so the backup contract tests can anchor the manifest's
@@ -465,6 +479,7 @@ impl LocalKmsClient {
             legacy_master_cipher,
             dek_crypto: AesDekCrypto::new(),
             key_write_locks: Mutex::new(HashMap::new()),
+            export_fence: Arc::new(tokio::sync::RwLock::new(())),
         };
         client.validate_existing_keys().await?;
         Ok(client)
@@ -507,6 +522,7 @@ impl LocalKmsClient {
             legacy_master_cipher,
             dek_crypto: AesDekCrypto::new(),
             key_write_locks: Mutex::new(HashMap::new()),
+            export_fence: Arc::new(tokio::sync::RwLock::new(())),
         })
     }
 
@@ -517,12 +533,40 @@ impl LocalKmsClient {
     /// delete with a rewrite. Cross-process writers sharing a key directory
     /// remain unsupported. Entries live for the client's lifetime; the table
     /// is bounded by the number of distinct key ids this process touches.
-    async fn lock_key_for_write(&self, key_id: &str) -> tokio::sync::OwnedMutexGuard<()> {
+    async fn lock_key_for_write(&self, key_id: &str) -> KeyWriteGuard {
+        // Fence first, per-key mutex second: the ordering is uniform across
+        // all writers, so an export waiting on the write side can never
+        // deadlock with a writer holding a key mutex.
+        let fence = Arc::clone(&self.export_fence).read_owned().await;
         let lock = {
             let mut locks = self.key_write_locks.lock().expect("Local KMS key write lock table poisoned");
             Arc::clone(locks.entry(key_id.to_string()).or_default())
         };
-        lock.lock_owned().await
+        KeyWriteGuard {
+            _fence: fence,
+            _key: lock.lock_owned().await,
+        }
+    }
+
+    /// Block every key-directory writer while a backup export collects its
+    /// snapshot, so all records belong to one generation.
+    ///
+    /// Mutating operations hold the read side (via [`Self::lock_key_for_write`]
+    /// or [`Self::save_new_master_key`]); the export holds the write side only
+    /// for the collection phase, never while encrypting or writing the bundle.
+    pub(crate) async fn acquire_export_fence(&self) -> tokio::sync::OwnedRwLockWriteGuard<()> {
+        Arc::clone(&self.export_fence).write_owned().await
+    }
+
+    /// Key directory root, exposed for the backup export module.
+    pub(crate) fn key_directory(&self) -> &Path {
+        &self.config.key_dir
+    }
+
+    /// Absolute path of the master-key KDF salt file, exposed for the backup
+    /// export module.
+    pub(crate) fn master_key_salt_file(&self) -> PathBuf {
+        Self::master_key_salt_path(&self.config)
     }
 
     /// Derive a 256-bit key from the master key string using a persistent Argon2id salt.
@@ -799,6 +843,11 @@ impl LocalKmsClient {
     }
 
     async fn save_new_master_key(&self, master_key: &MasterKeyInfo, key_material: &[u8]) -> Result<()> {
+        // Creates never take the per-key write lock (`NoClobber` publishing
+        // already linearizes them), so they join the export fence here. This
+        // must stay the only fence acquisition on the create path: the fence
+        // read lock is not reentrant while an export waits for the write side.
+        let _fence = Arc::clone(&self.export_fence).read_owned().await;
         let key_path = self.master_key_path(&master_key.key_id)?;
         let content = self.encode_master_key(master_key, key_material)?;
         let temp_path = key_path.with_extension(format!("tmp-{}", uuid::Uuid::new_v4()));
@@ -1533,6 +1582,14 @@ impl KmsBackend for LocalKmsBackend {
         })
     }
 
+    async fn enable_key(&self, key_id: &str) -> Result<()> {
+        self.client.enable_key(key_id, None).await
+    }
+
+    async fn disable_key(&self, key_id: &str) -> Result<()> {
+        self.client.disable_key(key_id, None).await
+    }
+
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }

crates/kms/src/backends/mod.rs

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

crates/kms/src/backends/scripted_vault.rs

@@ -0,0 +1,159 @@
+// 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.
+
+//! Minimal scripted HTTP responder standing in for a Vault server.
+//!
+//! Wiring tests need to observe how many Vault requests a code path performs
+//! (retries, read-confirm recovery) without a live Vault. The responder serves
+//! one canned response per incoming request in order, closes the connection
+//! after each response, and records the `METHOD /path` sequence for
+//! assertions. It intentionally implements just enough HTTP/1.1 for the
+//! `vaultrs` reqwest client: no keep-alive, no chunked bodies.
+
+use std::sync::{Arc, Mutex};
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
+
+/// One canned HTTP response.
+pub(crate) struct ScriptedResponse {
+    status: u16,
+    body: String,
+}
+
+impl ScriptedResponse {
+    /// A 200 response carrying `data` inside the standard Vault envelope.
+    pub(crate) fn ok(data: serde_json::Value) -> Self {
+        Self {
+            status: 200,
+            body: serde_json::json!({
+                "request_id": "scripted",
+                "lease_id": "",
+                "lease_duration": 0,
+                "renewable": false,
+                "data": data,
+            })
+            .to_string(),
+        }
+    }
+
+    /// An error response in Vault's `{"errors": [...]}` format.
+    pub(crate) fn error(status: u16, message: &str) -> Self {
+        Self {
+            status,
+            body: serde_json::json!({ "errors": [message] }).to_string(),
+        }
+    }
+}
+
+/// A scripted stand-in Vault listening on a loopback port.
+pub(crate) struct ScriptedVault {
+    /// Base address (`http://127.0.0.1:port`) to point a Vault client at.
+    pub(crate) address: String,
+    requests: Arc<Mutex<Vec<String>>>,
+}
+
+impl ScriptedVault {
+    /// Bind a loopback listener and serve `responses` one per request.
+    ///
+    /// Requests beyond the script get a 599 error so a test that under-scripts
+    /// fails loudly instead of hanging.
+    pub(crate) async fn serve(responses: Vec<ScriptedResponse>) -> Self {
+        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind scripted vault listener");
+        let address = format!("http://{}", listener.local_addr().expect("scripted vault local addr"));
+        let requests = Arc::new(Mutex::new(Vec::new()));
+        let recorded = Arc::clone(&requests);
+
+        tokio::spawn(async move {
+            let mut responses = responses.into_iter();
+            loop {
+                let Ok((mut stream, _)) = listener.accept().await else {
+                    return;
+                };
+                let Some(request_line) = read_request(&mut stream).await else {
+                    continue;
+                };
+                recorded
+                    .lock()
+                    .expect("scripted vault request log poisoned")
+                    .push(request_line);
+                let response = responses
+                    .next()
+                    .unwrap_or_else(|| ScriptedResponse::error(599, "scripted vault: script exhausted"));
+                let payload = format!(
+                    "HTTP/1.1 {} Scripted\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
+                    response.status,
+                    response.body.len(),
+                    response.body
+                );
+                let _ = stream.write_all(payload.as_bytes()).await;
+                let _ = stream.shutdown().await;
+            }
+        });
+
+        Self { address, requests }
+    }
+
+    /// The `METHOD /path` lines of every request served so far, in order.
+    pub(crate) fn requests(&self) -> Vec<String> {
+        self.requests.lock().expect("scripted vault request log poisoned").clone()
+    }
+}
+
+/// Read one HTTP/1.1 request (head plus content-length body) and return its
+/// `METHOD /path` line. Draining the body before responding keeps the client
+/// from seeing a connection reset while it is still writing.
+async fn read_request(stream: &mut TcpStream) -> Option<String> {
+    let mut buffer = Vec::new();
+    let mut chunk = [0u8; 4096];
+    let head_end = loop {
+        if let Some(position) = buffer.windows(4).position(|window| window == b"\r\n\r\n") {
+            break position + 4;
+        }
+        let read = stream.read(&mut chunk).await.ok()?;
+        if read == 0 {
+            return None;
+        }
+        buffer.extend_from_slice(&chunk[..read]);
+    };
+
+    let head = String::from_utf8_lossy(&buffer[..head_end]).into_owned();
+    let mut lines = head.lines();
+    let request_line = lines.next()?;
+    let mut parts = request_line.split_whitespace();
+    let method = parts.next()?;
+    let path = parts.next()?;
+    // rustify appends a lone "?" when an endpoint has no query parameters;
+    // strip it so assertions can use the plain path.
+    let path = path.strip_suffix('?').unwrap_or(path);
+
+    let content_length: usize = lines
+        .filter_map(|line| {
+            let (name, value) = line.split_once(':')?;
+            name.eq_ignore_ascii_case("content-length")
+                .then(|| value.trim().parse().ok())?
+        })
+        .next()
+        .unwrap_or(0);
+    let mut remaining = content_length.saturating_sub(buffer.len() - head_end);
+    while remaining > 0 {
+        let read = stream.read(&mut chunk).await.ok()?;
+        if read == 0 {
+            break;
+        }
+        remaining = remaining.saturating_sub(read);
+    }
+
+    Some(format!("{method} {path}"))
+}

crates/kms/src/backends/vault.rs

@@ -25,14 +25,17 @@ use crate::backends::{
 use crate::config::{KmsConfig, VaultConfig};
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
 use crate::error::{KmsError, Result};
+use crate::policy::{self, AttemptError, OpClass, RetryPolicy};
 use crate::types::*;
 use async_trait::async_trait;
 use base64::{Engine as _, engine::general_purpose};
 use jiff::Zoned;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
+use std::future::Future;
 use std::sync::Arc;
 use std::time::Duration;
+use tokio_util::sync::CancellationToken;
 use tracing::{debug, info, warn};
 use vaultrs::{api::kv2::requests::SetSecretRequestOptions, error::ClientError, kv2};
 
@@ -46,6 +49,13 @@ pub struct VaultKmsClient {
     key_path_prefix: String,
     /// DEK encryption implementation
     dek_crypto: AesDekCrypto,
+    /// Budgets wrapping every outbound Vault call (see `crate::policy`).
+    retry: RetryPolicy,
+    /// Cancellation point for the operation executor: aborts in-flight
+    /// attempts and backoff sleeps. Owned by the client and currently never
+    /// triggered — shutdown drops the whole client — but kept as the single
+    /// hook a future lifecycle owner can cancel through.
+    cancel: CancellationToken,
 }
 
 /// Key data stored in Vault
@@ -192,6 +202,8 @@ impl VaultKmsClient {
             key_path_prefix: config.key_path_prefix.clone(),
             config,
             dek_crypto: AesDekCrypto::new(),
+            retry: RetryPolicy::from_config(kms_config),
+            cancel: CancellationToken::new(),
         })
     }
 
@@ -204,6 +216,19 @@ impl VaultKmsClient {
         self.credentials.current()
     }
 
+    /// Run one Vault call under the operation policy.
+    ///
+    /// The closure performs a single classified attempt and takes a fresh
+    /// credential snapshot per attempt, so a retry after a credential rotation
+    /// uses the new token.
+    async fn run<T, F, Fut>(&self, operation: &'static str, class: OpClass, attempt: F) -> Result<T>
+    where
+        F: FnMut() -> Fut,
+        Fut: Future<Output = std::result::Result<T, AttemptError>>,
+    {
+        policy::execute(operation, class, &self.retry, &self.cancel, attempt).await
+    }
+
     /// Get the full path for a key in Vault
     fn key_path(&self, key_id: &str) -> String {
         format!("{}/{}", self.key_path_prefix, key_id)
@@ -244,14 +269,20 @@ impl VaultKmsClient {
     async fn get_key_version_record(&self, key_id: &str, version: u32) -> Result<VaultKeyVersionRecord> {
         let path = self.key_version_path(key_id, version);
 
-        let record: VaultKeyVersionRecord =
-            kv2::read(&self.vault()?.client, &self.kv_mount, &path)
-                .await
-                .map_err(|e| match e {
-                    ClientError::ResponseWrapError => KmsError::key_version_not_found(key_id, version),
-                    ClientError::APIError { code: 404, .. } => KmsError::key_version_not_found(key_id, version),
-                    _ => KmsError::backend_error(format!("Failed to read key version record from Vault: {e}")),
-                })?;
+        let path = path.as_str();
+        let record: VaultKeyVersionRecord = self
+            .run("vault_kv2_read_key_version", OpClass::ReadIdempotent, move || async move {
+                let vault = self.vault().map_err(AttemptError::fatal)?;
+                kv2::read(&vault.client, &self.kv_mount, path).await.map_err(|e| {
+                    AttemptError::from_vaultrs(e, |e| match e {
+                        ClientError::ResponseWrapError | ClientError::APIError { code: 404, .. } => {
+                            KmsError::key_version_not_found(key_id, version)
+                        }
+                        e => KmsError::backend_error(format!("Failed to read key version record from Vault: {e}")),
+                    })
+                })
+            })
+            .await?;
 
         if record.version != version {
             return Err(KmsError::material_corrupt(
@@ -284,26 +315,42 @@ impl VaultKmsClient {
     /// later write can be check-and-set against exactly this snapshot.
     async fn get_key_data_versioned(&self, key_id: &str) -> Result<(u32, VaultKeyData)> {
         let path = self.key_path(key_id);
+        let path = path.as_str();
 
-        let metadata = kv2::read_metadata(&self.vault()?.client, &self.kv_mount, &path)
-            .await
-            .map_err(|e| match e {
-                ClientError::ResponseWrapError => KmsError::key_not_found(key_id),
-                ClientError::APIError { code: 404, .. } => KmsError::key_not_found(key_id),
-                _ => KmsError::backend_error(format!("Failed to read key metadata from Vault: {e}")),
-            })?;
+        let metadata = self
+            .run("vault_kv2_read_key_metadata", OpClass::ReadIdempotent, move || async move {
+                let vault = self.vault().map_err(AttemptError::fatal)?;
+                kv2::read_metadata(&vault.client, &self.kv_mount, path).await.map_err(|e| {
+                    AttemptError::from_vaultrs(e, |e| match e {
+                        ClientError::ResponseWrapError | ClientError::APIError { code: 404, .. } => {
+                            KmsError::key_not_found(key_id)
+                        }
+                        e => KmsError::backend_error(format!("Failed to read key metadata from Vault: {e}")),
+                    })
+                })
+            })
+            .await?;
         let cas = u32::try_from(metadata.current_version)
             .map_err(|_| KmsError::backend_error(format!("KV2 secret version for key {key_id} exceeds u32")))?;
 
         // Read the exact secret version from the metadata to keep the (cas, data)
         // pair consistent even if another writer lands in between.
-        let key_data: VaultKeyData = kv2::read_version(&self.vault()?.client, &self.kv_mount, &path, metadata.current_version)
-            .await
-            .map_err(|e| match e {
-                ClientError::ResponseWrapError => KmsError::key_not_found(key_id),
-                ClientError::APIError { code: 404, .. } => KmsError::key_not_found(key_id),
-                _ => KmsError::backend_error(format!("Failed to read key from Vault: {e}")),
-            })?;
+        let secret_version = metadata.current_version;
+        let key_data: VaultKeyData = self
+            .run("vault_kv2_read_key_at_version", OpClass::ReadIdempotent, move || async move {
+                let vault = self.vault().map_err(AttemptError::fatal)?;
+                kv2::read_version(&vault.client, &self.kv_mount, path, secret_version)
+                    .await
+                    .map_err(|e| {
+                        AttemptError::from_vaultrs(e, |e| match e {
+                            ClientError::ResponseWrapError | ClientError::APIError { code: 404, .. } => {
+                                KmsError::key_not_found(key_id)
+                            }
+                            e => KmsError::backend_error(format!("Failed to read key from Vault: {e}")),
+                        })
+                    })
+            })
+            .await?;
 
         Ok((cas, key_data))
     }
@@ -315,19 +362,29 @@ impl VaultKmsClient {
     /// further check-and-set writes.
     async fn cas_store_key_data(&self, key_id: &str, key_data: &VaultKeyData, cas: u32) -> Result<u32> {
         let path = self.key_path(key_id);
+        let path = path.as_str();
 
-        let written =
-            kv2::set_with_options(&self.vault()?.client, &self.kv_mount, &path, key_data, SetSecretRequestOptions { cas })
-                .await
-                .map_err(|e| {
-                    if is_cas_conflict(&e) {
-                        KmsError::invalid_operation(format!(
-                            "Concurrent modification of key {key_id} detected, retry the rotation"
-                        ))
-                    } else {
-                        KmsError::backend_error(format!("Failed to store key in Vault: {e}"))
-                    }
-                })?;
+        // Single attempt: replaying a lost-response write would double-apply
+        // the mutation, and a CAS conflict is a normal concurrency signal that
+        // must reach the caller untouched.
+        let written = self
+            .run("vault_kv2_cas_write_key", OpClass::MutatingNonIdempotent, move || async move {
+                let vault = self.vault().map_err(AttemptError::fatal)?;
+                kv2::set_with_options(&vault.client, &self.kv_mount, path, key_data, SetSecretRequestOptions { cas })
+                    .await
+                    .map_err(|e| {
+                        AttemptError::from_vaultrs(e, |e| {
+                            if is_cas_conflict(&e) {
+                                KmsError::invalid_operation(format!(
+                                    "Concurrent modification of key {key_id} detected, retry the rotation"
+                                ))
+                            } else {
+                                KmsError::backend_error(format!("Failed to store key in Vault: {e}"))
+                            }
+                        })
+                    })
+            })
+            .await?;
 
         u32::try_from(written.version)
             .map_err(|_| KmsError::backend_error(format!("KV2 secret version for key {key_id} exceeds u32")))
@@ -340,23 +397,42 @@ impl VaultKmsClient {
     /// existing record is acceptable. The record is never overwritten.
     async fn try_create_key_version_record(&self, key_id: &str, record: &VaultKeyVersionRecord) -> Result<bool> {
         let path = self.key_version_path(key_id, record.version);
+        let path = path.as_str();
 
-        match kv2::set_with_options(&self.vault()?.client, &self.kv_mount, &path, record, SetSecretRequestOptions { cas: 0 })
-            .await
-        {
-            Ok(_) => Ok(true),
-            Err(e) if is_cas_conflict(&e) => Ok(false),
-            Err(e) => Err(KmsError::backend_error(format!("Failed to store key version record in Vault: {e}"))),
-        }
+        // Single attempt: the create-only CAS makes a duplicate replay fail
+        // with a conflict, which the caller resolves by reading the record
+        // back, so retrying here would only mask that recovery path.
+        self.run("vault_kv2_create_key_version", OpClass::MutatingNonIdempotent, move || async move {
+            let vault = self.vault().map_err(AttemptError::fatal)?;
+            match kv2::set_with_options(&vault.client, &self.kv_mount, path, record, SetSecretRequestOptions { cas: 0 }).await {
+                Ok(_) => Ok(true),
+                Err(e) if is_cas_conflict(&e) => Ok(false),
+                Err(e) => Err(AttemptError::from_vaultrs(e, |e| {
+                    KmsError::backend_error(format!("Failed to store key version record in Vault: {e}"))
+                })),
+            }
+        })
+        .await
     }
 
     /// Store key data in Vault
     async fn store_key_data(&self, key_id: &str, key_data: &VaultKeyData) -> Result<()> {
         let path = self.key_path(key_id);
+        let path = path.as_str();
 
-        kv2::set(&self.vault()?.client, &self.kv_mount, &path, key_data)
-            .await
-            .map_err(|e| KmsError::backend_error(format!("Failed to store key in Vault: {e}")))?;
+        // Single attempt: this is a whole-record overwrite without a CAS
+        // precondition, so a replay after a lost response could clobber a
+        // concurrent writer.
+        self.run("vault_kv2_write_key", OpClass::MutatingNonIdempotent, move || async move {
+            let vault = self.vault().map_err(AttemptError::fatal)?;
+            kv2::set(&vault.client, &self.kv_mount, path, key_data)
+                .await
+                .map(|_| ())
+                .map_err(|e| {
+                    AttemptError::from_vaultrs(e, |e| KmsError::backend_error(format!("Failed to store key in Vault: {e}")))
+                })
+        })
+        .await?;
 
         debug!("Stored key {} in Vault at path {}", key_id, path);
         Ok(())
@@ -404,14 +480,21 @@ impl VaultKmsClient {
     /// Retrieve key data from Vault
     async fn get_key_data(&self, key_id: &str) -> Result<VaultKeyData> {
         let path = self.key_path(key_id);
+        let path = path.as_str();
 
-        let secret: VaultKeyData = kv2::read(&self.vault()?.client, &self.kv_mount, &path)
-            .await
-            .map_err(|e| match e {
-                vaultrs::error::ClientError::ResponseWrapError => KmsError::key_not_found(key_id),
-                vaultrs::error::ClientError::APIError { code: 404, .. } => KmsError::key_not_found(key_id),
-                _ => KmsError::backend_error(format!("Failed to read key from Vault: {e}")),
-            })?;
+        let secret: VaultKeyData = self
+            .run("vault_kv2_read_key", OpClass::ReadIdempotent, move || async move {
+                let vault = self.vault().map_err(AttemptError::fatal)?;
+                kv2::read(&vault.client, &self.kv_mount, path).await.map_err(|e| {
+                    AttemptError::from_vaultrs(e, |e| match e {
+                        ClientError::ResponseWrapError | ClientError::APIError { code: 404, .. } => {
+                            KmsError::key_not_found(key_id)
+                        }
+                        e => KmsError::backend_error(format!("Failed to read key from Vault: {e}")),
+                    })
+                })
+            })
+            .await?;
 
         debug!("Retrieved key {} from Vault, tags: {:?}", key_id, secret.tags);
         Ok(secret)
@@ -419,56 +502,87 @@ impl VaultKmsClient {
 
     /// List all keys stored in Vault
     async fn list_vault_keys(&self) -> Result<Vec<String>> {
-        // List keys under the prefix
-        match kv2::list(&self.vault()?.client, &self.kv_mount, &self.key_path_prefix).await {
-            Ok(keys) => {
+        // List keys under the prefix; `None` means the prefix does not exist
+        // yet (no keys were ever created).
+        let keys = self
+            .run("vault_kv2_list_keys", OpClass::ReadIdempotent, move || async move {
+                let vault = self.vault().map_err(AttemptError::fatal)?;
+                match kv2::list(&vault.client, &self.kv_mount, &self.key_path_prefix).await {
+                    Ok(keys) => Ok(Some(keys)),
+                    Err(ClientError::ResponseWrapError) | Err(ClientError::APIError { code: 404, .. }) => Ok(None),
+                    Err(e) => Err(AttemptError::from_vaultrs(e, |e| {
+                        KmsError::backend_error(format!("Failed to list keys in Vault: {e}"))
+                    })),
+                }
+            })
+            .await?;
+
+        match keys {
+            Some(keys) => {
                 let keys = filter_key_directory_entries(keys);
                 debug!("Found {} keys in Vault", keys.len());
                 Ok(keys)
             }
-            Err(vaultrs::error::ClientError::ResponseWrapError) => {
-                // No keys exist yet
-                Ok(Vec::new())
-            }
-            Err(vaultrs::error::ClientError::APIError { code: 404, .. }) => {
-                // Path doesn't exist - no keys exist yet
+            None => {
                 debug!("Key path doesn't exist in Vault (404), returning empty list");
                 Ok(Vec::new())
             }
-            Err(e) => Err(KmsError::backend_error(format!("Failed to list keys in Vault: {e}"))),
         }
     }
 
     /// Physically delete a key from Vault storage
     async fn delete_key(&self, key_id: &str) -> Result<()> {
         let path = self.key_path(key_id);
+        let path = path.as_str();
 
         // Purge immutable version records first: if any purge fails, the top-level
         // record still exists and the deletion can be retried. The reverse order
         // would leave orphaned master key material in Vault after the key vanished.
         let versions_dir = self.key_versions_dir(key_id);
-        match kv2::list(&self.vault()?.client, &self.kv_mount, &versions_dir).await {
-            Ok(versions) => {
-                for version in versions {
-                    let version_path = format!("{versions_dir}/{version}");
-                    kv2::delete_metadata(&self.vault()?.client, &self.kv_mount, &version_path)
-                        .await
-                        .map_err(|e| KmsError::backend_error(format!("Failed to delete key version record from Vault: {e}")))?;
+        let versions_dir = versions_dir.as_str();
+        // `None` means no version records exist (the key was never rotated).
+        let versions = self
+            .run("vault_kv2_list_key_versions", OpClass::ReadIdempotent, move || async move {
+                let vault = self.vault().map_err(AttemptError::fatal)?;
+                match kv2::list(&vault.client, &self.kv_mount, versions_dir).await {
+                    Ok(versions) => Ok(Some(versions)),
+                    Err(ClientError::ResponseWrapError) | Err(ClientError::APIError { code: 404, .. }) => Ok(None),
+                    Err(e) => Err(AttemptError::from_vaultrs(e, |e| {
+                        KmsError::backend_error(format!("Failed to list key version records in Vault: {e}"))
+                    })),
                 }
-            }
-            // No version records exist (the key was never rotated).
-            Err(ClientError::ResponseWrapError) | Err(ClientError::APIError { code: 404, .. }) => {}
-            Err(e) => return Err(KmsError::backend_error(format!("Failed to list key version records in Vault: {e}"))),
+            })
+            .await?;
+        for version in versions.unwrap_or_default() {
+            let version_path = format!("{versions_dir}/{version}");
+            let version_path = version_path.as_str();
+            self.run("vault_kv2_delete_key_version", OpClass::MutatingNonIdempotent, move || async move {
+                let vault = self.vault().map_err(AttemptError::fatal)?;
+                match kv2::delete_metadata(&vault.client, &self.kv_mount, version_path).await {
+                    // A version record that is already gone is a completed
+                    // delete (e.g. this deletion is being re-run after a lost
+                    // response), not a failure.
+                    Ok(_) | Err(ClientError::ResponseWrapError) | Err(ClientError::APIError { code: 404, .. }) => Ok(()),
+                    Err(e) => Err(AttemptError::from_vaultrs(e, |e| {
+                        KmsError::backend_error(format!("Failed to delete key version record from Vault: {e}"))
+                    })),
+                }
+            })
+            .await?;
         }
 
         // For this specific key path, we can safely delete the metadata
         // since each key has its own unique path under the prefix
-        kv2::delete_metadata(&self.vault()?.client, &self.kv_mount, &path)
-            .await
-            .map_err(|e| match e {
-                vaultrs::error::ClientError::APIError { code: 404, .. } => KmsError::key_not_found(key_id),
-                _ => KmsError::backend_error(format!("Failed to delete key metadata from Vault: {e}")),
-            })?;
+        self.run("vault_kv2_delete_key", OpClass::MutatingNonIdempotent, move || async move {
+            let vault = self.vault().map_err(AttemptError::fatal)?;
+            kv2::delete_metadata(&vault.client, &self.kv_mount, path).await.map_err(|e| {
+                AttemptError::from_vaultrs(e, |e| match e {
+                    ClientError::APIError { code: 404, .. } => KmsError::key_not_found(key_id),
+                    e => KmsError::backend_error(format!("Failed to delete key metadata from Vault: {e}")),
+                })
+            })
+        })
+        .await?;
 
         debug!("Permanently deleted key {} metadata from Vault at path {}", key_id, path);
         Ok(())
@@ -586,9 +700,43 @@ impl KmsClient for VaultKmsClient {
     async fn create_key(&self, key_id: &str, algorithm: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
         debug!("Creating master key: {} with algorithm: {}", key_id, algorithm);
 
-        // Check if key already exists
-        if self.get_key_data(key_id).await.is_ok() {
-            return Err(KmsError::key_already_exists(key_id));
+        // Existence pre-check with read-confirm recovery: a create whose
+        // response was lost gets retried by callers, and used to be
+        // misreported as KeyAlreadyExists. If the stored key is exactly what
+        // this create would have produced (same algorithm, active, usable
+        // material), report the stored key as the create result. Anything
+        // else keeps failing: create never adopts a key it would not have
+        // produced. A failed pre-check read must fail the create rather than
+        // fall through to a blind overwrite of a possibly existing key.
+        match self.get_key_data(key_id).await {
+            Ok(existing) => {
+                return if existing.algorithm == algorithm
+                    && existing.status == KeyStatus::Active
+                    && decode_stored_key_material(key_id, &existing.encrypted_key_material).is_ok()
+                {
+                    info!(
+                        key_id,
+                        "Vault KMS create found an identical active key; treating it as a recovered create"
+                    );
+                    Ok(MasterKeyInfo {
+                        key_id: key_id.to_string(),
+                        version: existing.version,
+                        algorithm: existing.algorithm,
+                        usage: existing.usage,
+                        status: existing.status,
+                        description: existing.description,
+                        metadata: existing.metadata,
+                        created_at: existing.created_at,
+                        rotated_at: None,
+                        created_by: None,
+                        deletion_date: existing.deletion_date,
+                    })
+                } else {
+                    Err(KmsError::key_already_exists(key_id))
+                };
+            }
+            Err(KmsError::KeyNotFound { .. }) => {}
+            Err(error) => return Err(error),
         }
 
         // Generate key material
@@ -1203,8 +1351,183 @@ impl KmsBackend for VaultKmsBackend {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::backends::scripted_vault::{ScriptedResponse, ScriptedVault};
     use crate::config::{VaultAuthMethod, VaultConfig};
 
+    /// Vault + KMS config pair pointing at a scripted loopback Vault.
+    fn scripted_configs(address: &str) -> (VaultConfig, KmsConfig) {
+        let vault_config = VaultConfig {
+            address: address.to_string(),
+            auth_method: VaultAuthMethod::Token {
+                token: "scripted-token".to_string(),
+            },
+            kv_mount: "secret".to_string(),
+            key_path_prefix: "rustfs/kms/keys".to_string(),
+            mount_path: "transit".to_string(),
+            namespace: None,
+            tls: None,
+        };
+        let kms_config = KmsConfig {
+            timeout: Duration::from_secs(5),
+            retry_attempts: 3,
+            ..KmsConfig::default()
+        };
+        (vault_config, kms_config)
+    }
+
+    async fn scripted_client(responses: Vec<ScriptedResponse>) -> (ScriptedVault, VaultKmsClient) {
+        let vault = ScriptedVault::serve(responses).await;
+        let (vault_config, kms_config) = scripted_configs(&vault.address);
+        let client = VaultKmsClient::new(vault_config, &kms_config)
+            .await
+            .expect("scripted Vault client");
+        (vault, client)
+    }
+
+    fn healthy_key_data() -> VaultKeyData {
+        VaultKeyData {
+            algorithm: "AES_256".to_string(),
+            usage: KeyUsage::EncryptDecrypt,
+            created_at: Zoned::now(),
+            status: KeyStatus::Active,
+            version: 1,
+            description: None,
+            metadata: HashMap::new(),
+            tags: HashMap::new(),
+            deletion_date: None,
+            encrypted_key_material: general_purpose::STANDARD.encode([0x42u8; 32]),
+            baseline_version: None,
+        }
+    }
+
+    /// KV2 read payload (the `data` field of the Vault envelope) for a key record.
+    fn kv2_read_data(key_data: &VaultKeyData) -> serde_json::Value {
+        serde_json::json!({
+            "data": serde_json::to_value(key_data).expect("serialize key data"),
+            "metadata": {
+                "created_time": "2026-01-01T00:00:00Z",
+                "deletion_time": "",
+                "custom_metadata": null,
+                "destroyed": false,
+                "version": 1,
+            },
+        })
+    }
+
+    #[tokio::test]
+    async fn wired_read_retries_transient_status_then_succeeds() {
+        let (vault, client) = scripted_client(vec![
+            ScriptedResponse::error(503, "temporarily unavailable"),
+            ScriptedResponse::ok(kv2_read_data(&healthy_key_data())),
+        ])
+        .await;
+
+        let key_data = client
+            .get_key_data("wired-key")
+            .await
+            .expect("read must retry past a transient 503");
+        assert_eq!(key_data.algorithm, "AES_256");
+
+        let requests = vault.requests();
+        assert_eq!(requests.len(), 2, "one failed attempt plus one retry: {requests:?}");
+        assert!(
+            requests
+                .iter()
+                .all(|line| line == "GET /v1/secret/data/rustfs/kms/keys/wired-key"),
+            "both attempts must hit the same read endpoint: {requests:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn wired_read_does_not_retry_permission_errors() {
+        let (vault, client) = scripted_client(vec![ScriptedResponse::error(403, "permission denied")]).await;
+
+        client
+            .get_key_data("wired-key")
+            .await
+            .expect_err("a 403 must fail the read outright");
+
+        let requests = vault.requests();
+        assert_eq!(requests.len(), 1, "fatal statuses must not be retried: {requests:?}");
+    }
+
+    #[tokio::test]
+    async fn wired_write_is_never_retried_on_transient_status() {
+        let (vault, client) = scripted_client(vec![ScriptedResponse::error(503, "sealed")]).await;
+
+        let error = client
+            .store_key_data("wired-key", &healthy_key_data())
+            .await
+            .expect_err("the scripted 503 must fail the write");
+        assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
+
+        let requests = vault.requests();
+        assert_eq!(
+            requests,
+            vec!["POST /v1/secret/data/rustfs/kms/keys/wired-key".to_string()],
+            "a non-idempotent write must run exactly once even on a retryable status"
+        );
+    }
+
+    #[tokio::test]
+    async fn wired_cas_conflict_is_surfaced_without_retry() {
+        let (vault, client) = scripted_client(vec![ScriptedResponse::error(
+            400,
+            "check-and-set parameter did not match the current version",
+        )])
+        .await;
+
+        let error = client
+            .cas_store_key_data("wired-key", &healthy_key_data(), 7)
+            .await
+            .expect_err("the scripted CAS conflict must fail the write");
+        assert!(
+            matches!(error, KmsError::InvalidOperation { .. }),
+            "a CAS conflict is a concurrency signal, not a backend failure: {error:?}"
+        );
+
+        let requests = vault.requests();
+        assert_eq!(requests.len(), 1, "a CAS conflict must never be retried: {requests:?}");
+    }
+
+    #[tokio::test]
+    async fn wired_create_key_read_confirms_identical_existing_key() {
+        // The stored key is exactly what create_key("wired-key", "AES_256")
+        // would have produced, so a retried create whose first response was
+        // lost recovers by reading it back instead of failing.
+        let (vault, client) = scripted_client(vec![ScriptedResponse::ok(kv2_read_data(&healthy_key_data()))]).await;
+
+        let recovered = client
+            .create_key("wired-key", "AES_256", None)
+            .await
+            .expect("an identical active key must read-confirm as a recovered create");
+        assert_eq!(recovered.version, 1);
+        assert_eq!(recovered.algorithm, "AES_256");
+
+        let requests = vault.requests();
+        assert_eq!(
+            requests,
+            vec!["GET /v1/secret/data/rustfs/kms/keys/wired-key".to_string()],
+            "a recovered create must not write anything"
+        );
+    }
+
+    #[tokio::test]
+    async fn wired_create_key_still_fails_on_mismatched_existing_key() {
+        let mut disabled = healthy_key_data();
+        disabled.status = KeyStatus::Disabled;
+        let (vault, client) = scripted_client(vec![ScriptedResponse::ok(kv2_read_data(&disabled))]).await;
+
+        let error = client
+            .create_key("wired-key", "AES_256", None)
+            .await
+            .expect_err("a non-active existing key must keep failing the create");
+        assert!(matches!(error, KmsError::KeyAlreadyExists { .. }), "got {error:?}");
+
+        let requests = vault.requests();
+        assert_eq!(requests.len(), 1, "the mismatch must be decided from the single read: {requests:?}");
+    }
+
     /// Poison matrix for the read-side material gate. Every corruption class must fail
     /// closed with its typed error; reintroducing any "self-heal" (regenerate on empty or
     /// undecodable material) turns one of these expected errors into an Ok and fails the

crates/kms/src/backends/vault_transit.rs

@@ -24,15 +24,19 @@ use crate::backends::{
 use crate::config::{KmsConfig, VaultTransitConfig};
 use crate::encryption::{DataKeyEnvelope, generate_key_material};
 use crate::error::{KmsError, Result};
+use crate::policy::{self, AttemptError, OpClass, RetryPolicy};
 use crate::types::*;
 use async_trait::async_trait;
 use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
 use jiff::Zoned;
 use serde::{Deserialize, Serialize};
 use std::collections::{BTreeMap, HashMap};
+use std::future::Future;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::RwLock;
+use tokio_util::sync::CancellationToken;
+use tracing::info;
 use vaultrs::{
     api::transit::{
         KeyType,
@@ -147,6 +151,13 @@ pub struct VaultTransitKmsClient {
     /// Path prefix under metadata_kv_mount for storing transit key metadata records
     metadata_key_prefix: String,
     metadata_cache: RwLock<HashMap<String, TransitKeyMetadata>>,
+    /// Budgets wrapping every outbound Vault call (see `crate::policy`).
+    retry: RetryPolicy,
+    /// Cancellation point for the operation executor: aborts in-flight
+    /// attempts and backoff sleeps. Owned by the client and currently never
+    /// triggered — shutdown drops the whole client — but kept as the single
+    /// hook a future lifecycle owner can cancel through.
+    cancel: CancellationToken,
 }
 
 impl VaultTransitKmsClient {
@@ -171,6 +182,8 @@ impl VaultTransitKmsClient {
             metadata_key_prefix: config.metadata_key_prefix.clone(),
             config,
             metadata_cache: RwLock::new(HashMap::new()),
+            retry: RetryPolicy::from_config(kms_config),
+            cancel: CancellationToken::new(),
         })
     }
 
@@ -183,6 +196,19 @@ impl VaultTransitKmsClient {
         self.credentials.current()
     }
 
+    /// Run one Vault call under the operation policy.
+    ///
+    /// The closure performs a single classified attempt and takes a fresh
+    /// credential snapshot per attempt, so a retry after a credential rotation
+    /// uses the new token.
+    async fn run<T, F, Fut>(&self, operation: &'static str, class: OpClass, attempt: F) -> Result<T>
+    where
+        F: FnMut() -> Fut,
+        Fut: Future<Output = std::result::Result<T, AttemptError>>,
+    {
+        policy::execute(operation, class, &self.retry, &self.cancel, attempt).await
+    }
+
     fn canonicalize_context(encryption_context: &HashMap<String, String>) -> Result<Option<String>> {
         if encryption_context.is_empty() {
             return Ok(None);
@@ -196,28 +222,40 @@ impl VaultTransitKmsClient {
         Ok(Some(BASE64.encode(serialized)))
     }
 
-    fn map_vault_error<T>(key_id: &str, error: vaultrs::error::ClientError, operation: &str) -> Result<T> {
+    fn map_vault_error(key_id: &str, error: vaultrs::error::ClientError, operation: &str) -> KmsError {
         match error {
-            vaultrs::error::ClientError::ResponseWrapError => Err(KmsError::key_not_found(key_id)),
-            vaultrs::error::ClientError::APIError { code: 404, .. } => Err(KmsError::key_not_found(key_id)),
-            other => Err(KmsError::backend_error(format!(
-                "Vault Transit {operation} failed for key {key_id}: {other}"
-            ))),
+            vaultrs::error::ClientError::ResponseWrapError => KmsError::key_not_found(key_id),
+            vaultrs::error::ClientError::APIError { code: 404, .. } => KmsError::key_not_found(key_id),
+            other => KmsError::backend_error(format!("Vault Transit {operation} failed for key {key_id}: {other}")),
         }
     }
 
     async fn read_transit_key(&self, key_id: &str) -> Result<vaultrs::api::transit::responses::ReadKeyResponse> {
-        key::read(&self.vault()?.client, &self.config.mount_path, key_id)
-            .await
-            .or_else(|e| Self::map_vault_error(key_id, e, "read"))
+        self.run("vault_transit_read_key", OpClass::ReadIdempotent, move || async move {
+            let vault = self.vault().map_err(AttemptError::fatal)?;
+            key::read(&vault.client, &self.config.mount_path, key_id)
+                .await
+                .map_err(|e| AttemptError::from_vaultrs(e, |e| Self::map_vault_error(key_id, e, "read")))
+        })
+        .await
     }
 
     async fn create_transit_key(&self, key_id: &str) -> Result<()> {
-        let mut builder = CreateKeyRequestBuilder::default();
-        builder.key_type(KeyType::Aes256Gcm96);
-        key::create(&self.vault()?.client, &self.config.mount_path, key_id, Some(&mut builder))
-            .await
-            .map_err(|e| KmsError::backend_error(format!("Failed to create Vault Transit key {key_id}: {e}")))
+        // Single attempt: create carries external side effects and the caller
+        // owns the read-confirm recovery for lost responses.
+        self.run("vault_transit_create_key", OpClass::MutatingNonIdempotent, move || async move {
+            let vault = self.vault().map_err(AttemptError::fatal)?;
+            let mut builder = CreateKeyRequestBuilder::default();
+            builder.key_type(KeyType::Aes256Gcm96);
+            key::create(&vault.client, &self.config.mount_path, key_id, Some(&mut builder))
+                .await
+                .map_err(|e| {
+                    AttemptError::from_vaultrs(e, |e| {
+                        KmsError::backend_error(format!("Failed to create Vault Transit key {key_id}: {e}"))
+                    })
+                })
+        })
+        .await
     }
 
     async fn transit_encrypt(
@@ -227,14 +265,26 @@ impl VaultTransitKmsClient {
         encryption_context: &HashMap<String, String>,
     ) -> Result<String> {
         let plaintext_b64 = BASE64.encode(plaintext);
-        let mut builder = EncryptDataRequestBuilder::default();
-        if let Some(aad) = Self::canonicalize_context(encryption_context)? {
-            builder.associated_data(aad);
-        }
+        let plaintext_b64 = plaintext_b64.as_str();
+        let aad = Self::canonicalize_context(encryption_context)?;
+        let aad = aad.as_deref();
 
-        let response = data::encrypt(&self.vault()?.client, &self.config.mount_path, key_id, &plaintext_b64, Some(&mut builder))
-            .await
-            .map_err(|e| KmsError::backend_error(format!("Failed to encrypt data with Vault Transit key {key_id}: {e}")))?;
+        let response = self
+            .run("vault_transit_encrypt", OpClass::ReadIdempotent, move || async move {
+                let vault = self.vault().map_err(AttemptError::fatal)?;
+                let mut builder = EncryptDataRequestBuilder::default();
+                if let Some(aad) = aad {
+                    builder.associated_data(aad);
+                }
+                data::encrypt(&vault.client, &self.config.mount_path, key_id, plaintext_b64, Some(&mut builder))
+                    .await
+                    .map_err(|e| {
+                        AttemptError::from_vaultrs(e, |e| {
+                            KmsError::backend_error(format!("Failed to encrypt data with Vault Transit key {key_id}: {e}"))
+                        })
+                    })
+            })
+            .await?;
 
         Ok(response.ciphertext)
     }
@@ -245,14 +295,25 @@ impl VaultTransitKmsClient {
         ciphertext: &str,
         encryption_context: &HashMap<String, String>,
     ) -> Result<Vec<u8>> {
-        let mut builder = DecryptDataRequestBuilder::default();
-        if let Some(aad) = Self::canonicalize_context(encryption_context)? {
-            builder.associated_data(aad);
-        }
+        let aad = Self::canonicalize_context(encryption_context)?;
+        let aad = aad.as_deref();
 
-        let response = data::decrypt(&self.vault()?.client, &self.config.mount_path, key_id, ciphertext, Some(&mut builder))
-            .await
-            .map_err(|e| KmsError::backend_error(format!("Failed to decrypt data with Vault Transit key {key_id}: {e}")))?;
+        let response = self
+            .run("vault_transit_decrypt", OpClass::ReadIdempotent, move || async move {
+                let vault = self.vault().map_err(AttemptError::fatal)?;
+                let mut builder = DecryptDataRequestBuilder::default();
+                if let Some(aad) = aad {
+                    builder.associated_data(aad);
+                }
+                data::decrypt(&vault.client, &self.config.mount_path, key_id, ciphertext, Some(&mut builder))
+                    .await
+                    .map_err(|e| {
+                        AttemptError::from_vaultrs(e, |e| {
+                            KmsError::backend_error(format!("Failed to decrypt data with Vault Transit key {key_id}: {e}"))
+                        })
+                    })
+            })
+            .await?;
 
         BASE64
             .decode(response.plaintext)
@@ -265,33 +326,93 @@ impl VaultTransitKmsClient {
 
     async fn read_metadata_from_kv(&self, key_id: &str) -> Result<Option<TransitKeyMetadata>> {
         let path = self.metadata_key_path(key_id);
-        match kv2::read::<TransitKeyMetadataPersisted>(&self.vault()?.client, &self.metadata_kv_mount, &path).await {
-            Ok(persisted) => Ok(Some(persisted.into())),
-            Err(vaultrs::error::ClientError::ResponseWrapError)
-            | Err(vaultrs::error::ClientError::APIError { code: 404, .. }) => Ok(None),
-            Err(e) => Err(KmsError::backend_error(format!("Failed to read transit key metadata from Vault KV: {e}"))),
-        }
+        let path = path.as_str();
+        self.run("vault_transit_read_metadata", OpClass::ReadIdempotent, move || async move {
+            let vault = self.vault().map_err(AttemptError::fatal)?;
+            match kv2::read::<TransitKeyMetadataPersisted>(&vault.client, &self.metadata_kv_mount, path).await {
+                Ok(persisted) => Ok(Some(persisted.into())),
+                Err(vaultrs::error::ClientError::ResponseWrapError)
+                | Err(vaultrs::error::ClientError::APIError { code: 404, .. }) => Ok(None),
+                Err(e) => Err(AttemptError::from_vaultrs(e, |e| {
+                    KmsError::backend_error(format!("Failed to read transit key metadata from Vault KV: {e}"))
+                })),
+            }
+        })
+        .await
     }
 
     async fn write_metadata_to_kv(&self, key_id: &str, metadata: &TransitKeyMetadata) -> Result<()> {
         let path = self.metadata_key_path(key_id);
+        let path = path.as_str();
         let persisted: TransitKeyMetadataPersisted = metadata.clone().into();
-        kv2::set(&self.vault()?.client, &self.metadata_kv_mount, &path, &persisted)
-            .await
-            .map(|_| ())
-            .map_err(|e| KmsError::backend_error(format!("Failed to write transit key metadata to Vault KV: {e}")))
+        let persisted = &persisted;
+        // Single attempt: this is a whole-record overwrite without a CAS
+        // precondition, so a replay after a lost response could clobber a
+        // concurrent writer.
+        self.run("vault_transit_write_metadata", OpClass::MutatingNonIdempotent, move || async move {
+            let vault = self.vault().map_err(AttemptError::fatal)?;
+            kv2::set(&vault.client, &self.metadata_kv_mount, path, persisted)
+                .await
+                .map(|_| ())
+                .map_err(|e| {
+                    AttemptError::from_vaultrs(e, |e| {
+                        KmsError::backend_error(format!("Failed to write transit key metadata to Vault KV: {e}"))
+                    })
+                })
+        })
+        .await
     }
 
     async fn delete_metadata_from_kv(&self, key_id: &str) -> Result<()> {
         let path = self.metadata_key_path(key_id);
-        match kv2::delete_metadata(&self.vault()?.client, &self.metadata_kv_mount, &path).await {
-            Ok(_) => Ok(()),
-            Err(vaultrs::error::ClientError::ResponseWrapError)
-            | Err(vaultrs::error::ClientError::APIError { code: 404, .. }) => Ok(()),
-            Err(e) => Err(KmsError::backend_error(format!(
-                "Failed to delete transit key metadata from Vault KV: {e}"
-            ))),
-        }
+        let path = path.as_str();
+        self.run("vault_transit_delete_metadata", OpClass::MutatingNonIdempotent, move || async move {
+            let vault = self.vault().map_err(AttemptError::fatal)?;
+            match kv2::delete_metadata(&vault.client, &self.metadata_kv_mount, path).await {
+                // Metadata that is already gone is a completed delete.
+                Ok(_)
+                | Err(vaultrs::error::ClientError::ResponseWrapError)
+                | Err(vaultrs::error::ClientError::APIError { code: 404, .. }) => Ok(()),
+                Err(e) => Err(AttemptError::from_vaultrs(e, |e| {
+                    KmsError::backend_error(format!("Failed to delete transit key metadata from Vault KV: {e}"))
+                })),
+            }
+        })
+        .await
+    }
+
+    /// Flip `deletion_allowed` on the transit key so it can be deleted.
+    async fn allow_transit_key_deletion(&self, key_id: &str) -> Result<()> {
+        self.run("vault_transit_allow_deletion", OpClass::MutatingNonIdempotent, move || async move {
+            let vault = self.vault().map_err(AttemptError::fatal)?;
+            let mut builder = UpdateKeyConfigurationRequestBuilder::default();
+            builder.deletion_allowed(true);
+            key::update(&vault.client, &self.config.mount_path, key_id, Some(&mut builder))
+                .await
+                .map(|_| ())
+                .map_err(|e| {
+                    AttemptError::from_vaultrs(e, |e| {
+                        KmsError::backend_error(format!("Failed to allow deletion of Vault Transit key {key_id}: {e}"))
+                    })
+                })
+        })
+        .await
+    }
+
+    /// Physically delete the transit key material in Vault.
+    async fn delete_transit_key(&self, key_id: &str) -> Result<()> {
+        self.run("vault_transit_delete_key", OpClass::MutatingNonIdempotent, move || async move {
+            let vault = self.vault().map_err(AttemptError::fatal)?;
+            key::delete(&vault.client, &self.config.mount_path, key_id)
+                .await
+                .map(|_| ())
+                .map_err(|e| {
+                    AttemptError::from_vaultrs(e, |e| {
+                        KmsError::backend_error(format!("Failed to delete Vault Transit key {key_id}: {e}"))
+                    })
+                })
+        })
+        .await
     }
 
     async fn get_key_metadata(&self, key_id: &str) -> Result<TransitKeyMetadata> {
@@ -462,8 +583,40 @@ impl KmsClient for VaultTransitKmsClient {
             return Err(KmsError::unsupported_algorithm(algorithm));
         }
 
-        if self.read_transit_key(key_id).await.is_ok() {
-            return Err(KmsError::key_already_exists(key_id));
+        // Existence pre-check with read-confirm recovery: a create whose
+        // response was lost gets retried by callers, and used to be
+        // misreported as KeyAlreadyExists. Transit keys are always AES-256,
+        // so an existing enabled key of the default usage is exactly what
+        // this create would have produced; report it as the create result.
+        // Anything else keeps failing. A failed pre-check read must fail the
+        // create rather than fall through to re-creating over an unknown key.
+        match self.read_transit_key(key_id).await {
+            Ok(_) => {
+                let existing = self.get_key_metadata(key_id).await?;
+                return if existing.key_state == KeyState::Enabled && existing.key_usage == KeyUsage::EncryptDecrypt {
+                    info!(
+                        key_id,
+                        "Vault Transit create found an identical enabled key; treating it as a recovered create"
+                    );
+                    Ok(MasterKeyInfo {
+                        key_id: key_id.to_string(),
+                        version: existing.current_version,
+                        algorithm: algorithm.to_string(),
+                        usage: existing.key_usage,
+                        status: KeyStatus::Active,
+                        description: existing.description,
+                        metadata: existing.tags.clone(),
+                        created_at: existing.created_at,
+                        rotated_at: None,
+                        created_by: existing.created_by,
+                        deletion_date: None,
+                    })
+                } else {
+                    Err(KmsError::key_already_exists(key_id))
+                };
+            }
+            Err(KmsError::KeyNotFound { .. }) => {}
+            Err(error) => return Err(error),
         }
 
         self.create_transit_key(key_id).await?;
@@ -497,9 +650,14 @@ impl KmsClient for VaultTransitKmsClient {
     }
 
     async fn list_keys(&self, request: &ListKeysRequest, _context: Option<&OperationContext>) -> Result<ListKeysResponse> {
-        let all_keys = key::list(&self.vault()?.client, &self.config.mount_path)
-            .await
-            .map_err(|e| KmsError::backend_error(format!("Failed to list Vault Transit keys: {e}")))?
+        let all_keys = self
+            .run("vault_transit_list_keys", OpClass::ReadIdempotent, move || async move {
+                let vault = self.vault().map_err(AttemptError::fatal)?;
+                key::list(&vault.client, &self.config.mount_path).await.map_err(|e| {
+                    AttemptError::from_vaultrs(e, |e| KmsError::backend_error(format!("Failed to list Vault Transit keys: {e}")))
+                })
+            })
+            .await?
             .keys;
 
         let mut filtered = Vec::new();
@@ -576,9 +734,20 @@ impl KmsClient for VaultTransitKmsClient {
     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}")))?;
+        // Single attempt, never retried: replaying a rotate whose response was
+        // lost would advance the key version once more per replay.
+        self.run("vault_transit_rotate_key", OpClass::MutatingNonIdempotent, move || async move {
+            let vault = self.vault().map_err(AttemptError::fatal)?;
+            key::rotate(&vault.client, &self.config.mount_path, key_id)
+                .await
+                .map(|_| ())
+                .map_err(|e| {
+                    AttemptError::from_vaultrs(e, |e| {
+                        KmsError::backend_error(format!("Failed to rotate Vault Transit key {key_id}: {e}"))
+                    })
+                })
+        })
+        .await?;
 
         let mut metadata = self.get_key_metadata(key_id).await?;
         metadata.current_version += 1;
@@ -600,10 +769,16 @@ impl KmsClient for VaultTransitKmsClient {
     }
 
     async fn health_check(&self) -> Result<()> {
-        key::list(&self.vault()?.client, &self.config.mount_path)
-            .await
-            .map(|_| ())
-            .map_err(|e| KmsError::backend_error(format!("Vault Transit health check failed: {e}")))
+        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)
+                .await
+                .map(|_| ())
+                .map_err(|e| {
+                    AttemptError::from_vaultrs(e, |e| KmsError::backend_error(format!("Vault Transit health check failed: {e}")))
+                })
+        })
+        .await
     }
 
     fn backend_info(&self) -> BackendInfo {
@@ -660,8 +835,46 @@ impl VaultTransitKmsBackend {
 impl KmsBackend for VaultTransitKmsBackend {
     async fn create_key(&self, request: CreateKeyRequest) -> Result<CreateKeyResponse> {
         let key_id = request.key_name.clone().unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
-        if self.client.read_transit_key(&key_id).await.is_ok() {
-            return Err(KmsError::key_already_exists(&key_id));
+
+        // Existence pre-check with read-confirm recovery: a create whose
+        // response was lost gets retried by callers, and used to be
+        // misreported as KeyAlreadyExists. If the stored record is exactly
+        // what this request would have written, report it as the create
+        // result; any divergence keeps failing so a create can never adopt or
+        // reshape a key it would not have produced.
+        match self.client.read_transit_key(&key_id).await {
+            Ok(_) => {
+                let existing = self.client.get_key_metadata(&key_id).await?;
+                let requested = TransitKeyMetadata::from_create_request(&request);
+                return if existing.key_state == KeyState::Enabled
+                    && existing.key_usage == requested.key_usage
+                    && existing.description == requested.description
+                    && existing.tags == requested.tags
+                {
+                    info!(
+                        key_id,
+                        "Vault Transit create found an identical enabled key; treating it as a recovered create"
+                    );
+                    Ok(CreateKeyResponse {
+                        key_id: key_id.clone(),
+                        key_metadata: KeyMetadata {
+                            key_id,
+                            key_state: existing.key_state,
+                            key_usage: existing.key_usage,
+                            description: existing.description,
+                            creation_date: existing.created_at,
+                            deletion_date: existing.deletion_date,
+                            origin: existing.origin,
+                            key_manager: "VAULT_TRANSIT".to_string(),
+                            tags: existing.tags,
+                        },
+                    })
+                } else {
+                    Err(KmsError::key_already_exists(&key_id))
+                };
+            }
+            Err(KmsError::KeyNotFound { .. }) => {}
+            Err(error) => return Err(error),
         }
 
         self.client.create_transit_key(&key_id).await?;
@@ -734,22 +947,9 @@ impl KmsBackend for VaultTransitKmsBackend {
         let deletion_date = if request.force_immediate.unwrap_or(false) {
             if key_metadata.key_state == KeyState::PendingDeletion {
                 if !self.client.read_transit_key(&key_id).await?.deletion_allowed {
-                    let mut update_builder = UpdateKeyConfigurationRequestBuilder::default();
-                    update_builder.deletion_allowed(true);
-                    key::update(
-                        &self.client.vault()?.client,
-                        &self.client.config.mount_path,
-                        &key_id,
-                        Some(&mut update_builder),
-                    )
-                    .await
-                    .map_err(|e| {
-                        KmsError::backend_error(format!("Failed to allow deletion of Vault Transit key {key_id}: {e}"))
-                    })?;
+                    self.client.allow_transit_key_deletion(&key_id).await?;
                 }
-                key::delete(&self.client.vault()?.client, &self.client.config.mount_path, &key_id)
-                    .await
-                    .map_err(|e| KmsError::backend_error(format!("Failed to delete Vault Transit key {key_id}: {e}")))?;
+                self.client.delete_transit_key(&key_id).await?;
                 self.client.delete_key_metadata(&key_id).await?;
                 None
             } else {
@@ -852,20 +1052,9 @@ impl KmsBackend for VaultTransitKmsBackend {
         }
 
         if !self.client.read_transit_key(key_id).await?.deletion_allowed {
-            let mut update_builder = UpdateKeyConfigurationRequestBuilder::default();
-            update_builder.deletion_allowed(true);
-            key::update(
-                &self.client.vault()?.client,
-                &self.client.config.mount_path,
-                key_id,
-                Some(&mut update_builder),
-            )
-            .await
-            .map_err(|e| KmsError::backend_error(format!("Failed to allow deletion of Vault Transit key {key_id}: {e}")))?;
+            self.client.allow_transit_key_deletion(key_id).await?;
         }
-        key::delete(&self.client.vault()?.client, &self.client.config.mount_path, key_id)
-            .await
-            .map_err(|e| KmsError::backend_error(format!("Failed to delete Vault Transit key {key_id}: {e}")))?;
+        self.client.delete_transit_key(key_id).await?;
         self.client.delete_key_metadata(key_id).await?;
         Ok(ExpiredKeyRemoval::Removed)
     }
@@ -874,10 +1063,167 @@ impl KmsBackend for VaultTransitKmsBackend {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::backends::scripted_vault::{ScriptedResponse, ScriptedVault};
     use crate::config::{
         DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX, DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT, VaultAuthMethod, VaultTransitConfig,
     };
     use crate::types::KeyStatus;
+    use vaultrs::api::transit::responses::{ReadKeyData, ReadKeyResponse};
+
+    async fn scripted_client(responses: Vec<ScriptedResponse>) -> (ScriptedVault, VaultTransitKmsClient) {
+        let vault = ScriptedVault::serve(responses).await;
+        let config = VaultTransitConfig {
+            address: vault.address.clone(),
+            ..test_vault_transit_config()
+        };
+        let kms_config = KmsConfig {
+            timeout: Duration::from_secs(5),
+            retry_attempts: 3,
+            ..KmsConfig::default()
+        };
+        let client = VaultTransitKmsClient::new(config, &kms_config)
+            .await
+            .expect("scripted Vault Transit client");
+        (vault, client)
+    }
+
+    /// KV2 read payload for a persisted transit metadata record.
+    fn metadata_read_data(metadata: &TransitKeyMetadata) -> serde_json::Value {
+        let persisted: TransitKeyMetadataPersisted = metadata.clone().into();
+        serde_json::json!({
+            "data": serde_json::to_value(&persisted).expect("serialize transit metadata"),
+            "metadata": {
+                "created_time": "2026-01-01T00:00:00Z",
+                "deletion_time": "",
+                "custom_metadata": null,
+                "destroyed": false,
+                "version": 1,
+            },
+        })
+    }
+
+    /// Transit read-key payload for an existing symmetric key.
+    fn transit_key_read_data(key_id: &str) -> serde_json::Value {
+        let response = ReadKeyResponse {
+            key_type: KeyType::Aes256Gcm96,
+            deletion_allowed: false,
+            derived: false,
+            exportable: false,
+            allow_plaintext_backup: false,
+            keys: ReadKeyData::Symmetric(HashMap::from([("1".to_string(), 1_700_000_000_u64)])),
+            min_decryption_version: 1,
+            min_encryption_version: 0,
+            name: key_id.to_string(),
+            supports_encryption: true,
+            supports_decryption: true,
+            supports_derivation: false,
+            supports_signing: false,
+            imported: Some(false),
+        };
+        serde_json::to_value(&response).expect("serialize transit key read response")
+    }
+
+    #[tokio::test]
+    async fn wired_transit_encrypt_retries_transient_status() {
+        let metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default());
+        let (vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(metadata_read_data(&metadata)),
+            ScriptedResponse::error(429, "throttled"),
+            ScriptedResponse::ok(serde_json::json!({ "ciphertext": "vault:v1:scripted" })),
+        ])
+        .await;
+
+        let response = client
+            .encrypt(
+                &EncryptRequest {
+                    key_id: "wired-key".to_string(),
+                    plaintext: b"plaintext".to_vec(),
+                    encryption_context: HashMap::new(),
+                    grant_tokens: Vec::new(),
+                },
+                None,
+            )
+            .await
+            .expect("encrypt must retry past a transient 429");
+        assert_eq!(response.ciphertext, b"vault:v1:scripted".to_vec());
+
+        let requests = vault.requests();
+        assert_eq!(requests.len(), 3, "metadata read plus two encrypt attempts: {requests:?}");
+        assert_eq!(requests[1], "POST /v1/transit/encrypt/wired-key");
+        assert_eq!(requests[2], "POST /v1/transit/encrypt/wired-key");
+    }
+
+    #[tokio::test]
+    async fn wired_transit_rotate_is_never_retried() {
+        let metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default());
+        let (vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(metadata_read_data(&metadata)),
+            ScriptedResponse::error(503, "standby"),
+        ])
+        .await;
+
+        let error = client
+            .rotate_key("wired-key", None)
+            .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(), 2, "metadata read plus exactly one rotate attempt: {requests:?}");
+        assert_eq!(
+            requests[1], "POST /v1/transit/keys/wired-key/rotate",
+            "a rotation must never be replayed: {requests:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn wired_transit_create_read_confirms_identical_existing_key() {
+        // The stored key and metadata are exactly what this create would have
+        // produced, so a retried create whose first response was lost recovers
+        // by reading them back instead of failing.
+        let metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default());
+        let (vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(transit_key_read_data("wired-key")),
+            ScriptedResponse::ok(metadata_read_data(&metadata)),
+        ])
+        .await;
+
+        let recovered = client
+            .create_key("wired-key", "AES_256", None)
+            .await
+            .expect("an identical enabled key must read-confirm as a recovered create");
+        assert_eq!(recovered.status, KeyStatus::Active);
+
+        let requests = vault.requests();
+        assert_eq!(requests.len(), 2, "read-confirm must be decided from reads alone: {requests:?}");
+        assert!(
+            requests.iter().all(|line| line.starts_with("GET ")),
+            "a recovered create must not write anything: {requests:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn wired_transit_create_still_fails_on_mismatched_existing_key() {
+        let mut metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default());
+        metadata.key_state = KeyState::Disabled;
+        let (vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(transit_key_read_data("wired-key")),
+            ScriptedResponse::ok(metadata_read_data(&metadata)),
+        ])
+        .await;
+
+        let error = client
+            .create_key("wired-key", "AES_256", None)
+            .await
+            .expect_err("a non-enabled existing key must keep failing the create");
+        assert!(matches!(error, KmsError::KeyAlreadyExists { .. }), "got {error:?}");
+
+        let requests = vault.requests();
+        assert!(
+            requests.iter().all(|line| line.starts_with("GET ")),
+            "the rejected create must not write anything: {requests:?}"
+        );
+    }
 
     fn test_vault_transit_config() -> VaultTransitConfig {
         VaultTransitConfig {

crates/kms/src/backup/local_export.rs

@@ -0,0 +1,996 @@
+// 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.
+
+//! Local backend backup export: sealed, KEK-protected bundle production.
+//!
+//! This is the producer side only; restore lives in a follow-up change. The
+//! admin API is not wired here either — callers construct the request and
+//! supply the backup KEK explicitly.
+//!
+//! # Bundle layout
+//!
+//! A bundle is a directory (simple to produce, artifacts stream one file at a
+//! time, and partial output is trivially recognizable because the manifest is
+//! written last):
+//!
+//! ```text
+//! <destination>/
+//!   manifest.json                      # sealed BackupManifest, written last
+//!   artifacts/keys/<key_id>.key.enc    # one per stored key record
+//!   artifacts/master-key.salt.enc      # present when the salt file exists
+//! ```
+//!
+//! # Artifact payload framing
+//!
+//! Every artifact payload is `nonce (12 bytes) || AES-256-GCM ciphertext`,
+//! encrypted under the caller-supplied backup KEK with an AAD binding of
+//! `(context, backup_id, snapshot_generation, artifact path)`, so an artifact
+//! cannot be swapped into another bundle or renamed within its own bundle.
+//! Records already encrypted at rest stay encrypted inside the wrap;
+//! plaintext-dev-only records become ciphertext-only in the bundle, which is
+//! their mandatory re-wrap under the backup KEK.
+//!
+//! # Write protocol
+//!
+//! Artifacts are written and fsynced first, re-read and digest-verified, and
+//! only then is the sealed manifest (completeness marker plus final digest)
+//! published. A crash at any earlier point leaves a bundle without a
+//! manifest, which decodes as an incomplete bundle and can never be restored.
+
+use crate::backends::local::{LocalKmsClient, StoredKeyProtection};
+use crate::backup::capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
+use crate::backup::error::BackupError;
+use crate::backup::manifest::{
+    AeadAlgorithm, ArtifactDescriptor, ArtifactKind, BackupKekDescriptor, BackupManifest, CompletenessState, ContentDigest,
+    DigestAlgorithm, LocalKdfDescriptor, LocalKeyDerivation,
+};
+use crate::error::{KmsError, Result};
+use aes_gcm::{
+    Aes256Gcm, Key, Nonce,
+    aead::{Aead, KeyInit, Payload},
+};
+use jiff::Zoned;
+use rand::RngExt;
+use serde::Deserialize;
+use std::path::{Path, PathBuf};
+use tokio::fs;
+use tokio::io::AsyncWriteExt;
+use zeroize::Zeroizing;
+
+/// File name of the sealed manifest inside a bundle directory.
+pub const LOCAL_BUNDLE_MANIFEST_FILE: &str = "manifest.json";
+const ARTIFACTS_DIR: &str = "artifacts";
+const KEYS_DIR: &str = "artifacts/keys";
+const SALT_ARTIFACT_PATH: &str = "artifacts/master-key.salt.enc";
+const AEAD_NONCE_LEN: usize = 12;
+/// Domain-separation context for the artifact AAD binding.
+const BUNDLE_AAD_CONTEXT: &str = "rustfs-kms-local-backup:v1";
+
+/// Caller-supplied backup KEK: a trust root deliberately separate from the
+/// business KMS hierarchy (it must not be a key that is itself part of the
+/// state being backed up). Where the KEK comes from is the admin layer's
+/// concern; this module only consumes it.
+pub struct BackupKek {
+    kek_id: String,
+    kek_version: u32,
+    key: Zeroizing<[u8; 32]>,
+}
+
+impl BackupKek {
+    /// Wrap 32 bytes of KEK material. The material is zeroized on drop;
+    /// callers should zeroize their own copy of the input.
+    pub fn new(kek_id: impl Into<String>, kek_version: u32, key: [u8; 32]) -> Result<Self> {
+        let kek_id = kek_id.into();
+        if kek_id.is_empty() {
+            return Err(KmsError::validation_error("backup KEK id must not be empty"));
+        }
+        Ok(Self {
+            kek_id,
+            kek_version,
+            key: Zeroizing::new(key),
+        })
+    }
+
+    /// Manifest descriptor for this KEK.
+    pub fn descriptor(&self) -> BackupKekDescriptor {
+        BackupKekDescriptor {
+            kek_id: self.kek_id.clone(),
+            kek_version: self.kek_version,
+            aead_algorithm: AeadAlgorithm::Aes256Gcm,
+        }
+    }
+
+    fn cipher(&self) -> Aes256Gcm {
+        Aes256Gcm::new(&Key::<Aes256Gcm>::from(*self.key))
+    }
+}
+
+/// Parameters of one export run.
+///
+/// `snapshot_generation` is injected by the caller: the contract only
+/// requires it to be monotonic per deployment, and the source (persisted
+/// counter, coordinated clock) is decided by the admin layer, which keeps
+/// this module free of ambient time or state lookups.
+#[derive(Debug, Clone)]
+pub struct LocalBackupExportRequest {
+    /// Unique identifier for this backup.
+    pub backup_id: String,
+    /// Opaque identity of the producing deployment.
+    pub deployment_identity: String,
+    /// RustFS version string recorded in the manifest.
+    pub rustfs_version: String,
+    /// Monotonic snapshot generation this bundle belongs to.
+    pub snapshot_generation: u64,
+    /// Bundle output directory; must not exist yet or must be empty.
+    pub destination: PathBuf,
+}
+
+impl LocalBackupExportRequest {
+    fn validate(&self) -> Result<()> {
+        for (field, value) in [
+            ("backup_id", &self.backup_id),
+            ("deployment_identity", &self.deployment_identity),
+            ("rustfs_version", &self.rustfs_version),
+        ] {
+            if value.is_empty() {
+                return Err(KmsError::validation_error(format!("backup export {field} must not be empty")));
+            }
+        }
+        Ok(())
+    }
+}
+
+/// Minimal projection of a stored key record: only the fields the exporter
+/// needs. Unknown fields are ignored on purpose — the record travels into the
+/// bundle byte-identical, so the exporter must not constrain its schema.
+#[derive(Deserialize)]
+struct StoredRecordProbe {
+    key_id: String,
+    #[serde(default)]
+    at_rest_protection: StoredKeyProtection,
+}
+
+struct CollectedRecord {
+    key_id: String,
+    protection: StoredKeyProtection,
+    /// Raw record bytes exactly as stored. Zeroized on drop because
+    /// plaintext-dev-only records embed key material.
+    raw: Zeroizing<Vec<u8>>,
+}
+
+struct CollectedSnapshot {
+    records: Vec<CollectedRecord>,
+    salt: Option<Vec<u8>>,
+}
+
+/// Export the Local backend's key directory as a sealed backup bundle.
+///
+/// The directory scan runs under the export fence, so concurrent
+/// create/update/delete operations are either fully included or fully
+/// excluded — never half a record. Encryption and bundle writing happen after
+/// the fence is released to keep it short.
+///
+/// Returns the sealed manifest that was written to the bundle.
+pub async fn export_local_backup(
+    client: &LocalKmsClient,
+    kek: &BackupKek,
+    request: &LocalBackupExportRequest,
+) -> Result<BackupManifest> {
+    request.validate()?;
+    prepare_destination(&request.destination).await?;
+
+    let snapshot = collect_snapshot(client).await?;
+    if snapshot.records.is_empty() {
+        return Err(KmsError::invalid_operation(
+            "Local backup export found no key records; refusing to publish an empty bundle",
+        ));
+    }
+
+    let has_encrypted = snapshot
+        .records
+        .iter()
+        .any(|record| record.protection == StoredKeyProtection::EncryptedMasterKey);
+    if has_encrypted && snapshot.salt.is_none() {
+        return Err(KmsError::invalid_operation(
+            "key directory contains encrypted-master-key records but the master key salt file is missing; \
+             the bundle would be unrestorable",
+        ));
+    }
+
+    let manifest = build_and_write_bundle(kek, request, &snapshot).await?;
+    Ok(manifest)
+}
+
+/// Read and fully validate the manifest of a local bundle directory.
+///
+/// A directory without a manifest is an interrupted export: the manifest is
+/// written last, so its absence means the bundle never sealed.
+pub async fn read_local_bundle_manifest(bundle_dir: &Path) -> Result<BackupManifest> {
+    let manifest_path = bundle_dir.join(LOCAL_BUNDLE_MANIFEST_FILE);
+    let bytes = match fs::read(&manifest_path).await {
+        Ok(bytes) => bytes,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            return Err(BackupError::incomplete_bundle("bundle has no manifest; the export never sealed it").into());
+        }
+        Err(error) => return Err(error.into()),
+    };
+    let manifest = BackupManifest::decode(&bytes)?;
+    if manifest.backend != BackupBackendKind::Local {
+        return Err(
+            BackupError::corrupted(format!("bundle manifest declares backend {:?}, expected Local", manifest.backend)).into(),
+        );
+    }
+    Ok(manifest)
+}
+
+/// Read, verify, and decrypt one artifact of a local bundle.
+///
+/// Fail-closed order: KEK identity, artifact presence, declared length,
+/// encrypted digest, then AEAD authentication. The returned plaintext is
+/// zeroized on drop.
+pub async fn decrypt_bundle_artifact(
+    bundle_dir: &Path,
+    manifest: &BackupManifest,
+    descriptor: &ArtifactDescriptor,
+    kek: &BackupKek,
+) -> Result<Zeroizing<Vec<u8>>> {
+    manifest.backup_kek.ensure_matches(&kek.kek_id, kek.kek_version)?;
+    if descriptor.aead_algorithm != AeadAlgorithm::Aes256Gcm {
+        return Err(KmsError::unsupported_algorithm(format!(
+            "{:?} (local bundles are produced with AES-256-GCM)",
+            descriptor.aead_algorithm
+        )));
+    }
+
+    let artifact_path = bundle_dir.join(&descriptor.path);
+    let payload = match fs::read(&artifact_path).await {
+        Ok(payload) => payload,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            return Err(BackupError::missing_artifact(descriptor.path.clone()).into());
+        }
+        Err(error) => return Err(error.into()),
+    };
+
+    if (payload.len() as u64) < descriptor.len {
+        return Err(BackupError::truncated(format!(
+            "artifact '{}' is {} bytes, manifest declares {}",
+            descriptor.path,
+            payload.len(),
+            descriptor.len
+        ))
+        .into());
+    }
+    if payload.len() as u64 != descriptor.len {
+        return Err(BackupError::corrupted(format!(
+            "artifact '{}' is {} bytes, manifest declares {}",
+            descriptor.path,
+            payload.len(),
+            descriptor.len
+        ))
+        .into());
+    }
+    if ContentDigest::sha256_of(&payload) != descriptor.encrypted_digest {
+        return Err(BackupError::corrupted(format!("artifact '{}' does not match its manifest digest", descriptor.path)).into());
+    }
+    if payload.len() < AEAD_NONCE_LEN {
+        return Err(BackupError::corrupted(format!("artifact '{}' is too short to carry a nonce", descriptor.path)).into());
+    }
+
+    let (nonce_bytes, ciphertext) = payload.split_at(AEAD_NONCE_LEN);
+    let mut nonce = [0u8; AEAD_NONCE_LEN];
+    nonce.copy_from_slice(nonce_bytes);
+    let aad = artifact_aad(&manifest.backup_id, manifest.snapshot_generation, &descriptor.path);
+    let plaintext = kek
+        .cipher()
+        .decrypt(
+            &Nonce::from(nonce),
+            Payload {
+                msg: ciphertext,
+                aad: &aad,
+            },
+        )
+        .map_err(|_| {
+            KmsError::from(BackupError::corrupted(format!(
+                "artifact '{}' failed authenticated decryption under the supplied backup KEK",
+                descriptor.path
+            )))
+        })?;
+    Ok(Zeroizing::new(plaintext))
+}
+
+/// Scan the key directory under the export fence.
+async fn collect_snapshot(client: &LocalKmsClient) -> Result<CollectedSnapshot> {
+    let _fence = client.acquire_export_fence().await;
+
+    let mut records = Vec::new();
+    let mut entries = fs::read_dir(client.key_directory()).await?;
+    while let Some(entry) = entries.next_entry().await? {
+        let path = entry.path();
+        if !path.extension().is_some_and(|extension| extension == "key") {
+            continue;
+        }
+        let stem = path
+            .file_stem()
+            .and_then(|stem| stem.to_str())
+            .ok_or_else(|| KmsError::configuration_error("Local KMS key file name must be valid UTF-8"))?
+            .to_string();
+
+        let raw = Zeroizing::new(fs::read(&path).await?);
+        // Any unreadable record aborts the export: a bundle silently missing
+        // one key is worse than no bundle at all.
+        let probe: StoredRecordProbe = serde_json::from_slice(&raw)
+            .map_err(|error| KmsError::material_corrupt(&stem, format!("stored key record does not deserialize: {error}")))?;
+        if probe.key_id != stem {
+            return Err(KmsError::invalid_key(format!(
+                "Local KMS key file identity mismatch: expected {stem:?}, found {:?}",
+                probe.key_id
+            )));
+        }
+
+        records.push(CollectedRecord {
+            key_id: stem,
+            protection: probe.at_rest_protection,
+            raw,
+        });
+    }
+
+    let salt_path = client.master_key_salt_file();
+    let salt = match fs::read(&salt_path).await {
+        Ok(bytes) => Some(bytes),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
+        Err(error) => return Err(error.into()),
+    };
+
+    records.sort_by(|a, b| a.key_id.cmp(&b.key_id));
+    Ok(CollectedSnapshot { records, salt })
+}
+
+async fn build_and_write_bundle(
+    kek: &BackupKek,
+    request: &LocalBackupExportRequest,
+    snapshot: &CollectedSnapshot,
+) -> Result<BackupManifest> {
+    let mut artifacts = Vec::with_capacity(snapshot.records.len() + 1);
+    for record in &snapshot.records {
+        let artifact_path = format!("{KEYS_DIR}/{}.key.enc", record.key_id);
+        let descriptor = encrypt_and_write_artifact(kek, request, ArtifactKind::KeyMaterial, &artifact_path, &record.raw).await?;
+        artifacts.push(descriptor);
+    }
+    if let Some(salt) = &snapshot.salt {
+        let descriptor = encrypt_and_write_artifact(kek, request, ArtifactKind::MasterKeySalt, SALT_ARTIFACT_PATH, salt).await?;
+        artifacts.push(descriptor);
+    }
+
+    // Make the artifact directory entries durable before sealing: the sealed
+    // manifest must never survive a crash that its artifacts did not.
+    fsync_dir(&request.destination.join(KEYS_DIR)).await?;
+    fsync_dir(&request.destination.join(ARTIFACTS_DIR)).await?;
+
+    let manifest = BackupManifest {
+        format_version: BackupManifest::FORMAT_VERSION,
+        backup_id: request.backup_id.clone(),
+        created_at: Zoned::now(),
+        rustfs_version: request.rustfs_version.clone(),
+        deployment_identity: request.deployment_identity.clone(),
+        backend: BackupBackendKind::Local,
+        at_rest_protection: weakest_observed_protection(&snapshot.records),
+        responsibility: BackupResponsibility::FullMaterial,
+        snapshot_generation: request.snapshot_generation,
+        backup_kek: kek.descriptor(),
+        artifacts,
+        local_kdf: Some(local_kdf_descriptor(snapshot)),
+        key_versions: None,
+        capability_discovery: None,
+        completeness: CompletenessState::InProgress,
+        manifest_digest: ContentDigest {
+            algorithm: DigestAlgorithm::Sha256,
+            hex: String::new(),
+        },
+    };
+    let manifest = manifest.seal()?;
+    let manifest_bytes = manifest.encode()?;
+
+    write_new_file(&request.destination.join(LOCAL_BUNDLE_MANIFEST_FILE), &manifest_bytes).await?;
+    fsync_dir(&request.destination).await?;
+    Ok(manifest)
+}
+
+/// Encrypt one artifact, write it durably, and re-read it to verify the
+/// digest before it is allowed into the manifest.
+async fn encrypt_and_write_artifact(
+    kek: &BackupKek,
+    request: &LocalBackupExportRequest,
+    kind: ArtifactKind,
+    artifact_path: &str,
+    plaintext: &[u8],
+) -> Result<ArtifactDescriptor> {
+    let mut nonce = [0u8; AEAD_NONCE_LEN];
+    rand::rng().fill(&mut nonce[..]);
+    let aad = artifact_aad(&request.backup_id, request.snapshot_generation, artifact_path);
+    let ciphertext = kek
+        .cipher()
+        .encrypt(
+            &Nonce::from(nonce),
+            Payload {
+                msg: plaintext,
+                aad: &aad,
+            },
+        )
+        .map_err(|error| KmsError::cryptographic_error("backup_artifact_encrypt", error.to_string()))?;
+
+    let mut payload = Vec::with_capacity(AEAD_NONCE_LEN + ciphertext.len());
+    payload.extend_from_slice(&nonce);
+    payload.extend_from_slice(&ciphertext);
+
+    let absolute_path = request.destination.join(artifact_path);
+    write_new_file(&absolute_path, &payload).await?;
+
+    // Verify what actually landed on disk, not the in-memory buffer.
+    let written = fs::read(&absolute_path).await?;
+    let digest = ContentDigest::sha256_of(&written);
+    if written != payload {
+        return Err(KmsError::internal_error(format!(
+            "bundle artifact '{artifact_path}' read back differently than written"
+        )));
+    }
+
+    Ok(ArtifactDescriptor {
+        kind,
+        path: artifact_path.to_string(),
+        len: payload.len() as u64,
+        aead_algorithm: AeadAlgorithm::Aes256Gcm,
+        encrypted_digest: digest,
+    })
+}
+
+/// AAD binding an artifact to its bundle identity and path. A JSON tuple
+/// gives unambiguous field boundaries without a hand-rolled framing format.
+fn artifact_aad(backup_id: &str, snapshot_generation: u64, artifact_path: &str) -> Vec<u8> {
+    serde_json::to_vec(&(BUNDLE_AAD_CONTEXT, backup_id, snapshot_generation, artifact_path))
+        .expect("AAD tuple of strings and integers always serializes")
+}
+
+/// The bundle-level protection label is the weakest state observed across
+/// records: any plaintext-dev-only record marks the whole bundle, then any
+/// legacy-unspecified marker (unknown until read), and only a uniformly
+/// encrypted directory is labeled encrypted-master-key.
+fn weakest_observed_protection(records: &[CollectedRecord]) -> AtRestProtection {
+    let mut has_legacy = false;
+    for record in records {
+        match record.protection {
+            StoredKeyProtection::PlaintextDevOnly => return AtRestProtection::PlaintextDevOnly,
+            StoredKeyProtection::LegacyUnspecified => has_legacy = true,
+            StoredKeyProtection::EncryptedMasterKey => {}
+        }
+    }
+    if has_legacy {
+        AtRestProtection::LegacyUnspecified
+    } else {
+        AtRestProtection::EncryptedMasterKey
+    }
+}
+
+fn local_kdf_descriptor(snapshot: &CollectedSnapshot) -> LocalKdfDescriptor {
+    let mut modes = Vec::new();
+    for (marker, mode) in [
+        (StoredKeyProtection::EncryptedMasterKey, AtRestProtection::EncryptedMasterKey),
+        (StoredKeyProtection::PlaintextDevOnly, AtRestProtection::PlaintextDevOnly),
+        (StoredKeyProtection::LegacyUnspecified, AtRestProtection::LegacyUnspecified),
+    ] {
+        if snapshot.records.iter().any(|record| record.protection == marker) {
+            modes.push(mode);
+        }
+    }
+
+    // With a salt on disk the backend derives via Argon2id; without one only
+    // the pre-beta.9 SHA-256 derivation can apply. For plaintext-only
+    // directories the derivation is informational.
+    let derivation = if snapshot.salt.is_some() {
+        LocalKeyDerivation::current_argon2id()
+    } else {
+        LocalKeyDerivation::LegacySha256
+    };
+
+    LocalKdfDescriptor {
+        derivation,
+        protection_modes: modes,
+        // The verifier shape is left to the restore change; the schema keeps
+        // it optional so bundles without one stay valid.
+        master_key_verifier: None,
+    }
+}
+
+async fn prepare_destination(destination: &Path) -> Result<()> {
+    if fs::try_exists(destination).await? {
+        let mut entries = fs::read_dir(destination)
+            .await
+            .map_err(|error| KmsError::invalid_operation(format!("backup destination is not a readable directory: {error}")))?;
+        if entries.next_entry().await?.is_some() {
+            return Err(KmsError::invalid_operation(
+                "backup destination directory is not empty; refusing to mix bundles",
+            ));
+        }
+    }
+    fs::create_dir_all(destination.join(KEYS_DIR)).await?;
+    Ok(())
+}
+
+async fn write_new_file(path: &Path, bytes: &[u8]) -> Result<()> {
+    let mut file = fs::OpenOptions::new()
+        .write(true)
+        .create_new(true)
+        .open(path)
+        .await
+        .map_err(|error| KmsError::io_error(format!("failed to create bundle file {}: {error}", path.display())))?;
+    file.write_all(bytes).await?;
+    file.sync_all().await?;
+    Ok(())
+}
+
+/// Fsync a directory so freshly created bundle entries survive power loss.
+/// No-op on non-Unix platforms where directories cannot be opened for
+/// syncing (mirrors the local backend's durable commit helper).
+async fn fsync_dir(path: &Path) -> Result<()> {
+    #[cfg(unix)]
+    {
+        let path = path.to_path_buf();
+        tokio::task::spawn_blocking(move || std::fs::File::open(&path)?.sync_all())
+            .await
+            .map_err(|error| KmsError::io_error(error.to_string()))??;
+    }
+    #[cfg(not(unix))]
+    let _ = path;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::backends::KmsClient;
+    use crate::config::LocalConfig;
+    use std::sync::Arc;
+    use tempfile::TempDir;
+
+    async fn encrypted_client() -> (LocalKmsClient, TempDir) {
+        let temp = TempDir::new().expect("temp dir");
+        let client = LocalKmsClient::new(LocalConfig {
+            key_dir: temp.path().to_path_buf(),
+            master_key: Some("test-master-key".to_string()),
+            file_permissions: Some(0o600),
+        })
+        .await
+        .expect("client should initialize");
+        (client, temp)
+    }
+
+    async fn dev_client() -> (LocalKmsClient, TempDir) {
+        let temp = TempDir::new().expect("temp dir");
+        let client = LocalKmsClient::new(LocalConfig {
+            key_dir: temp.path().to_path_buf(),
+            master_key: None,
+            file_permissions: Some(0o600),
+        })
+        .await
+        .expect("client should initialize");
+        (client, temp)
+    }
+
+    fn test_kek() -> BackupKek {
+        BackupKek::new("backup-kek-test", 1, [0x42; 32]).expect("kek")
+    }
+
+    fn export_request(destination: PathBuf) -> LocalBackupExportRequest {
+        LocalBackupExportRequest {
+            backup_id: "backup-0001".to_string(),
+            deployment_identity: "deployment-test".to_string(),
+            rustfs_version: "1.0.0-test".to_string(),
+            snapshot_generation: 7,
+            destination,
+        }
+    }
+
+    fn walk_files(dir: &Path, out: &mut Vec<PathBuf>) {
+        for entry in std::fs::read_dir(dir).expect("read dir") {
+            let path = entry.expect("dir entry").path();
+            if path.is_dir() {
+                walk_files(&path, out);
+            } else {
+                out.push(path);
+            }
+        }
+    }
+
+    fn contains_subslice(haystack: &[u8], needle: &[u8]) -> bool {
+        !needle.is_empty() && haystack.windows(needle.len()).any(|window| window == needle)
+    }
+
+    #[tokio::test]
+    async fn export_round_trips_and_decrypts_to_source_records() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("alpha", "AES_256", None).await.expect("create alpha");
+        client.create_key("beta", "AES_256", None).await.expect("create beta");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+
+        assert_eq!(manifest.backend, BackupBackendKind::Local);
+        assert_eq!(manifest.responsibility, BackupResponsibility::FullMaterial);
+        assert_eq!(manifest.at_rest_protection, AtRestProtection::EncryptedMasterKey);
+        assert_eq!(manifest.snapshot_generation, 7);
+        let kdf = manifest.local_kdf.as_ref().expect("local kdf descriptor");
+        assert_eq!(kdf.derivation, LocalKeyDerivation::current_argon2id());
+        assert_eq!(kdf.protection_modes, vec![AtRestProtection::EncryptedMasterKey]);
+
+        // alpha, beta (sorted), then the salt artifact.
+        assert_eq!(manifest.artifacts.len(), 3);
+        assert_eq!(manifest.artifacts[0].path, "artifacts/keys/alpha.key.enc");
+        assert_eq!(manifest.artifacts[1].path, "artifacts/keys/beta.key.enc");
+        assert_eq!(manifest.artifacts[2].kind, ArtifactKind::MasterKeySalt);
+
+        let reread = read_local_bundle_manifest(&destination)
+            .await
+            .expect("manifest should decode");
+        assert_eq!(reread, manifest);
+
+        for (artifact, key_id) in [(&manifest.artifacts[0], "alpha"), (&manifest.artifacts[1], "beta")] {
+            let decrypted = decrypt_bundle_artifact(&destination, &manifest, artifact, &kek)
+                .await
+                .expect("artifact should decrypt");
+            let source = fs::read(client.key_directory().join(format!("{key_id}.key")))
+                .await
+                .expect("source record");
+            assert_eq!(decrypted.as_slice(), source.as_slice(), "record {key_id} must round-trip verbatim");
+        }
+
+        let salt = decrypt_bundle_artifact(&destination, &manifest, &manifest.artifacts[2], &kek)
+            .await
+            .expect("salt should decrypt");
+        let source_salt = fs::read(client.master_key_salt_file()).await.expect("source salt");
+        assert_eq!(salt.as_slice(), source_salt.as_slice());
+    }
+
+    #[tokio::test]
+    async fn plaintext_dev_only_material_is_rewrapped_and_absent_from_bundle() {
+        let (client, _key_dir) = dev_client().await;
+        client.create_key("dev-key", "AES_256", None).await.expect("create key");
+
+        let material = client
+            .decrypt_key_material_for_export("dev-key")
+            .await
+            .expect("material should be readable");
+        let source_record = fs::read(client.key_directory().join("dev-key.key")).await.expect("record");
+        let record_json: serde_json::Value = serde_json::from_slice(&source_record).expect("record parses");
+        let material_base64 = record_json
+            .get("encrypted_key_material")
+            .and_then(|value| value.as_str())
+            .expect("material field")
+            .to_string();
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+
+        assert_eq!(manifest.at_rest_protection, AtRestProtection::PlaintextDevOnly);
+        let kdf = manifest.local_kdf.as_ref().expect("local kdf descriptor");
+        assert_eq!(kdf.protection_modes, vec![AtRestProtection::PlaintextDevOnly]);
+        assert_eq!(kdf.derivation, LocalKeyDerivation::LegacySha256);
+        assert!(
+            !manifest.artifacts.iter().any(|a| a.kind == ArtifactKind::MasterKeySalt),
+            "dev-mode directory has no salt to bundle"
+        );
+
+        // Byte-level: neither the raw material nor its base64 form may appear
+        // anywhere in the bundle. The mandatory KEK re-wrap is what hides it.
+        let mut files = Vec::new();
+        walk_files(&destination, &mut files);
+        assert!(!files.is_empty());
+        for file in files {
+            let bytes = std::fs::read(&file).expect("bundle file");
+            assert!(
+                !contains_subslice(&bytes, material.as_ref()),
+                "raw key material leaked into {}",
+                file.display()
+            );
+            assert!(
+                !contains_subslice(&bytes, material_base64.as_bytes()),
+                "base64 key material leaked into {}",
+                file.display()
+            );
+        }
+
+        // The wrapped record still round-trips for restore.
+        let decrypted = decrypt_bundle_artifact(&destination, &manifest, &manifest.artifacts[0], &kek)
+            .await
+            .expect("artifact should decrypt");
+        assert_eq!(decrypted.as_slice(), source_record.as_slice());
+    }
+
+    #[tokio::test]
+    async fn export_fence_blocks_writers_until_released() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("existing", "AES_256", None).await.expect("create key");
+        let client = Arc::new(client);
+
+        let fence = client.acquire_export_fence().await;
+
+        let writer = {
+            let client = Arc::clone(&client);
+            tokio::spawn(async move {
+                client.create_key("new-key", "AES_256", None).await.expect("create");
+                client.disable_key("existing", None).await.expect("disable");
+            })
+        };
+
+        for _ in 0..64 {
+            tokio::task::yield_now().await;
+        }
+        assert!(!writer.is_finished(), "writers must stay blocked while the export fence is held");
+
+        drop(fence);
+        writer.await.expect("writer should finish after fence release");
+        assert!(
+            fs::try_exists(client.key_directory().join("new-key.key"))
+                .await
+                .expect("exists")
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn concurrent_writers_yield_complete_records() {
+        let (client, _key_dir) = encrypted_client().await;
+        for index in 0..5 {
+            client
+                .create_key(&format!("seed-{index}"), "AES_256", None)
+                .await
+                .expect("seed key");
+        }
+        let client = Arc::new(client);
+
+        let writer = {
+            let client = Arc::clone(&client);
+            tokio::spawn(async move {
+                for index in 0..30 {
+                    client
+                        .create_key(&format!("concurrent-{index}"), "AES_256", None)
+                        .await
+                        .expect("create");
+                    let target = format!("seed-{}", index % 5);
+                    if index % 2 == 0 {
+                        client.disable_key(&target, None).await.expect("disable");
+                    } else {
+                        client.enable_key(&target, None).await.expect("enable");
+                    }
+                }
+            })
+        };
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed under concurrent writers");
+        writer.await.expect("writer task");
+
+        // Whatever subset of writers landed before the fence, every record in
+        // the bundle must be complete: parseable, self-identifying, and with
+        // non-empty material. No torn records, no half-updates.
+        let reread = read_local_bundle_manifest(&destination).await.expect("manifest decodes");
+        assert_eq!(reread, manifest);
+        for artifact in manifest.artifacts.iter().filter(|a| a.kind == ArtifactKind::KeyMaterial) {
+            let record = decrypt_bundle_artifact(&destination, &manifest, artifact, &kek)
+                .await
+                .expect("record decrypts");
+            let value: serde_json::Value = serde_json::from_slice(&record).expect("record is complete JSON");
+            let key_id = value.get("key_id").and_then(|v| v.as_str()).expect("key_id present");
+            assert_eq!(artifact.path, format!("artifacts/keys/{key_id}.key.enc"));
+            let material = value
+                .get("encrypted_key_material")
+                .and_then(|v| v.as_str())
+                .expect("material present");
+            assert!(!material.is_empty());
+        }
+    }
+
+    #[tokio::test]
+    async fn tampered_and_truncated_bundles_fail_closed() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("victim", "AES_256", None).await.expect("create key");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+        let artifact = &manifest.artifacts[0];
+        let artifact_file = destination.join(&artifact.path);
+        let original_artifact = std::fs::read(&artifact_file).expect("artifact bytes");
+
+        // Tampered artifact byte: digest verification rejects it.
+        let mut tampered = original_artifact.clone();
+        let last = tampered.len() - 1;
+        tampered[last] ^= 0x01;
+        std::fs::write(&artifact_file, &tampered).expect("write tampered");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &kek)
+            .await
+            .expect_err("tampered artifact must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Corrupted { .. })), "got {error:?}");
+
+        // Truncated artifact: typed truncation error.
+        std::fs::write(&artifact_file, &original_artifact[..original_artifact.len() - 4]).expect("truncate");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &kek)
+            .await
+            .expect_err("truncated artifact must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Truncated { .. })), "got {error:?}");
+        std::fs::write(&artifact_file, &original_artifact).expect("restore artifact");
+
+        // Tampered manifest (generation flip): sealed digest mismatch.
+        let manifest_file = destination.join(LOCAL_BUNDLE_MANIFEST_FILE);
+        let original_manifest = std::fs::read(&manifest_file).expect("manifest bytes");
+        let tampered_manifest = String::from_utf8(original_manifest.clone())
+            .expect("manifest is utf-8")
+            .replace("\"snapshot_generation\":7", "\"snapshot_generation\":8");
+        assert_ne!(tampered_manifest.as_bytes(), original_manifest.as_slice(), "tamper must apply");
+        std::fs::write(&manifest_file, tampered_manifest).expect("write tampered manifest");
+        let error = read_local_bundle_manifest(&destination)
+            .await
+            .expect_err("tampered manifest must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Corrupted { .. })), "got {error:?}");
+
+        // Truncated manifest.
+        std::fs::write(&manifest_file, &original_manifest[..original_manifest.len() / 2]).expect("truncate manifest");
+        let error = read_local_bundle_manifest(&destination)
+            .await
+            .expect_err("truncated manifest must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Truncated { .. })), "got {error:?}");
+
+        // Missing manifest: the bundle never sealed.
+        std::fs::remove_file(&manifest_file).expect("remove manifest");
+        let error = read_local_bundle_manifest(&destination)
+            .await
+            .expect_err("bundle without manifest must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::IncompleteBundle { .. })), "got {error:?}");
+    }
+
+    #[tokio::test]
+    async fn wrong_kek_is_rejected_before_decryption() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("victim", "AES_256", None).await.expect("create key");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+        let artifact = &manifest.artifacts[0];
+
+        let wrong_id = BackupKek::new("other-kek", 1, [0x42; 32]).expect("kek");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &wrong_id)
+            .await
+            .expect_err("mismatched KEK id must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::WrongKek { .. })), "got {error:?}");
+
+        let wrong_version = BackupKek::new("backup-kek-test", 2, [0x42; 32]).expect("kek");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &wrong_version)
+            .await
+            .expect_err("mismatched KEK version must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::WrongKek { .. })), "got {error:?}");
+
+        // Right identity, wrong material: AEAD authentication fails closed.
+        let wrong_material = BackupKek::new("backup-kek-test", 1, [0x24; 32]).expect("kek");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &wrong_material)
+            .await
+            .expect_err("wrong KEK material must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Corrupted { .. })), "got {error:?}");
+    }
+
+    #[tokio::test]
+    async fn missing_salt_with_encrypted_records_fails_export() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("victim", "AES_256", None).await.expect("create key");
+        fs::remove_file(client.master_key_salt_file()).await.expect("remove salt");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let error = export_local_backup(&client, &test_kek(), &export_request(bundle.path().join("bundle")))
+            .await
+            .expect_err("export without salt must fail");
+        assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}");
+        assert!(error.to_string().contains("salt"), "got {error}");
+    }
+
+    #[tokio::test]
+    async fn refuses_empty_key_dir_and_nonempty_destination() {
+        let (client, _key_dir) = dev_client().await;
+        let bundle = TempDir::new().expect("bundle dir");
+        let error = export_local_backup(&client, &test_kek(), &export_request(bundle.path().join("bundle")))
+            .await
+            .expect_err("empty key dir must not produce a bundle");
+        assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}");
+
+        client.create_key("dev-key", "AES_256", None).await.expect("create key");
+        let occupied = bundle.path().join("occupied");
+        std::fs::create_dir_all(&occupied).expect("mkdir");
+        std::fs::write(occupied.join("stale"), b"leftover").expect("occupy");
+        let error = export_local_backup(&client, &test_kek(), &export_request(occupied))
+            .await
+            .expect_err("non-empty destination must be refused");
+        assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}");
+    }
+
+    #[tokio::test]
+    async fn legacy_records_export_verbatim_with_weakest_protection_label() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("modern", "AES_256", None).await.expect("create key");
+        client.create_key("legacy-key", "AES_256", None).await.expect("create key");
+
+        // Strip the protection marker to fabricate a pre-beta.9 record, the
+        // same way the local backend's own legacy-compat tests do.
+        let legacy_path = client.key_directory().join("legacy-key.key");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&legacy_path).await.expect("record")).expect("record parses");
+        record
+            .as_object_mut()
+            .expect("record is an object")
+            .remove("at_rest_protection");
+        let legacy_bytes = serde_json::to_vec_pretty(&record).expect("record serializes");
+        fs::write(&legacy_path, &legacy_bytes).await.expect("write legacy record");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+
+        assert_eq!(manifest.at_rest_protection, AtRestProtection::LegacyUnspecified);
+        let kdf = manifest.local_kdf.as_ref().expect("local kdf descriptor");
+        assert_eq!(
+            kdf.protection_modes,
+            vec![AtRestProtection::EncryptedMasterKey, AtRestProtection::LegacyUnspecified]
+        );
+
+        let legacy_artifact = manifest
+            .artifacts
+            .iter()
+            .find(|a| a.path == "artifacts/keys/legacy-key.key.enc")
+            .expect("legacy artifact");
+        let decrypted = decrypt_bundle_artifact(&destination, &manifest, legacy_artifact, &kek)
+            .await
+            .expect("legacy artifact decrypts");
+        assert_eq!(decrypted.as_slice(), legacy_bytes.as_slice(), "legacy record must travel verbatim");
+    }
+
+    #[tokio::test]
+    async fn record_identity_mismatch_aborts_export() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("good", "AES_256", None).await.expect("create key");
+        std::fs::copy(client.key_directory().join("good.key"), client.key_directory().join("evil.key"))
+            .expect("plant mismatched record");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let error = export_local_backup(&client, &test_kek(), &export_request(bundle.path().join("bundle")))
+            .await
+            .expect_err("identity mismatch must abort the export");
+        assert!(matches!(error, KmsError::InvalidKey { .. }), "got {error:?}");
+    }
+}

crates/kms/src/backup/mod.rs

@@ -12,13 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Backup/restore contract types for KMS state.
+//! Backup/restore contracts and backup production for KMS state.
 //!
-//! This module is contract-only: it defines the versioned backup manifest,
-//! the per-backend responsibility matrix, typed failure modes, and the
-//! restore dry-run report. Nothing here is wired into handlers or backends;
-//! backup export, restore orchestration, and the admin API build on these
-//! types in follow-up changes.
+//! The contract side defines the versioned backup manifest, the per-backend
+//! responsibility matrix, typed failure modes, and the restore dry-run
+//! report. [`local_export`] implements the producer side for the Local
+//! backend as a crate-internal API; restore orchestration and the admin API
+//! build on these pieces in follow-up changes.
 //!
 //! # Bundle model
 //!
@@ -50,6 +50,7 @@
 mod capability;
 mod dry_run;
 mod error;
+pub mod local_export;
 mod manifest;
 
 pub use capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
@@ -57,6 +58,10 @@ pub use dry_run::{
     ExternalDependencyMismatch, RestoreBlocker, RestoreBlockerCode, RestoreConflict, RestoreConflictKind, RestoreDryRunReport,
 };
 pub use error::BackupError;
+pub use local_export::{
+    BackupKek, LOCAL_BUNDLE_MANIFEST_FILE, LocalBackupExportRequest, decrypt_bundle_artifact, export_local_backup,
+    read_local_bundle_manifest,
+};
 pub use manifest::{
     AeadAlgorithm, ArtifactDescriptor, ArtifactKind, BackupKekDescriptor, BackupManifest, CompletenessState, ContentDigest,
     DigestAlgorithm, LocalKdfDescriptor, LocalKeyDerivation, ReservedSlot,

crates/kms/src/lib.rs

@@ -73,9 +73,6 @@ pub mod deletion_worker;
 mod encryption;
 mod error;
 pub mod manager;
-// The executor is wired into the Vault backends in a follow-up change; until
-// then the module is only exercised by its own tests.
-#[allow(dead_code)]
 mod policy;
 pub mod service;
 pub mod service_manager;

crates/kms/src/manager.rs

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

crates/kms/src/policy.rs

@@ -16,8 +16,8 @@
 //!
 //! Vault-backed operations leave the process boundary, so every call needs a
 //! per-attempt timeout, a total operation deadline, and classification-driven
-//! bounded retries. This module provides the engine only; the Vault backends
-//! wire their call sites through [`execute`] in a follow-up change.
+//! bounded retries. The Vault backends and the credential provider wire every
+//! outbound `vaultrs` call through [`execute`].
 //!
 //! Retry safety is driven by two orthogonal classifications:
 //! - [`OpClass`] states whether replaying the operation is safe at all.
@@ -112,6 +112,32 @@ pub(crate) struct AttemptError {
     pub(crate) error: KmsError,
 }
 
+impl AttemptError {
+    /// A failure that must never be retried, regardless of operation class.
+    pub(crate) fn fatal(error: KmsError) -> Self {
+        Self {
+            class: ErrorClass::Fatal,
+            error,
+        }
+    }
+
+    /// Classify a `vaultrs` failure and map it onto a domain error.
+    ///
+    /// Classification reads the raw error before `map` consumes it, so call
+    /// sites keep their site-specific error mapping (404 to key-not-found and
+    /// so on) without losing the status code the retry decision needs.
+    pub(crate) fn from_vaultrs(
+        error: vaultrs::error::ClientError,
+        map: impl FnOnce(vaultrs::error::ClientError) -> KmsError,
+    ) -> Self {
+        let class = classify_vaultrs(&error);
+        Self {
+            class,
+            error: map(error),
+        }
+    }
+}
+
 /// Budgets applied by [`execute`].
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) struct RetryPolicy {

crates/policy/src/policy/action.rs

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

rustfs/src/admin/handlers/kms.rs

@@ -14,12 +14,13 @@
 
 //! KMS admin handlers for HTTP API
 
-use super::{kms_dynamic, kms_keys, kms_management};
+use super::{kms_dynamic, kms_key_lifecycle, kms_keys, kms_management};
 use crate::admin::router::{AdminOperation, S3Router};
 
 pub fn register_kms_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
     kms_management::register_kms_management_route(r)?;
     kms_dynamic::register_kms_dynamic_route(r)?;
     kms_keys::register_kms_key_route(r)?;
+    kms_key_lifecycle::register_kms_key_lifecycle_route(r)?;
     Ok(())
 }

rustfs/src/admin/handlers/kms_key_lifecycle.rs

@@ -0,0 +1,456 @@
+// 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.
+
+//! KMS key lifecycle admin API handlers: enable, disable and rotate.
+
+use super::kms_keys::extract_query_params;
+use crate::admin::auth::validate_admin_request;
+use crate::admin::router::{AdminOperation, Operation, S3Router};
+use crate::admin::runtime_sources::current_kms_runtime_service_manager;
+use crate::auth::{check_key_valid, get_session_token};
+use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use hyper::{HeaderMap, Method, StatusCode};
+use matchit::Params;
+use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
+use rustfs_kms::{
+    KmsError, KmsManager,
+    types::{DescribeKeyRequest, KeyMetadata},
+};
+use rustfs_policy::policy::action::{Action, KmsAction};
+use s3s::header::CONTENT_TYPE;
+use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+use tracing::{error, info, warn};
+
+const LOG_COMPONENT_ADMIN: &str = "admin";
+const LOG_SUBSYSTEM_KMS_KEYS: &str = "kms_key_lifecycle";
+const EVENT_ADMIN_KMS_KEYS_STATE: &str = "admin_kms_keys_state";
+
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct KmsKeyLifecycleRequest {
+    pub key_id: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct KmsKeyLifecycleResponse {
+    pub success: bool,
+    pub message: String,
+    pub key_id: String,
+    pub key_metadata: Option<KeyMetadata>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum LifecycleOperation {
+    Enable,
+    Disable,
+    Rotate,
+}
+
+impl LifecycleOperation {
+    fn action(self) -> &'static str {
+        match self {
+            Self::Enable => "enable_key",
+            Self::Disable => "disable_key",
+            Self::Rotate => "rotate_key",
+        }
+    }
+
+    fn verb(self) -> &'static str {
+        match self {
+            Self::Enable => "enable",
+            Self::Disable => "disable",
+            Self::Rotate => "rotate",
+        }
+    }
+
+    fn past_tense(self) -> &'static str {
+        match self {
+            Self::Enable => "enabled",
+            Self::Disable => "disabled",
+            Self::Rotate => "rotated",
+        }
+    }
+}
+
+fn kms_enable_key_actions() -> Vec<Action> {
+    vec![Action::KmsAction(KmsAction::EnableKeyAction)]
+}
+
+fn kms_disable_key_actions() -> Vec<Action> {
+    vec![Action::KmsAction(KmsAction::DisableKeyAction)]
+}
+
+fn kms_rotate_key_actions() -> Vec<Action> {
+    vec![Action::KmsAction(KmsAction::RotateKeyAction)]
+}
+
+pub fn register_kms_key_lifecycle_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
+    r.insert(
+        Method::POST,
+        format!("{}{}", ADMIN_PREFIX, "/v3/kms/keys/enable").as_str(),
+        AdminOperation(&EnableKmsKeyHandler {}),
+    )?;
+
+    r.insert(
+        Method::POST,
+        format!("{}{}", ADMIN_PREFIX, "/v3/kms/keys/disable").as_str(),
+        AdminOperation(&DisableKmsKeyHandler {}),
+    )?;
+
+    r.insert(
+        Method::POST,
+        format!("{}{}", ADMIN_PREFIX, "/v3/kms/keys/rotate").as_str(),
+        AdminOperation(&RotateKmsKeyHandler {}),
+    )?;
+
+    Ok(())
+}
+
+/// Map a lifecycle failure to an HTTP status.
+///
+/// `UnsupportedCapability` is a permanent gap of the configured backend, not a
+/// missing resource: it must surface as 501, never 404. State-machine
+/// rejections (`InvalidOperation`) keep the 400 mapping used by the other
+/// `/v3/kms/keys` handlers.
+fn lifecycle_error_status(error: &KmsError) -> StatusCode {
+    match error {
+        KmsError::KeyNotFound { .. } => StatusCode::NOT_FOUND,
+        KmsError::UnsupportedCapability { .. } => StatusCode::NOT_IMPLEMENTED,
+        KmsError::InvalidOperation { .. } | KmsError::ValidationError { .. } => StatusCode::BAD_REQUEST,
+        // Damaged or missing key material is an integrity fault of an existing
+        // key: it must surface as a server error, never as NOT_FOUND (the key
+        // exists) and never as a retryable backend outage.
+        KmsError::MaterialMissing { .. }
+        | KmsError::MaterialCorrupt { .. }
+        | KmsError::MaterialAuthenticationFailed { .. }
+        | KmsError::UnsupportedFormatVersion { .. } => StatusCode::INTERNAL_SERVER_ERROR,
+        _ => StatusCode::INTERNAL_SERVER_ERROR,
+    }
+}
+
+/// Run one lifecycle operation against the manager and build the HTTP reply.
+/// Split from the handlers so the status/response contract is unit-testable
+/// without the admin auth stack.
+async fn execute_lifecycle(
+    manager: &KmsManager,
+    key_id: &str,
+    operation: LifecycleOperation,
+) -> (StatusCode, KmsKeyLifecycleResponse) {
+    let result = match operation {
+        LifecycleOperation::Enable => manager.enable_key(key_id).await,
+        LifecycleOperation::Disable => manager.disable_key(key_id).await,
+        LifecycleOperation::Rotate => manager.rotate_key(key_id).await,
+    };
+
+    match result {
+        Ok(()) => {
+            info!(
+                event = EVENT_ADMIN_KMS_KEYS_STATE,
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_KMS_KEYS,
+                action = operation.action(),
+                key_id = %key_id,
+                state = "completed",
+                "admin kms keys state"
+            );
+            // Best effort: the state change already succeeded, so a failing
+            // describe only drops the metadata echo from the response.
+            let key_metadata = match manager
+                .describe_key(DescribeKeyRequest {
+                    key_id: key_id.to_string(),
+                })
+                .await
+            {
+                Ok(response) => Some(response.key_metadata),
+                Err(error) => {
+                    warn!(
+                        event = EVENT_ADMIN_KMS_KEYS_STATE,
+                        component = LOG_COMPONENT_ADMIN,
+                        subsystem = LOG_SUBSYSTEM_KMS_KEYS,
+                        action = operation.action(),
+                        key_id = %key_id,
+                        error = %error,
+                        "key lifecycle change succeeded but describe failed"
+                    );
+                    None
+                }
+            };
+            (
+                StatusCode::OK,
+                KmsKeyLifecycleResponse {
+                    success: true,
+                    message: format!("key {} successfully", operation.past_tense()),
+                    key_id: key_id.to_string(),
+                    key_metadata,
+                },
+            )
+        }
+        Err(error) => {
+            error!(
+                event = EVENT_ADMIN_KMS_KEYS_STATE,
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_KMS_KEYS,
+                action = operation.action(),
+                key_id = %key_id,
+                result = "failed",
+                error = %error,
+                "admin kms keys state"
+            );
+            (
+                lifecycle_error_status(&error),
+                KmsKeyLifecycleResponse {
+                    success: false,
+                    message: format!("Failed to {} key: {error}", operation.verb()),
+                    key_id: key_id.to_string(),
+                    key_metadata: None,
+                },
+            )
+        }
+    }
+}
+
+fn json_response(status: StatusCode, response: &KmsKeyLifecycleResponse) -> S3Result<S3Response<(StatusCode, Body)>> {
+    let data = serde_json::to_vec(response).map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?;
+    let mut headers = HeaderMap::new();
+    headers.insert(CONTENT_TYPE, "application/json".parse().expect("static content type should parse"));
+    Ok(S3Response::with_headers((status, Body::from(data)), headers))
+}
+
+fn unavailable_response(message: &str, key_id: String) -> S3Result<S3Response<(StatusCode, Body)>> {
+    json_response(
+        StatusCode::SERVICE_UNAVAILABLE,
+        &KmsKeyLifecycleResponse {
+            success: false,
+            message: message.to_string(),
+            key_id,
+            key_metadata: None,
+        },
+    )
+}
+
+async fn handle_lifecycle_request(
+    mut req: S3Request<Body>,
+    actions: Vec<Action>,
+    operation: LifecycleOperation,
+) -> S3Result<S3Response<(StatusCode, Body)>> {
+    let Some(cred) = req.credentials else {
+        return Err(s3_error!(InvalidRequest, "authentication required"));
+    };
+
+    let (cred, owner) = check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
+
+    validate_admin_request(
+        &req.headers,
+        &cred,
+        owner,
+        false,
+        actions,
+        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+    )
+    .await?;
+
+    let body = req
+        .input
+        .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
+        .await
+        .map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))?;
+
+    let request: KmsKeyLifecycleRequest = if body.is_empty() {
+        let query_params = extract_query_params(&req.uri);
+        let Some(key_id) = query_params.get("keyId") else {
+            return json_response(
+                StatusCode::BAD_REQUEST,
+                &KmsKeyLifecycleResponse {
+                    success: false,
+                    message: "missing required parameter: 'keyId'".to_string(),
+                    key_id: "".to_string(),
+                    key_metadata: None,
+                },
+            );
+        };
+        KmsKeyLifecycleRequest { key_id: key_id.clone() }
+    } else {
+        serde_json::from_slice(&body).map_err(|e| s3_error!(InvalidRequest, "invalid JSON: {}", e))?
+    };
+
+    let Some(service_manager) = kms_service_manager_from_context() else {
+        return unavailable_response("kms service manager is not initialized", request.key_id);
+    };
+
+    let Some(manager) = service_manager.get_manager().await else {
+        return unavailable_response("kms service is not running", request.key_id);
+    };
+
+    let (status, response) = execute_lifecycle(&manager, &request.key_id, operation).await;
+    json_response(status, &response)
+}
+
+fn kms_service_manager_from_context() -> Option<Arc<rustfs_kms::KmsServiceManager>> {
+    current_kms_runtime_service_manager()
+}
+
+/// Enable a disabled KMS key
+pub struct EnableKmsKeyHandler;
+
+#[async_trait::async_trait]
+impl Operation for EnableKmsKeyHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        handle_lifecycle_request(req, kms_enable_key_actions(), LifecycleOperation::Enable).await
+    }
+}
+
+/// Disable a KMS key; existing data remains decryptable
+pub struct DisableKmsKeyHandler;
+
+#[async_trait::async_trait]
+impl Operation for DisableKmsKeyHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        handle_lifecycle_request(req, kms_disable_key_actions(), LifecycleOperation::Disable).await
+    }
+}
+
+/// Rotate a KMS key to a new version
+pub struct RotateKmsKeyHandler;
+
+#[async_trait::async_trait]
+impl Operation for RotateKmsKeyHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        handle_lifecycle_request(req, kms_rotate_key_actions(), LifecycleOperation::Rotate).await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rustfs_kms::backends::local::LocalKmsBackend;
+    use rustfs_kms::config::KmsConfig;
+    use rustfs_kms::types::{CreateKeyRequest, KeyState};
+    use rustfs_policy::policy::action::AdminAction;
+
+    async fn local_manager(temp_dir: &tempfile::TempDir) -> KmsManager {
+        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+        let backend = Arc::new(
+            LocalKmsBackend::new(config.clone())
+                .await
+                .expect("local backend should build"),
+        );
+        KmsManager::new(backend, config)
+    }
+
+    async fn create_key(manager: &KmsManager, key_name: &str) -> String {
+        manager
+            .create_key(CreateKeyRequest {
+                key_name: Some(key_name.to_string()),
+                ..Default::default()
+            })
+            .await
+            .expect("key should be created")
+            .key_id
+    }
+
+    fn key_state(response: &KmsKeyLifecycleResponse) -> KeyState {
+        response
+            .key_metadata
+            .as_ref()
+            .expect("lifecycle response should echo key metadata")
+            .key_state
+            .clone()
+    }
+
+    #[tokio::test]
+    async fn enable_disable_enable_round_trip() {
+        let temp_dir = tempfile::tempdir().expect("temp dir");
+        let manager = local_manager(&temp_dir).await;
+        let key_id = create_key(&manager, "lifecycle-round-trip").await;
+
+        let (status, response) = execute_lifecycle(&manager, &key_id, LifecycleOperation::Disable).await;
+        assert_eq!(status, StatusCode::OK);
+        assert!(response.success);
+        assert_eq!(key_state(&response), KeyState::Disabled);
+
+        let (status, response) = execute_lifecycle(&manager, &key_id, LifecycleOperation::Enable).await;
+        assert_eq!(status, StatusCode::OK);
+        assert!(response.success);
+        assert_eq!(key_state(&response), KeyState::Enabled);
+
+        // Enabling an already enabled key stays idempotent.
+        let (status, response) = execute_lifecycle(&manager, &key_id, LifecycleOperation::Enable).await;
+        assert_eq!(status, StatusCode::OK);
+        assert_eq!(key_state(&response), KeyState::Enabled);
+    }
+
+    #[tokio::test]
+    async fn local_rotate_is_unsupported_capability_not_missing_key() {
+        let temp_dir = tempfile::tempdir().expect("temp dir");
+        let manager = local_manager(&temp_dir).await;
+        let key_id = create_key(&manager, "rotate-unsupported").await;
+
+        let (status, response) = execute_lifecycle(&manager, &key_id, LifecycleOperation::Rotate).await;
+        assert_eq!(status, StatusCode::NOT_IMPLEMENTED);
+        assert_ne!(status, StatusCode::NOT_FOUND);
+        assert!(!response.success);
+        assert!(
+            response.message.contains("not supported"),
+            "message should name the capability gap: {}",
+            response.message
+        );
+
+        // A disabled key must not change the picture: rotation stays rejected.
+        let (status, _) = execute_lifecycle(&manager, &key_id, LifecycleOperation::Disable).await;
+        assert_eq!(status, StatusCode::OK);
+        let (status, response) = execute_lifecycle(&manager, &key_id, LifecycleOperation::Rotate).await;
+        assert_ne!(status, StatusCode::OK);
+        assert!(!response.success);
+    }
+
+    #[tokio::test]
+    async fn lifecycle_on_missing_key_maps_to_not_found() {
+        let temp_dir = tempfile::tempdir().expect("temp dir");
+        let manager = local_manager(&temp_dir).await;
+
+        let (status, response) = execute_lifecycle(&manager, "no-such-key", LifecycleOperation::Enable).await;
+        assert_eq!(status, StatusCode::NOT_FOUND);
+        assert!(!response.success);
+    }
+
+    #[test]
+    fn lifecycle_error_status_contract() {
+        let unsupported = KmsError::unsupported_capability("local", "rotate_key");
+        assert_eq!(lifecycle_error_status(&unsupported), StatusCode::NOT_IMPLEMENTED);
+        assert_ne!(lifecycle_error_status(&unsupported), StatusCode::NOT_FOUND);
+
+        assert_eq!(lifecycle_error_status(&KmsError::invalid_key_state("disabled")), StatusCode::BAD_REQUEST);
+        assert_eq!(lifecycle_error_status(&KmsError::key_not_found("gone")), StatusCode::NOT_FOUND);
+    }
+
+    #[test]
+    fn lifecycle_auth_actions_use_dedicated_kms_actions() {
+        assert_eq!(kms_enable_key_actions(), vec![Action::KmsAction(KmsAction::EnableKeyAction)]);
+        assert_eq!(kms_disable_key_actions(), vec![Action::KmsAction(KmsAction::DisableKeyAction)]);
+        assert_eq!(kms_rotate_key_actions(), vec![Action::KmsAction(KmsAction::RotateKeyAction)]);
+
+        for actions in [kms_enable_key_actions(), kms_disable_key_actions(), kms_rotate_key_actions()] {
+            assert!(!actions.contains(&Action::AdminAction(AdminAction::ServerInfoAdminAction)));
+        }
+    }
+
+    #[test]
+    fn lifecycle_request_rejects_unknown_fields() {
+        let error = serde_json::from_str::<KmsKeyLifecycleRequest>(r#"{"key_id":"key","unexpected_field":true}"#)
+            .expect_err("unknown lifecycle request field should fail");
+        assert!(error.to_string().contains("unknown field"));
+    }
+}

rustfs/src/admin/handlers/kms_keys.rs

@@ -94,7 +94,7 @@ pub struct GenerateDataKeyApiResponse {
     pub ciphertext_blob: String, // Base64 encoded
 }
 
-fn extract_query_params(uri: &hyper::Uri) -> HashMap<String, String> {
+pub(super) fn extract_query_params(uri: &hyper::Uri) -> HashMap<String, String> {
     let mut params = HashMap::new();
     if let Some(query) = uri.query() {
         query.split('&').for_each(|pair| {

rustfs/src/admin/handlers/mod.rs

@@ -33,6 +33,7 @@ pub mod inspect_archive;
 pub mod is_admin;
 pub mod kms;
 pub mod kms_dynamic;
+pub mod kms_key_lifecycle;
 pub mod kms_keys;
 pub mod kms_management;
 pub mod metrics;

rustfs/src/admin/route_policy.rs

@@ -58,8 +58,11 @@ const KMS_CLEAR_CACHE: AdminActionRef = AdminActionRef::new("kms:ClearCache");
 const KMS_CONFIGURE: AdminActionRef = AdminActionRef::new("kms:Configure");
 const KMS_DELETE_KEY: AdminActionRef = AdminActionRef::new("kms:DeleteKey");
 const KMS_DESCRIBE_KEY: AdminActionRef = AdminActionRef::new("kms:DescribeKey");
+const KMS_DISABLE_KEY: AdminActionRef = AdminActionRef::new("kms:DisableKey");
+const KMS_ENABLE_KEY: AdminActionRef = AdminActionRef::new("kms:EnableKey");
 const KMS_GENERATE_DATA_KEY: AdminActionRef = AdminActionRef::new("kms:GenerateDataKey");
 const KMS_LIST_KEYS: AdminActionRef = AdminActionRef::new("kms:ListKeys");
+const KMS_ROTATE_KEY: AdminActionRef = AdminActionRef::new("kms:RotateKey");
 const KMS_SERVICE_CONTROL: AdminActionRef = AdminActionRef::new("kms:ServiceControl");
 const LIST_GROUPS: AdminActionRef = AdminActionRef::new("ListGroupsAdminAction");
 const LIST_TEMPORARY_ACCOUNTS: AdminActionRef = AdminActionRef::new("ListTemporaryAccountsAdminAction");
@@ -777,6 +780,14 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         KMS_DESCRIBE_KEY,
         RouteRiskLevel::Sensitive,
     ),
+    admin(HttpMethod::Post, "/rustfs/admin/v3/kms/keys/enable", KMS_ENABLE_KEY, RouteRiskLevel::High),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/kms/keys/disable",
+        KMS_DISABLE_KEY,
+        RouteRiskLevel::High,
+    ),
+    admin(HttpMethod::Post, "/rustfs/admin/v3/kms/keys/rotate", KMS_ROTATE_KEY, RouteRiskLevel::High),
     public(
         HttpMethod::Get,
         "/rustfs/admin/v3/oidc/providers",

rustfs/src/admin/route_registration_test.rs

@@ -340,6 +340,9 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
         admin_route(Method::POST, "/v3/kms/keys/cancel-deletion"),
         admin_route(Method::GET, "/v3/kms/keys"),
         admin_route_sample(Method::GET, "/v3/kms/keys/{key_id}", "/v3/kms/keys/test-key"),
+        admin_route(Method::POST, "/v3/kms/keys/enable"),
+        admin_route(Method::POST, "/v3/kms/keys/disable"),
+        admin_route(Method::POST, "/v3/kms/keys/rotate"),
         admin_route(Method::GET, "/v3/oidc/providers"),
         admin_route_sample(Method::GET, "/v3/oidc/authorize/{provider_id}", "/v3/oidc/authorize/default"),
         admin_route_sample(Method::GET, "/v3/oidc/callback/{provider_id}", "/v3/oidc/callback/default"),

rustfs/src/init.rs

@@ -468,6 +468,13 @@ pub async fn init_kms_system(config: &config::Config) -> std::io::Result<()> {
     // Initialize global KMS service manager (starts in NotConfigured state)
     let service_manager = startup_runtime_sources::init_kms_service_manager();
 
+    // A key referenced by any bucket's encryption configuration must never be
+    // deleted. Register the gate before the service can start so every
+    // deletion-worker spawn observes it; the gate fails closed while the
+    // object store is not ready.
+    service_manager
+        .set_deletion_reference_checker(std::sync::Arc::new(crate::kms_deletion_gate::BucketEncryptionReferenceChecker));
+
     // If KMS is enabled in configuration, configure and start the service
     if config.kms_enable {
         info!(

rustfs/src/kms_deletion_gate.rs

@@ -0,0 +1,147 @@
+// 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.
+
+//! Reference gate consulted before a scheduled KMS key deletion destroys
+//! material: a key that any bucket's encryption configuration still points at
+//! must not be removed. Object-level references (existing envelopes) are out
+//! of scope here; those objects stay decryptable only until the key is gone,
+//! which is why the pending-deletion window exists.
+
+use crate::runtime_sources::current_object_store_handle;
+use crate::storage_api::kms::contract::bucket::{BucketOperations, BucketOptions};
+use crate::storage_api::kms::{ECStore, StorageError, get_bucket_sse_config};
+use async_trait::async_trait;
+use rustfs_kms::DeletionReferenceChecker;
+use s3s::dto::ServerSideEncryptionConfiguration;
+use std::sync::Arc;
+use tracing::warn;
+
+/// Reference reported when bucket configuration cannot be inspected at all.
+const BUCKET_CONFIG_UNAVAILABLE: &str = "bucket-encryption-config:unavailable";
+
+/// Blocks deletion of keys referenced by any bucket's SSE configuration
+/// (default bucket KMS key). Registered on the KMS service manager at startup
+/// and consulted by the background deletion worker.
+pub(crate) struct BucketEncryptionReferenceChecker;
+
+#[async_trait]
+impl DeletionReferenceChecker for BucketEncryptionReferenceChecker {
+    async fn references(&self, key_id: &str) -> Vec<String> {
+        collect_references(key_id, current_object_store_handle()).await
+    }
+}
+
+async fn collect_references(key_id: &str, store: Option<Arc<ECStore>>) -> Vec<String> {
+    // Fail closed: destroying key material is irreversible while the deletion
+    // worker retries every sweep, so an uninspectable configuration must block
+    // the removal rather than wave it through. This also covers the worker
+    // racing server startup, before the object store is published.
+    let Some(store) = store else {
+        warn!(key_id, "KMS deletion reference check: object store not ready; blocking removal");
+        return vec![BUCKET_CONFIG_UNAVAILABLE.to_string()];
+    };
+    let buckets = match store.list_bucket(&BucketOptions::default()).await {
+        Ok(buckets) => buckets,
+        Err(error) => {
+            warn!(key_id, %error, "KMS deletion reference check: listing buckets failed; blocking removal");
+            return vec![BUCKET_CONFIG_UNAVAILABLE.to_string()];
+        }
+    };
+
+    let mut references = Vec::new();
+    for bucket in &buckets {
+        let lookup = get_bucket_sse_config(&bucket.name).await;
+        references.extend(bucket_reference(&bucket.name, lookup, key_id));
+    }
+    references
+}
+
+/// `Some(reference)` when the bucket's encryption configuration references
+/// `key_id` or cannot be read (fail closed); `None` when the bucket provably
+/// does not reference the key.
+fn bucket_reference(
+    bucket: &str,
+    lookup: Result<ServerSideEncryptionConfiguration, StorageError>,
+    key_id: &str,
+) -> Option<String> {
+    match lookup {
+        Ok(config) if sse_config_references_key(&config, key_id) => Some(format!("bucket:{bucket}")),
+        Ok(_) => None,
+        Err(StorageError::ConfigNotFound) => None,
+        Err(error) => {
+            warn!(
+                bucket,
+                key_id,
+                %error,
+                "KMS deletion reference check: unreadable bucket encryption config; blocking removal"
+            );
+            Some(format!("bucket:{bucket}:encryption-config-unreadable"))
+        }
+    }
+}
+
+fn sse_config_references_key(config: &ServerSideEncryptionConfiguration, key_id: &str) -> bool {
+    config.rules.iter().any(|rule| {
+        rule.apply_server_side_encryption_by_default
+            .as_ref()
+            .and_then(|sse| sse.kms_master_key_id.as_deref())
+            .is_some_and(|configured| configured == key_id)
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use s3s::dto::{ServerSideEncryption, ServerSideEncryptionByDefault, ServerSideEncryptionRule};
+
+    fn sse_kms_config(key_id: Option<&str>) -> ServerSideEncryptionConfiguration {
+        ServerSideEncryptionConfiguration {
+            rules: vec![ServerSideEncryptionRule {
+                apply_server_side_encryption_by_default: Some(ServerSideEncryptionByDefault {
+                    sse_algorithm: ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS),
+                    kms_master_key_id: key_id.map(str::to_string),
+                }),
+                bucket_key_enabled: None,
+            }],
+        }
+    }
+
+    #[test]
+    fn referencing_bucket_blocks_deletion() {
+        assert_eq!(
+            bucket_reference("sse-bucket", Ok(sse_kms_config(Some("kms-key-1"))), "kms-key-1"),
+            Some("bucket:sse-bucket".to_string())
+        );
+    }
+
+    #[test]
+    fn non_referencing_buckets_allow_deletion() {
+        // Different key, no configured key, and no SSE configuration at all.
+        assert_eq!(bucket_reference("other-key", Ok(sse_kms_config(Some("kms-key-2"))), "kms-key-1"), None);
+        assert_eq!(bucket_reference("no-key", Ok(sse_kms_config(None)), "kms-key-1"), None);
+        assert_eq!(bucket_reference("plain", Err(StorageError::ConfigNotFound), "kms-key-1"), None);
+    }
+
+    #[test]
+    fn unreadable_config_blocks_deletion() {
+        let reference = bucket_reference("broken", Err(StorageError::FaultyDisk), "kms-key-1");
+        assert_eq!(reference, Some("bucket:broken:encryption-config-unreadable".to_string()));
+    }
+
+    #[tokio::test]
+    async fn missing_object_store_blocks_deletion() {
+        let references = collect_references("kms-key-1", None).await;
+        assert_eq!(references, vec![BUCKET_CONFIG_UNAVAILABLE.to_string()]);
+    }
+}

rustfs/src/lib.rs

@@ -82,6 +82,7 @@ pub mod diagnose;
 pub mod embedded;
 pub mod error;
 pub mod init;
+pub(crate) mod kms_deletion_gate;
 pub mod license;
 pub mod memory_observability;
 pub mod profiling;

rustfs/src/storage_api.rs

@@ -72,6 +72,24 @@ pub(crate) mod error {
     pub(crate) use crate::storage::storage_api::{QuotaError, StorageError};
 }
 
+pub(crate) mod kms {
+    pub(crate) mod contract {
+        pub(crate) mod bucket {
+            pub(crate) use super::super::super::storage_contracts::{BucketOperations, BucketOptions};
+        }
+    }
+
+    pub(crate) use crate::storage::storage_api::{ECStore, StorageError};
+
+    /// Bucket SSE configuration for the KMS deletion reference gate;
+    /// `Err(StorageError::ConfigNotFound)` when the bucket has none.
+    pub(crate) async fn get_bucket_sse_config(bucket: &str) -> Result<s3s::dto::ServerSideEncryptionConfiguration, StorageError> {
+        crate::storage::storage_api::ecstore_bucket::metadata_sys::get_sse_config(bucket)
+            .await
+            .map(|(config, _)| config)
+    }
+}
+
 pub(crate) mod protocols {
     pub(crate) mod client {
         pub(crate) use crate::storage::storage_api::access_consumer::ReqInfo;