feat(kms): introduce KMS unavailability error and enhance data key handling (#5184)

Cargo.lock

@@ -9459,6 +9459,7 @@ dependencies = [
  "rustfs-utils",
  "serde",
  "serde_json",
+ "sha2 0.11.0",
  "temp-env",
  "tempfile",
  "thiserror 2.0.19",

crates/ecstore/src/object_api/readers.rs

@@ -25,7 +25,7 @@ use chacha20poly1305::ChaCha20Poly1305;
 #[cfg(feature = "rio-v2")]
 use hmac::{Hmac, Mac};
 use md5::{Digest, Md5};
-use rustfs_kms::types::ObjectEncryptionContext;
+use rustfs_kms::{KmsUnavailableError, is_data_key_envelope, types::ObjectEncryptionContext};
 use rustfs_utils::http::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
 use rustfs_utils::path::path_join_buf;
 #[cfg(feature = "rio-v2")]
@@ -1604,7 +1604,13 @@ async fn resolve_managed_material(bucket: &str, object: &str, metadata: &HashMap
     let kms_context: Option<HashMap<String, String>> = None;
     let object_context = build_object_encryption_context(bucket, object, kms_context.as_ref());
 
-    let decrypted_key = if let Some(service) = crate::runtime::sources::object_encryption_service().await {
+    // Persisted wrapping format is the read-side source of truth. The
+    // advertised SSE scheme and current KMS availability are write policy
+    // and runtime state, neither of which identifies the historical provider.
+    let decrypted_key = if is_data_key_envelope(&encrypted_dek) {
+        let service = crate::runtime::sources::object_encryption_service()
+            .await
+            .ok_or_else(|| Error::other(KmsUnavailableError))?;
         #[cfg(feature = "rio-v2")]
         let data_key = if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
             service.decrypt_legacy_data_key(&encrypted_dek).await
@@ -2412,6 +2418,70 @@ mod tests {
         .await;
     }
 
+    #[tokio::test]
+    async fn resolve_managed_material_selects_provider_from_persisted_dek() {
+        use rustfs_kms::KmsConfig;
+        use tempfile::TempDir;
+
+        let key_dir = TempDir::new().expect("create KMS key directory");
+        let manager = rustfs_kms::init_global_kms_service_manager();
+        manager
+            .reconfigure(KmsConfig::local(key_dir.path().to_path_buf()).with_insecure_development_defaults())
+            .await
+            .expect("start test KMS service");
+
+        async_with_vars([("__RUSTFS_SSE_SIMPLE_CMK", Some(BASE64_STANDARD.encode([7u8; 32])))], async {
+            let data_key = [0x24; 32];
+            let base_nonce = [0x14; 12];
+            let encrypted_dek = encrypt_managed_dek_for_test(data_key, [7u8; 32]);
+            let metadata = HashMap::from([
+                (
+                    INTERNAL_ENCRYPTION_KEY_HEADER.to_string(),
+                    BASE64_STANDARD.encode(encrypted_dek.as_bytes()),
+                ),
+                (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), BASE64_STANDARD.encode(base_nonce)),
+                (INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "legacy-local-key".to_string()),
+            ]);
+
+            let material = resolve_managed_material("bucket", "object", &metadata)
+                .await
+                .expect("legacy local DEK should not be routed to the running KMS");
+            assert_eq!(material.key_bytes, data_key);
+            assert_eq!(material.base_nonce, base_nonce);
+        })
+        .await;
+
+        manager.stop().await.expect("stop test KMS service");
+
+        let kms_envelope = br#"{
+            "key_id": "test-key-id",
+            "master_key_id": "master-key-id",
+            "key_spec": "AES_256",
+            "encrypted_key": [1, 2, 3, 4],
+            "nonce": [5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16],
+            "encryption_context": {},
+            "created_at": "2024-01-01T00:00:00+00:00"
+        }"#;
+        let metadata = HashMap::from([
+            (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), BASE64_STANDARD.encode(kms_envelope)),
+            (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), BASE64_STANDARD.encode([0x14; 12])),
+            (INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "test-key-id".to_string()),
+        ]);
+        let error = match resolve_managed_material("bucket", "object", &metadata).await {
+            Ok(_) => panic!("KMS envelope must not fall back to the local provider"),
+            Err(error) => error,
+        };
+        let Error::Io(io_error) = error else {
+            panic!("KMS absence should retain its typed source");
+        };
+        assert!(
+            io_error
+                .get_ref()
+                .and_then(|source| source.downcast_ref::<KmsUnavailableError>())
+                .is_some()
+        );
+    }
+
     #[cfg(feature = "rio-v2")]
     #[tokio::test]
     async fn resolve_managed_material_accepts_chacha20_poly1305_header_variant() {

crates/kms/Cargo.toml

@@ -44,6 +44,7 @@ argon2 = { workspace = true }
 chacha20poly1305 = { workspace = true }
 rand = { workspace = true, features = ["serde"] }
 base64 = { workspace = true }
+sha2 = { workspace = true }
 zeroize = { workspace = true, features = ["derive"] }
 
 # Configuration and storage

crates/kms/src/backends/local.rs

@@ -30,6 +30,7 @@ use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
 use jiff::Zoned;
 use rand::RngExt;
 use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
 use std::collections::HashMap;
 use std::path::PathBuf;
 use std::time::Duration;
@@ -51,6 +52,8 @@ pub struct LocalKmsClient {
     key_cache: RwLock<HashMap<String, MasterKeyInfo>>,
     /// Master encryption key for encrypting stored keys
     master_cipher: Option<Aes256Gcm>,
+    /// Legacy pre-beta.9 master cipher for reading pre-Argon2 key files
+    legacy_master_cipher: Option<Aes256Gcm>,
     /// DEK encryption implementation
     dek_crypto: AesDekCrypto,
 }
@@ -97,19 +100,21 @@ impl LocalKmsClient {
         }
 
         // Initialize master cipher if master key is provided
-        let master_cipher = if let Some(ref master_key) = config.master_key {
+        let (master_cipher, legacy_master_cipher) = if let Some(ref master_key) = config.master_key {
             let salt = Self::load_or_create_master_key_salt(&config).await?;
             let key = Self::derive_master_key(master_key, &salt)?;
-            Some(Aes256Gcm::new(&key))
+            let legacy_key = Self::derive_legacy_master_key(master_key)?;
+            (Some(Aes256Gcm::new(&key)), Some(Aes256Gcm::new(&legacy_key)))
         } else {
             warn!("No master key provided - local KMS key material will use explicit plaintext-dev-only storage");
-            None
+            (None, None)
         };
 
         Ok(Self {
             config,
             key_cache: RwLock::new(HashMap::new()),
             master_cipher,
+            legacy_master_cipher,
             dek_crypto: AesDekCrypto::new(),
         })
     }
@@ -132,6 +137,14 @@ impl LocalKmsClient {
         Ok(key)
     }
 
+    fn derive_legacy_master_key(master_key: &str) -> Result<Key<Aes256Gcm>> {
+        let mut hasher = Sha256::new();
+        hasher.update(master_key.as_bytes());
+        hasher.update(b"rustfs-kms-local");
+        Key::<Aes256Gcm>::try_from(hasher.finalize().as_slice())
+            .map_err(|_| KmsError::cryptographic_error("legacy_key", "Invalid key length"))
+    }
+
     fn master_key_salt_path(config: &LocalConfig) -> PathBuf {
         config.key_dir.join(LOCAL_KMS_MASTER_KEY_SALT_FILE)
     }
@@ -206,6 +219,9 @@ impl LocalKmsClient {
         // Decrypt key material if master cipher is available.
         let key_material = match effective_protection {
             StoredKeyProtection::EncryptedMasterKey => {
+                // RUSTFS_COMPAT_TODO(rustfs-5063): Remove after upgrades rewrite all pre-beta.9 key files.
+                // Pre-beta.9 files have no protection marker and use the legacy
+                // SHA-256 KDF, while later pre-marker files use Argon2.
                 let cipher = self.master_cipher.as_ref().ok_or_else(|| {
                     KmsError::configuration_error(format!(
                         "Local KMS key {key_id} is encrypted at rest and requires a configured master key"
@@ -219,9 +235,20 @@ impl LocalKmsClient {
                 nonce_array.copy_from_slice(&stored_key.nonce);
                 let nonce = Nonce::from(nonce_array);
 
-                cipher
+                match cipher.decrypt(&nonce, encrypted_bytes.as_ref()) {
+                    Ok(key_material) => key_material,
+                    Err(current_error) if stored_key.at_rest_protection == StoredKeyProtection::LegacyUnspecified => {
+                        let legacy_cipher = self.legacy_master_cipher.as_ref().ok_or_else(|| {
+                            KmsError::configuration_error(format!(
+                                "Local KMS key {key_id} is encrypted at rest and requires a configured master key"
+                            ))
+                        })?;
+                        legacy_cipher
                             .decrypt(&nonce, encrypted_bytes.as_ref())
-                    .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))?
+                            .map_err(|_| KmsError::cryptographic_error("decrypt", current_error.to_string()))?
+                    }
+                    Err(error) => return Err(KmsError::cryptographic_error("decrypt", error.to_string())),
+                }
             }
             StoredKeyProtection::PlaintextDevOnly | StoredKeyProtection::LegacyUnspecified => {
                 if self.master_cipher.is_some() && stored_key.at_rest_protection == StoredKeyProtection::PlaintextDevOnly {
@@ -1226,4 +1253,71 @@ mod tests {
             .expect("legacy encrypted record should remain readable");
         assert_eq!(key_info.key_id, "legacy-encrypted-key");
     }
+
+    #[tokio::test]
+    async fn test_load_master_key_accepts_beta5_sha256_encrypted_record() {
+        let temp_dir = TempDir::new().expect("create beta.5 fixture directory");
+        let stored_key = serde_json::json!({
+            "key_id": "beta5-key",
+            "version": 1u32,
+            "algorithm": "AES_256",
+            "usage": "EncryptDecrypt",
+            "status": "Active",
+            "description": serde_json::Value::Null,
+            "metadata": HashMap::<String, String>::new(),
+            "created_at": "2024-01-01T00:00:00+00:00",
+            "rotated_at": serde_json::Value::Null,
+            "created_by": "beta5-fixture",
+            "encrypted_key_material": "xjwGa4Lj4qzKg6XQl8s2btyFkPHPChMAkjqs268TFGyvFUv8WjDD5HQCUDLViZmt",
+            "nonce": [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]
+        });
+        fs::write(
+            temp_dir.path().join("beta5-key.key"),
+            serde_json::to_vec_pretty(&stored_key).expect("serialize beta.5 fixture"),
+        )
+        .await
+        .expect("write beta.5 fixture");
+        let mut explicit_protection = stored_key.clone();
+        let explicit_object = explicit_protection.as_object_mut().expect("beta.5 fixture is a JSON object");
+        explicit_object.insert("key_id".to_string(), serde_json::json!("beta5-explicit-key"));
+        explicit_object.insert("at_rest_protection".to_string(), serde_json::json!("encrypted-master-key"));
+        fs::write(
+            temp_dir.path().join("beta5-explicit-key.key"),
+            serde_json::to_vec_pretty(&explicit_protection).expect("serialize explicit-protection fixture"),
+        )
+        .await
+        .expect("write explicit-protection fixture");
+
+        let client = LocalKmsClient::new(LocalConfig {
+            key_dir: temp_dir.path().to_path_buf(),
+            master_key: Some("beta5-test-master-key".to_string()),
+            file_permissions: Some(0o600),
+        })
+        .await
+        .expect("initialize local KMS for beta.5 fixture");
+
+        let material = client
+            .get_key_material("beta5-key")
+            .await
+            .expect("decrypt beta.5 SHA-256 protected key");
+        assert_eq!(material, vec![0x42; 32]);
+        let explicit_error = client
+            .get_key_material("beta5-explicit-key")
+            .await
+            .expect_err("explicit current protection must not fall back to the beta.5 KDF");
+        assert!(matches!(explicit_error, KmsError::CryptographicError { .. }));
+
+        let wrong_key_client = LocalKmsClient::new(LocalConfig {
+            key_dir: temp_dir.path().to_path_buf(),
+            master_key: Some("wrong-beta5-master-key".to_string()),
+            file_permissions: Some(0o600),
+        })
+        .await
+        .expect("initialize local KMS with wrong beta.5 master key");
+        let error = wrong_key_client
+            .get_key_material("beta5-key")
+            .await
+            .expect_err("wrong beta.5 master key must not decrypt the fixture");
+        assert!(matches!(error, KmsError::CryptographicError { .. }));
+    }
 }

crates/kms/src/encryption/dek.rs

@@ -24,6 +24,7 @@ use crate::error::{KmsError, Result};
 use async_trait::async_trait;
 use jiff::Zoned;
 use rand::Rng;
+use serde::de::IgnoredAny;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
@@ -44,6 +45,30 @@ pub struct DataKeyEnvelope {
     pub created_at: Zoned,
 }
 
+#[derive(Deserialize)]
+struct DataKeyEnvelopeMarker {
+    #[serde(rename = "key_id")]
+    _key_id: IgnoredAny,
+    #[serde(rename = "master_key_id")]
+    _master_key_id: IgnoredAny,
+    #[serde(rename = "key_spec")]
+    _key_spec: IgnoredAny,
+    #[serde(rename = "encrypted_key")]
+    _encrypted_key: IgnoredAny,
+    #[serde(rename = "nonce")]
+    _nonce: IgnoredAny,
+    #[serde(rename = "encryption_context")]
+    _encryption_context: IgnoredAny,
+    #[serde(rename = "created_at")]
+    _created_at: IgnoredAny,
+}
+
+/// Returns whether ciphertext is a RustFS KMS data-key envelope.
+pub fn is_data_key_envelope(ciphertext: &[u8]) -> bool {
+    ciphertext.iter().copied().find(|byte| !byte.is_ascii_whitespace()) == Some(b'{')
+        && serde_json::from_slice::<DataKeyEnvelopeMarker>(ciphertext).is_ok()
+}
+
 /// Trait for encrypting and decrypting data encryption keys (DEK)
 ///
 /// This trait abstracts the encryption operations used to protect
@@ -329,4 +354,46 @@ mod tests {
         assert_eq!(deserialized.key_id, "test-key-id");
         assert_eq!(deserialized.master_key_id, "master-key-id");
     }
+
+    #[test]
+    fn test_data_key_envelope_discriminator_rejects_local_formats() {
+        let kms_envelope = br#"{
+            "key_id": "test-key-id",
+            "master_key_id": "master-key-id",
+            "key_spec": "AES_256",
+            "encrypted_key": [1, 2, 3, 4],
+            "nonce": [5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16],
+            "encryption_context": {},
+            "created_at": "2024-01-01T00:00:00+00:00"
+        }"#;
+        let minio_legacy = br#"{"aead":"AES-256-GCM-HMAC-SHA-256","iv":[1],"nonce":[2],"bytes":[3]}"#;
+        let duplicate_key_id = [b"{\"key_id\":\"duplicate\",".as_slice(), &kms_envelope[1..]].concat();
+
+        assert!(is_data_key_envelope(kms_envelope));
+        assert!(is_data_key_envelope(&[b" \n".as_slice(), kms_envelope].concat()));
+        assert!(!is_data_key_envelope(&duplicate_key_id));
+        assert!(!is_data_key_envelope(b"bm9uY2U=:Y2lwaGVydGV4dA=="));
+        assert!(!is_data_key_envelope(minio_legacy));
+
+        let envelope_value: serde_json::Value = serde_json::from_slice(kms_envelope).expect("parse KMS envelope fixture");
+        for required_field in [
+            "key_id",
+            "master_key_id",
+            "key_spec",
+            "encrypted_key",
+            "nonce",
+            "encryption_context",
+            "created_at",
+        ] {
+            let mut incomplete = envelope_value.clone();
+            incomplete
+                .as_object_mut()
+                .expect("KMS envelope fixture is an object")
+                .remove(required_field);
+            assert!(
+                !is_data_key_envelope(&serde_json::to_vec(&incomplete).expect("serialize incomplete envelope")),
+                "missing {required_field} must not classify as a KMS envelope"
+            );
+        }
+    }
 }

crates/kms/src/encryption/mod.rs

@@ -17,4 +17,4 @@
 pub mod ciphers;
 pub mod dek;
 
-pub use dek::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
+pub use dek::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material, is_data_key_envelope};

crates/kms/src/error.rs

@@ -19,6 +19,11 @@ use thiserror::Error;
 /// Result type for KMS operations
 pub type Result<T> = std::result::Result<T, KmsError>;
 
+/// KMS runtime is unavailable for an encryption operation.
+#[derive(Error, Debug, Clone, Copy)]
+#[error("KMS encryption service is unavailable")]
+pub struct KmsUnavailableError;
+
 /// KMS error types covering all possible failure scenarios
 #[derive(Error, Debug, Clone)]
 pub enum KmsError {

crates/kms/src/lib.rs

@@ -83,7 +83,8 @@ pub use api_types::{
     TagKeyRequest, TagKeyResponse, UntagKeyRequest, UntagKeyResponse, UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse,
 };
 pub use config::*;
-pub use error::{KmsError, Result};
+pub use encryption::is_data_key_envelope;
+pub use error::{KmsError, KmsUnavailableError, Result};
 pub use manager::KmsManager;
 pub use service::{DataKey, ObjectEncryptionService};
 pub use service_manager::{

docs/architecture/compat-cleanup-register.md

@@ -19,6 +19,7 @@ for later deletion.
 - `disk-mutation-body-digest` internode mutating disk RPCs: servers temporarily accept mutating disk RPCs (RenameData, DeleteVersion, DeleteVersions, WriteMetadata, UpdateMetadata, WriteAll, Delete, DeletePaths, RenameFile, RenamePart, DeleteVolume, MakeVolume, MakeVolumes) that carry no signature-bound canonical body digest, so peers from releases that predate body-digest signing remain available during rolling upgrades. Accepted digestless mutations increment the internode body-digest fallback counter; that counter must read zero fleet-wide across a release window before RUSTFS_INTERNODE_RPC_BODY_DIGEST_STRICT is enabled. Because body-bound requests now consume replay-cache nonces on the receiver, deploy the raised RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY default fleet-wide before enabling strict mode, and watch the internode replay-cache overflow counter for undersized capacity during the rollout. Remove the digestless fallback after the minimum supported RustFS peer version body-binds every mutating disk RPC.
 - `heal-status-rpc-v1` node heal status capability: new peers treat an unimplemented BackgroundHealStatus RPC as an explicitly incomplete rolling-upgrade response. Remove the fallback after the minimum supported RustFS peer version implements BackgroundHealStatus.
 - `backlog-1316` legacy encrypted multipart range seek: the feature remains opt-in until every server that can initiate, write, or complete multipart uploads supports the candidate-to-final marker protocol and uploadId commit lock, and pre-upgrade multipart uploads have drained. Remove the RUSTFS_ENCRYPTED_RANGE_SEEK switch after the minimum supported release does so; keep the quorum marker and malformed-layout full-read guards permanently.
+- `rustfs-5063` pre-beta.9 Local KMS recovery: persisted Local KMS configs from beta.8 and earlier predate the explicit insecure-development flag, and encrypted key files use the legacy SHA-256 KDF. Remove the config fallback after supported upgrades have rewritten or explicitly resaved all pre-beta.9 configs with the development-default field, and remove the legacy KDF after supported upgrades have rewritten all pre-beta.9 Local KMS key files with explicit at-rest protection.
 
 ## Review Checklist
 

rustfs/src/admin/handlers/kms_dynamic.rs

@@ -128,6 +128,32 @@ async fn save_kms_config(config: &KmsConfig) -> Result<(), String> {
     Ok(())
 }
 
+fn decode_persisted_kms_config(data: &[u8]) -> serde_json::Result<(KmsConfig, bool)> {
+    let mut config: KmsConfig = serde_json::from_slice(data)?;
+    let value: serde_json::Value = serde_json::from_slice(data)?;
+    let is_missing_development_flag = value
+        .as_object()
+        .is_some_and(|object| !object.contains_key("allow_insecure_dev_defaults"));
+    let mut uses_legacy_local_defaults = false;
+
+    if is_missing_development_flag
+        && matches!(&config.backend_config, rustfs_kms::BackendConfig::Local(_))
+        && config.validate().is_err()
+    {
+        // RUSTFS_COMPAT_TODO(rustfs-5063): Remove after pre-beta.9 configurations are rewritten with this field.
+        // Pre-beta.9 persisted Local KMS configurations predate the explicit
+        // development-default flag.
+        config.allow_insecure_dev_defaults = true;
+        if config.validate().is_ok() {
+            uses_legacy_local_defaults = true;
+        } else {
+            config.allow_insecure_dev_defaults = false;
+        }
+    }
+
+    Ok((config, uses_legacy_local_defaults))
+}
+
 /// Load KMS configuration from cluster storage
 #[instrument]
 pub async fn load_kms_config() -> Option<KmsConfig> {
@@ -145,8 +171,18 @@ pub async fn load_kms_config() -> Option<KmsConfig> {
     };
 
     match read_admin_config(store, KMS_CONFIG_PATH).await {
-        Ok(data) => match serde_json::from_slice::<KmsConfig>(&data) {
+        Ok(data) => match decode_persisted_kms_config(&data) {
-            Ok(config) => {
+            Ok((config, is_legacy_local)) => {
+                if is_legacy_local {
+                    warn!(
+                        component = LOG_COMPONENT_ADMIN,
+                        subsystem = LOG_SUBSYSTEM_KMS,
+                        event = "kms_legacy_local_config_loaded",
+                        storage_path = KMS_CONFIG_PATH,
+                        state = "legacy_config_accepted",
+                        "admin kms dynamic state"
+                    );
+                }
                 info!(
                     component = LOG_COMPONENT_ADMIN,
                     subsystem = LOG_SUBSYSTEM_KMS,
@@ -921,8 +957,9 @@ impl Operation for ReconfigureKmsHandler {
 
 #[cfg(test)]
 mod tests {
-    use super::{kms_configure_actions, kms_service_control_actions};
+    use super::{decode_persisted_kms_config, kms_configure_actions, kms_service_control_actions};
     use rustfs_policy::policy::action::{Action, AdminAction, KmsAction};
+    use tempfile::TempDir;
 
     fn assert_has_action(actions: &[Action], action: Action) {
         assert!(actions.contains(&action), "expected action list to contain {action:?}");
@@ -943,4 +980,67 @@ mod tests {
         assert_lacks_action(&kms_configure_actions(), Action::AdminAction(AdminAction::ServerInfoAdminAction));
         assert_lacks_action(&kms_service_control_actions(), Action::AdminAction(AdminAction::ServerInfoAdminAction));
     }
+
+    #[test]
+    fn persisted_beta5_local_config_retains_legacy_development_mode() {
+        let temp_dir = TempDir::new().expect("create legacy local KMS directory");
+        let config = rustfs_kms::KmsConfig::local(temp_dir.path().to_path_buf());
+        let mut value = serde_json::to_value(config).expect("serialize local KMS config");
+        value
+            .as_object_mut()
+            .expect("KMS config is a JSON object")
+            .remove("allow_insecure_dev_defaults");
+
+        let (config, migrated) = decode_persisted_kms_config(&serde_json::to_vec(&value).expect("serialize beta.5 config"))
+            .expect("decode beta.5 persisted config");
+        assert!(migrated);
+        assert!(config.allow_insecure_dev_defaults);
+        assert!(config.validate().is_ok());
+    }
+
+    #[test]
+    fn persisted_local_config_with_explicit_secure_mode_stays_secure() {
+        let temp_dir = TempDir::new().expect("create secure local KMS directory");
+        let config = rustfs_kms::KmsConfig::local(temp_dir.path().to_path_buf());
+
+        let (config, migrated) = decode_persisted_kms_config(&serde_json::to_vec(&config).expect("serialize current config"))
+            .expect("decode current persisted config");
+        assert!(!migrated);
+        assert!(!config.allow_insecure_dev_defaults);
+        assert!(config.validate().is_err());
+    }
+
+    #[test]
+    fn persisted_config_rejects_duplicate_security_field() {
+        let temp_dir = TempDir::new().expect("create local KMS directory");
+        let config = rustfs_kms::KmsConfig::local(temp_dir.path().to_path_buf());
+        let serialized = serde_json::to_string(&config).expect("serialize current config");
+        let duplicate = serialized.replacen('{', r#"{"allow_insecure_dev_defaults":true,"#, 1);
+
+        assert!(decode_persisted_kms_config(duplicate.as_bytes()).is_err());
+    }
+
+    #[test]
+    fn persisted_secure_local_config_without_legacy_field_stays_secure() {
+        #[cfg(unix)]
+        let key_dir = std::path::PathBuf::from("/var/lib/rustfs/kms");
+        #[cfg(windows)]
+        let key_dir = std::path::PathBuf::from(r"C:\rustfs-kms");
+        let mut config = rustfs_kms::KmsConfig::local(key_dir);
+        let rustfs_kms::BackendConfig::Local(local) = &mut config.backend_config else {
+            panic!("local constructor must create local backend config");
+        };
+        local.master_key = Some("configured-master-key".to_string());
+        let mut value = serde_json::to_value(config).expect("serialize secure local KMS config");
+        value
+            .as_object_mut()
+            .expect("KMS config is a JSON object")
+            .remove("allow_insecure_dev_defaults");
+
+        let (config, migrated) = decode_persisted_kms_config(&serde_json::to_vec(&value).expect("serialize old config"))
+            .expect("decode secure persisted config");
+        assert!(!migrated);
+        assert!(!config.allow_insecure_dev_defaults);
+        assert!(config.validate().is_ok());
+    }
 }

rustfs/src/error.rs

@@ -14,6 +14,7 @@
 
 use crate::storage_api::error::contract::range::HTTPRangeError;
 use crate::storage_api::error::{QuotaError, StorageError};
+use rustfs_kms::KmsUnavailableError;
 use s3s::{S3Error, S3ErrorCode};
 
 #[derive(Debug)]
@@ -231,6 +232,19 @@ impl From<StorageError> for ApiError {
             };
         }
 
+        if let StorageError::Io(ref io_err) = err
+            && io_err
+                .get_ref()
+                .and_then(|inner| inner.downcast_ref::<KmsUnavailableError>())
+                .is_some()
+        {
+            return ApiError {
+                code: S3ErrorCode::ServiceUnavailable,
+                message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable),
+                source: Some(Box::new(err)),
+            };
+        }
+
         let code = match &err {
             StorageError::NotImplemented => S3ErrorCode::NotImplemented,
             StorageError::InvalidArgument(_, _, _) => S3ErrorCode::InvalidArgument,
@@ -464,6 +478,14 @@ mod tests {
         assert!(downcast_storage_error.is_some());
     }
 
+    #[test]
+    fn test_kms_service_unavailable_maps_to_retryable_error() {
+        let api_error = ApiError::from(StorageError::other(KmsUnavailableError));
+
+        assert_eq!(api_error.code, S3ErrorCode::ServiceUnavailable);
+        assert_eq!(api_error.message, "The service is unavailable. Please retry.");
+    }
+
     #[test]
     fn test_api_error_from_storage_error_mappings() {
         let test_cases = vec![

rustfs/src/storage/sse.rs

@@ -87,7 +87,7 @@ use http::{HeaderMap, HeaderValue};
 use rand::Rng;
 #[cfg(feature = "rio-v2")]
 use rand::RngExt;
-use rustfs_kms::{DataKey, types::ObjectEncryptionContext};
+use rustfs_kms::{DataKey, KmsUnavailableError, is_data_key_envelope, types::ObjectEncryptionContext};
 use rustfs_utils::get_env_opt_str;
 use s3s::S3ErrorCode;
 use s3s::dto::ServerSideEncryption;
@@ -1618,10 +1618,7 @@ async fn apply_managed_encryption_material(
 
     let provider = get_sse_dek_provider().await?;
     let object_context = build_object_encryption_context(bucket, key, ssekms_context.as_ref());
-    let (data_key, encrypted_data_key) = provider
+    let (data_key, encrypted_data_key) = provider.generate_sse_dek(&object_context, &kms_key_to_use).await?;
-        .generate_sse_dek(&object_context, &kms_key_to_use)
-        .await
-        .map_err(|e| ApiError::from(StorageError::other(format!("Failed to create data key: {e}"))))?;
 
     let algorithm = server_side_encryption.as_str().to_string();
     #[cfg(feature = "rio-v2")]
@@ -1744,8 +1741,25 @@ async fn apply_managed_decryption_material(
     };
     let object_context = build_object_encryption_context(bucket, key, kms_context.as_ref());
 
-    // Use factory pattern to get provider (test or production mode)
+    // Persisted wrapping format is the read-side source of truth. The
-    let provider = get_sse_dek_provider().await?;
+    // advertised SSE scheme and current KMS availability are write policy
+    // and runtime state, neither of which identifies the historical provider.
+    let provider: Arc<dyn SseDekProvider> = if is_data_key_envelope(&encrypted_data_key) {
+        // When a test-injected provider is registered via set_sse_dek_provider_for_test,
+        // use it instead of creating a fresh KmsSseDekProvider which cannot resolve
+        // a KMS service without an AppContext.
+        //
+        // Reads GLOBAL_KMS_DEK_PROVIDER (populated only by set_sse_dek_provider_for_test),
+        // never GLOBAL_SSE_DEK_PROVIDER, so a local provider cached by a prior
+        // get_local_sse_dek_provider call cannot be selected to unwrap a KMS envelope.
+        if let Some(cached) = GLOBAL_KMS_DEK_PROVIDER.read().ok().and_then(|guard| guard.as_ref().cloned()) {
+            cached
+        } else {
+            Arc::new(KmsSseDekProvider::new().await?)
+        }
+    } else {
+        get_local_sse_dek_provider().await?
+    };
     #[cfg(feature = "rio-v2")]
     let decrypted_data_key = if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
         provider
@@ -1760,8 +1774,7 @@ async fn apply_managed_decryption_material(
     let decrypted_data_key = provider
         .decrypt_sse_dek(&encrypted_data_key, &kms_key_id, &object_context)
         .await;
-    let decrypted_data_key =
+    let decrypted_data_key = decrypted_data_key?;
-        decrypted_data_key.map_err(|e| ApiError::from(StorageError::other(format!("Failed to decrypt data key: {e}"))))?;
     #[cfg(feature = "rio-v2")]
     let (key_bytes, base_nonce, key_kind) = if let Some(sealed_key) = minio_sealed_key {
         (
@@ -1873,7 +1886,7 @@ impl KmsSseDekProvider {
         provider
             .current_service()
             .await
-            .ok_or_else(|| ApiError::from(StorageError::other("KMS encryption service is not initialized")))?;
+            .ok_or_else(|| ApiError::from(StorageError::other(KmsUnavailableError)))?;
         Ok(provider)
     }
 
@@ -1885,7 +1898,7 @@ impl KmsSseDekProvider {
         provider
             .current_service()
             .await
-            .ok_or_else(|| ApiError::from(StorageError::other("KMS encryption service is not initialized")))?;
+            .ok_or_else(|| ApiError::from(StorageError::other(KmsUnavailableError)))?;
         Ok(provider)
     }
 
@@ -1910,7 +1923,7 @@ impl SseDekProvider for KmsSseDekProvider {
         let service = self
             .current_service()
             .await
-            .ok_or_else(|| ApiError::from(StorageError::other("KMS encryption service is not initialized")))?;
+            .ok_or_else(|| ApiError::from(StorageError::other(KmsUnavailableError)))?;
         let (data_key, encrypted_data_key) = service
             .create_data_key(&kms_key_option, context)
             .await
@@ -1928,7 +1941,7 @@ impl SseDekProvider for KmsSseDekProvider {
         let service = self
             .current_service()
             .await
-            .ok_or_else(|| ApiError::from(StorageError::other("KMS encryption service is not initialized")))?;
+            .ok_or_else(|| ApiError::from(StorageError::other(KmsUnavailableError)))?;
         let data_key = service
             .decrypt_data_key(encrypted_dek, context)
             .await
@@ -1947,7 +1960,7 @@ impl SseDekProvider for KmsSseDekProvider {
         let service = self
             .current_service()
             .await
-            .ok_or_else(|| ApiError::from(StorageError::other("KMS encryption service is not initialized")))?;
+            .ok_or_else(|| ApiError::from(StorageError::other(KmsUnavailableError)))?;
         let data_key = service
             .decrypt_legacy_data_key(encrypted_dek)
             .await
@@ -1961,41 +1974,33 @@ impl SseDekProvider for KmsSseDekProvider {
 // Test/Simple DEK Provider
 // ============================================================================
 
-/// Simple SSE DEK provider for testing purposes
+/// Local SSE DEK provider for deployments without a KMS.
 ///
-/// This provider reads a single 32-byte customer master key (CMK) from the
+/// Uses `RUSTFS_SSE_S3_MASTER_KEY` (base64-encoded 32-byte key) to wrap
-/// `__RUSTFS_SSE_SIMPLE_CMK` environment variable. The key must be base64-encoded.
+/// data-encryption keys with AES-256-GCM.  This is the production fallback
+/// when no KMS service is configured.
 ///
-/// # Environment Variable Format
+/// # Environment Variable
 ///
 /// ```text
-/// __RUSTFS_SSE_SIMPLE_CMK=<base64_encoded_32_byte_key>
+/// RUSTFS_SSE_S3_MASTER_KEY=<base64_encoded_32_byte_key>
 /// ```
-///
+pub(crate) struct LocalSseDekProvider {
-/// Example:
-/// ```bash
-/// export __RUSTFS_SSE_SIMPLE_CMK="AKHul86TBMMJ3+VrGlh9X3dHJsOtSXOXHOODPwmAnOo="
-/// ```
-///
-/// # Key Generation
-///
-/// Use the provided script to generate a valid key:
-/// ```bash
-/// # Windows
-/// .\scripts\generate-sse-keys.ps1
-///
-/// # Linux/Unix/macOS
-/// ./scripts/generate-sse-keys.sh
-/// ```
-pub(crate) struct TestSseDekProvider {
     master_key: [u8; 32],
 }
 
+/// Test-only alias so existing test code that references `TestSseDekProvider`
+/// continues to compile without changes.
+#[cfg(test)]
+#[allow(non_camel_case_types)]
+pub(crate) type TestSseDekProvider = LocalSseDekProvider;
+
 /// Parse the base64-encoded 32-byte master key from `__RUSTFS_SSE_SIMPLE_CMK`.
 ///
 /// Returns an error (never crashes) for a missing, non-base64, wrong-length, or
 /// all-zero key so callers on the request path can fail the request instead of
 /// taking the whole server down (backlog#806).
+#[cfg(test)]
 fn parse_simple_sse_cmk(cmk_value: &str) -> Result<[u8; 32], ApiError> {
     let trimmed = cmk_value.trim();
     if trimmed.is_empty() {
@@ -2018,21 +2023,23 @@ fn parse_simple_sse_cmk(cmk_value: &str) -> Result<[u8; 32], ApiError> {
     Ok(master_key)
 }
 
-impl TestSseDekProvider {
+impl LocalSseDekProvider {
-    /// Create a SimpleSseDekProvider with a predefined key (for testing)
+    /// Create a LocalSseDekProvider with a predefined key (for testing)
     #[cfg(test)]
     pub fn new_with_key(master_key: [u8; 32]) -> Self {
         Self { master_key }
     }
 
+    /// Create a TestSseDekProvider from `__RUSTFS_SSE_SIMPLE_CMK` (test-only).
+    #[cfg(test)]
     pub fn new() -> Result<Self, ApiError> {
         let cmk_value = std::env::var("__RUSTFS_SSE_SIMPLE_CMK").unwrap_or_default();
         // A missing/invalid key must surface as a request error, never crash the
-        // whole server: `TestSseDekProvider::new` is reached from the SSE request
+        // whole server: `LocalSseDekProvider::new` is reached from the SSE request
         // path (get_sse_dek_provider), so `process::exit(1)` here turned a bad
         // `__RUSTFS_SSE_SIMPLE_CMK` into a process crash-loop DoS (backlog#806).
         let master_key = parse_simple_sse_cmk(&cmk_value)?;
-        tracing::info!("Successfully loaded SSE master key (32 bytes)");
+        tracing::info!("Successfully loaded SSE master key (32 bytes) from __RUSTFS_SSE_SIMPLE_CMK");
         Ok(Self { master_key })
     }
 
@@ -2042,7 +2049,7 @@ impl TestSseDekProvider {
     /// The failures here are server configuration problems, not internal
     /// faults: surface them as `InvalidRequest` (HTTP 400) so a managed-SSE
     /// request against an unconfigured server does not report 500 (rustfs#4844).
-    pub fn new_for_local_sse() -> Result<Self, ApiError> {
+    pub fn new_from_env() -> Result<Self, ApiError> {
         fn sse_not_configured(message: impl Into<String>) -> ApiError {
             ApiError {
                 code: S3ErrorCode::InvalidRequest,
@@ -2122,7 +2129,7 @@ impl TestSseDekProvider {
 }
 
 #[async_trait]
-impl SseDekProvider for TestSseDekProvider {
+impl SseDekProvider for LocalSseDekProvider {
     async fn generate_sse_dek(
         &self,
         _context: &ObjectEncryptionContext,
@@ -2167,9 +2174,22 @@ impl SseDekProvider for TestSseDekProvider {
 // Factory Function for SSE DEK Provider
 // ============================================================================
 
-/// Global SSE DEK provider cache
+/// Global SSE DEK provider cache for local / test providers.
+///
+/// Populated by `get_local_sse_dek_provider` and (for backward-compat in tests)
+/// `set_sse_dek_provider_for_test`.  Read by `get_local_sse_dek_provider` only —
+/// the KMS-envelope decrypt path uses `GLOBAL_KMS_DEK_PROVIDER` so that a cached
+/// local provider can never be selected for a KMS-wrapped data-key.
 static GLOBAL_SSE_DEK_PROVIDER: LazyLock<RwLock<Option<Arc<dyn SseDekProvider>>>> = LazyLock::new(|| RwLock::new(None));
 
+/// Global KMS DEK provider cache for test-injected KMS providers.
+///
+/// Populated **only** by `set_sse_dek_provider_for_test` (test-only).
+/// Read **only** by the KMS-envelope branch of `apply_managed_decryption_material`.
+/// Separation from `GLOBAL_SSE_DEK_PROVIDER` prevents a previously-cached local
+/// provider from being selected to unwrap a KMS data-key envelope.
+static GLOBAL_KMS_DEK_PROVIDER: LazyLock<RwLock<Option<Arc<dyn SseDekProvider>>>> = LazyLock::new(|| RwLock::new(None));
+
 /// Get or initialize the global SSE DEK provider
 ///
 /// Factory function that automatically selects the appropriate provider:
@@ -2192,6 +2212,17 @@ pub async fn get_sse_dek_provider() -> Result<Arc<dyn SseDekProvider>, ApiError>
         return Ok(Arc::new(KmsSseDekProvider::new().await?));
     }
 
+    get_local_sse_dek_provider().await
+}
+
+async fn get_local_sse_dek_provider() -> Result<Arc<dyn SseDekProvider>, ApiError> {
+    // An explicitly injected test provider overrides both the cache and the
+    // environment-based selection below.
+    #[cfg(test)]
+    if let Some(injected) = test_injected_sse_dek_provider() {
+        return Ok(injected);
+    }
+
     // Check if already initialized
     if let Some(provider) = GLOBAL_SSE_DEK_PROVIDER
         .read()
@@ -2202,14 +2233,26 @@ pub async fn get_sse_dek_provider() -> Result<Arc<dyn SseDekProvider>, ApiError>
         return Ok(provider);
     }
 
-    // Determine provider: KMS when available, else test env, else local SSE-S3 fallback (no KMS)
+    // In test mode, prefer the simple CMK provider when the env var is set.
-    let provider: Arc<dyn SseDekProvider> = if std::env::var("__RUSTFS_SSE_SIMPLE_CMK").is_ok() {
+    #[cfg(test)]
-        debug!("Using SimpleSseDekProvider (test mode) based on __RUSTFS_SSE_SIMPLE_CMK");
+    {
-        Arc::new(TestSseDekProvider::new()?)
+        if std::env::var("__RUSTFS_SSE_SIMPLE_CMK").is_ok() {
-    } else {
+            debug!("Using LocalSseDekProvider (test mode) based on __RUSTFS_SSE_SIMPLE_CMK");
+            let provider: Arc<dyn SseDekProvider> = Arc::new(LocalSseDekProvider::new()?);
+            let mut slot = GLOBAL_SSE_DEK_PROVIDER
+                .write()
+                .map_err(|_| ApiError::from(StorageError::other("Failed to update global SSE DEK provider cache")))?;
+            if let Some(existing) = slot.as_ref() {
+                return Ok(existing.clone());
+            }
+            *slot = Some(provider.clone());
+            return Ok(provider);
+        }
+    }
+
+    // Production fallback: local SSE-S3 provider (no KMS configured).
     debug!("Using local SSE-S3 provider (KMS not configured)");
-        Arc::new(TestSseDekProvider::new_for_local_sse()?)
+    let provider: Arc<dyn SseDekProvider> = Arc::new(LocalSseDekProvider::new_from_env()?);
-    };
 
     let mut slot = GLOBAL_SSE_DEK_PROVIDER
         .write()
@@ -2222,26 +2265,41 @@ pub async fn get_sse_dek_provider() -> Result<Arc<dyn SseDekProvider>, ApiError>
     Ok(provider)
 }
 
-/// Reset the global SSE DEK provider (for testing only)
+/// Reset both global SSE DEK provider caches (for testing only)
 ///
-/// Note: OnceLock doesn't support reset in stable Rust.
+/// Clears GLOBAL_SSE_DEK_PROVIDER (local/test providers) and
-/// Tests should set environment variables before first call to `get_sse_dek_provider()`.
+/// GLOBAL_KMS_DEK_PROVIDER (test-injected KMS providers).
 #[cfg(test)]
 #[allow(dead_code)]
 pub fn reset_sse_dek_provider() {
     if let Ok(mut slot) = GLOBAL_SSE_DEK_PROVIDER.write() {
         *slot = None;
     }
+    if let Ok(mut slot) = GLOBAL_KMS_DEK_PROVIDER.write() {
+        *slot = None;
+    }
 }
 
 #[cfg(test)]
-#[cfg(feature = "rio-v2")]
+#[allow(dead_code)]
 pub fn set_sse_dek_provider_for_test(provider: Arc<dyn SseDekProvider>) {
+    if let Ok(mut slot) = GLOBAL_KMS_DEK_PROVIDER.write() {
+        *slot = Some(provider.clone());
+    }
     if let Ok(mut slot) = GLOBAL_SSE_DEK_PROVIDER.write() {
         *slot = Some(provider);
     }
 }
 
+/// Provider explicitly injected via `set_sse_dek_provider_for_test`, if any.
+///
+/// Reads `GLOBAL_KMS_DEK_PROVIDER` because that slot is populated *only* by the
+/// test setter, so a hit here always means an explicit test injection.
+#[cfg(test)]
+fn test_injected_sse_dek_provider() -> Option<Arc<dyn SseDekProvider>> {
+    GLOBAL_KMS_DEK_PROVIDER.read().ok().and_then(|guard| guard.as_ref().cloned())
+}
+
 // ============================================================================
 // Legacy Functions (SSE-S3 / SSE-KMS)
 // ============================================================================
@@ -2518,26 +2576,27 @@ fn ssec_invalid_request(message: &str) -> ApiError {
 #[cfg(test)]
 #[allow(unused_imports)]
 mod tests {
-    #[cfg(feature = "rio-v2")]
     use super::{
-        DARE_CIPHER_AES_256_GCM, DARE_CIPHER_CHACHA20_POLY1305, MINIO_INTERNAL_ENCRYPTION_SEAL_ALGORITHM, SEALED_KEY_IV_SIZE,
+        ApiError, DataKey, DecryptionRequest, EncryptionKeyKind, EncryptionMaterial, EncryptionRequest,
-        SEALED_KEY_SIZE, is_legacy_rustfs_managed_metadata, is_supported_sealed_object_key_cipher,
+        INTERNAL_ENCRYPTION_ALGORITHM_HEADER, INTERNAL_ENCRYPTION_IV_HEADER, INTERNAL_ENCRYPTION_KEY_HEADER,
-    };
+        INTERNAL_ENCRYPTION_KEY_ID_HEADER, KmsSseDekProvider, KmsUnavailableError, MINIO_INTERNAL_ENCRYPTION_ALGORITHM_HEADER,
-    use super::{
+        MINIO_INTERNAL_ENCRYPTION_IV_HEADER, MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER,
-        DecryptionRequest, EncryptionKeyKind, EncryptionMaterial, EncryptionRequest, INTERNAL_ENCRYPTION_ALGORITHM_HEADER,
+        MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER, MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER,
-        INTERNAL_ENCRYPTION_IV_HEADER, INTERNAL_ENCRYPTION_KEY_HEADER, INTERNAL_ENCRYPTION_KEY_ID_HEADER, KmsSseDekProvider,
+        MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER, MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER,
-        MINIO_INTERNAL_ENCRYPTION_ALGORITHM_HEADER, MINIO_INTERNAL_ENCRYPTION_IV_HEADER,
+        MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER, PrepareEncryptionRequest, SSEC_ORIGINAL_SIZE_HEADER, SSEType,
-        MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER, MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER,
+        SseDekProvider, SsecParams, StorageError, TestSseDekProvider, apply_managed_decryption_material,
-        MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER, MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER,
+        apply_managed_encryption_material, encryption_material_to_metadata, extract_server_side_encryption_from_headers,
-        MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER, MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER,
-        PrepareEncryptionRequest, SSEC_ORIGINAL_SIZE_HEADER, SSEType, SseDekProvider, SsecParams, StorageError,
-        TestSseDekProvider, encryption_material_to_metadata, extract_server_side_encryption_from_headers,
         extract_ssec_params_from_headers, extract_ssekms_context_from_headers, generate_ssec_nonce, is_managed_sse,
         map_get_object_reader_error, mark_encrypted_multipart_metadata, normalize_managed_metadata, reset_sse_dek_provider,
         resolve_effective_kms_key_id, sse_decryption, sse_encryption, sse_prepare_encryption, strip_managed_encryption_metadata,
         validate_sse_headers_for_read, validate_sse_headers_for_write, validate_ssec_for_read, validate_ssec_params,
         verify_ssec_key_match,
     };
+    #[cfg(feature = "rio-v2")]
+    use super::{
+        DARE_CIPHER_AES_256_GCM, DARE_CIPHER_CHACHA20_POLY1305, MINIO_INTERNAL_ENCRYPTION_SEAL_ALGORITHM, SEALED_KEY_IV_SIZE,
+        SEALED_KEY_SIZE, is_legacy_rustfs_managed_metadata, is_supported_sealed_object_key_cipher,
+    };
 
     #[test]
     fn parse_simple_sse_cmk_rejects_bad_keys_without_crashing() {
@@ -2582,6 +2641,28 @@ mod tests {
         SSE_TEST_LOCK.get_or_init(|| Mutex::new(())).lock().await
     }
 
+    struct UnavailableSseDekProvider;
+
+    #[async_trait::async_trait]
+    impl SseDekProvider for UnavailableSseDekProvider {
+        async fn generate_sse_dek(
+            &self,
+            _context: &ObjectEncryptionContext,
+            _kms_key_id: &str,
+        ) -> Result<(DataKey, Vec<u8>), ApiError> {
+            Err(ApiError::from(StorageError::other(KmsUnavailableError)))
+        }
+
+        async fn decrypt_sse_dek(
+            &self,
+            _encrypted_dek: &[u8],
+            _kms_key_id: &str,
+            _context: &ObjectEncryptionContext,
+        ) -> Result<[u8; 32], ApiError> {
+            Err(ApiError::from(StorageError::other(KmsUnavailableError)))
+        }
+    }
+
     fn local_sse_master_key_b64() -> String {
         BASE64_STANDARD.encode([0x24u8; 32])
     }
@@ -4095,6 +4176,190 @@ mod tests {
         println!("✅ Full cycle (generate -> encrypt DEK -> decrypt DEK -> decrypt data) test passed!");
     }
 
+    #[tokio::test]
+    async fn test_managed_decryption_selects_provider_from_persisted_dek() {
+        use rustfs_kms::config::KmsConfig;
+        use tempfile::TempDir;
+
+        let _guard = lock_sse_test_state().await;
+        reset_sse_dek_provider();
+        let manager = rustfs_kms::init_global_kms_service_manager();
+        let key_dir = TempDir::new().expect("create KMS key directory");
+        manager
+            .reconfigure(KmsConfig::local(key_dir.path().to_path_buf()).with_insecure_development_defaults())
+            .await
+            .expect("start test KMS service");
+
+        let local_master_key = [7u8; 32];
+        let local_provider = TestSseDekProvider::new_with_key(local_master_key);
+        let context = ObjectEncryptionContext::new("bucket".to_string(), "object".to_string());
+        let (data_key, encrypted_dek) = local_provider
+            .generate_sse_dek(&context, "legacy-local-key")
+            .await
+            .expect("generate beta.5 local DEK");
+
+        async_with_vars(
+            [
+                ("__RUSTFS_SSE_SIMPLE_CMK", None::<String>),
+                ("RUSTFS_SSE_S3_MASTER_KEY", Some(BASE64_STANDARD.encode(local_master_key))),
+            ],
+            async {
+                let metadata = HashMap::from([
+                    ("x-amz-server-side-encryption".to_string(), ServerSideEncryption::AWS_KMS.to_string()),
+                    (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), BASE64_STANDARD.encode(encrypted_dek)),
+                    (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), BASE64_STANDARD.encode(data_key.nonce)),
+                    (INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "legacy-local-key".to_string()),
+                ]);
+
+                let material = apply_managed_decryption_material("bucket", "object", &metadata)
+                    .await
+                    .expect("legacy local DEK should not be routed to the running KMS")
+                    .expect("managed metadata should produce decryption material");
+                assert_eq!(material.key_bytes, data_key.plaintext_key);
+                assert_eq!(material.sse_type, SSEType::SseKms);
+            },
+        )
+        .await;
+
+        manager.stop().await.expect("stop test KMS service");
+        reset_sse_dek_provider();
+
+        let kms_envelope = br#"{
+            "key_id": "test-key-id",
+            "master_key_id": "master-key-id",
+            "key_spec": "AES_256",
+            "encrypted_key": [1, 2, 3, 4],
+            "nonce": [5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16],
+            "encryption_context": {},
+            "created_at": "2024-01-01T00:00:00+00:00"
+        }"#;
+        let metadata = HashMap::from([
+            ("x-amz-server-side-encryption".to_string(), ServerSideEncryption::AES256.to_string()),
+            (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), BASE64_STANDARD.encode(kms_envelope)),
+            (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), BASE64_STANDARD.encode([0x14; 12])),
+            (INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "test-key-id".to_string()),
+        ]);
+        let error = match apply_managed_decryption_material("bucket", "object", &metadata).await {
+            Ok(_) => panic!("KMS envelope must not fall back to the local provider"),
+            Err(error) => error,
+        };
+        assert_eq!(error.code, S3ErrorCode::ServiceUnavailable);
+
+        reset_sse_dek_provider();
+    }
+
+    /// Regression test for "local provider cached → dynamically enable KMS → decrypt KMS envelope".
+    ///
+    /// Verifies that a `TestSseDekProvider` previously cached in `GLOBAL_SSE_DEK_PROVIDER`
+    /// is NEVER selected to unwrap a KMS data-key envelope. The KMS-envelope branch must
+    /// read `GLOBAL_KMS_DEK_PROVIDER` (or fall back to `KmsSseDekProvider::new()`), not
+    /// the local-provider cache.
+    #[tokio::test]
+    async fn test_kms_envelope_never_routes_to_cached_local_provider() {
+        use rustfs_kms::config::KmsConfig;
+        use tempfile::TempDir;
+
+        let _guard = lock_sse_test_state().await;
+        reset_sse_dek_provider();
+
+        // 1. Populate GLOBAL_SSE_DEK_PROVIDER with a local provider — the kind
+        //    that `get_local_sse_dek_provider` would cache when KMS is absent.
+        let local_master_key = [0xAAu8; 32];
+        *super::GLOBAL_SSE_DEK_PROVIDER
+            .write()
+            .expect("write local provider into local cache") = Some(Arc::new(TestSseDekProvider::new_with_key(local_master_key)));
+
+        // 2. Start a KMS service (dynamic enable).
+        let manager = rustfs_kms::init_global_kms_service_manager();
+        let key_dir = TempDir::new().expect("create KMS key directory");
+        manager
+            .reconfigure(KmsConfig::local(key_dir.path().to_path_buf()).with_insecure_development_defaults())
+            .await
+            .expect("start test KMS service");
+
+        // 3. Construct a KMS JSON envelope — the persisted format of a KMS-wrapped DEK.
+        //    is_data_key_envelope() will return true for this payload.
+        let kms_envelope = br#"{
+            "key_id": "envelope-key",
+            "master_key_id": "master-key-id",
+            "key_spec": "AES_256",
+            "encrypted_key": [10, 20, 30, 40],
+            "nonce": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12],
+            "encryption_context": {},
+            "created_at": "2024-01-01T00:00:00+00:00"
+        }"#;
+        let metadata = HashMap::from([
+            ("x-amz-server-side-encryption".to_string(), ServerSideEncryption::AES256.to_string()),
+            (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), BASE64_STANDARD.encode(kms_envelope)),
+            (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), BASE64_STANDARD.encode([0x14; 12])),
+            (INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "envelope-key".to_string()),
+        ]);
+
+        // 4. Decrypt the KMS envelope. Before the fix, this would pick up the cached
+        //    TestSseDekProvider from GLOBAL_SSE_DEK_PROVIDER and either panic (wrong
+        //    format) or produce garbage. After the fix, it routes to KmsSseDekProvider,
+        //    which fails because the envelope contains dummy encrypted bytes that the
+        //    test KMS cannot decrypt — but crucially the error code is NOT a local-
+        //    provider error.
+        let error = match apply_managed_decryption_material("bucket", "object", &metadata).await {
+            Ok(_) => panic!("dummy KMS envelope must not produce valid decryption material"),
+            Err(error) => error,
+        };
+        // The KMS service was reached (ServiceUnavailable or the KMS's own decrypt
+        // failure), not a local-provider format error.
+        assert!(
+            error.code == S3ErrorCode::ServiceUnavailable || error.code == S3ErrorCode::InternalError,
+            "KMS envelope must be routed to KMS provider, not local; got code {:?} msg '{}'",
+            error.code,
+            error.message,
+        );
+
+        manager.stop().await.expect("stop test KMS service");
+        reset_sse_dek_provider();
+    }
+
+    #[tokio::test]
+    async fn test_managed_encryption_preserves_provider_unavailable_error() {
+        let _guard = lock_sse_test_state().await;
+        let manager = rustfs_kms::init_global_kms_service_manager();
+        manager.stop().await.expect("stop global KMS service");
+        reset_sse_dek_provider();
+        *super::GLOBAL_SSE_DEK_PROVIDER.write().expect("update SSE DEK provider cache") =
+            Some(Arc::new(UnavailableSseDekProvider));
+
+        let error = match apply_managed_encryption_material(
+            "bucket",
+            "object",
+            ServerSideEncryption::from_static(ServerSideEncryption::AES256),
+            None,
+            None,
+            0,
+        )
+        .await
+        {
+            Ok(_) => panic!("provider unavailability must fail managed encryption"),
+            Err(error) => error,
+        };
+        assert_eq!(error.code, S3ErrorCode::ServiceUnavailable);
+
+        let metadata = HashMap::from([
+            ("x-amz-server-side-encryption".to_string(), ServerSideEncryption::AES256.to_string()),
+            (
+                INTERNAL_ENCRYPTION_KEY_HEADER.to_string(),
+                BASE64_STANDARD.encode(b"local-provider-format"),
+            ),
+            (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), BASE64_STANDARD.encode([0x14; 12])),
+            (INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "test-key-id".to_string()),
+        ]);
+        let error = match apply_managed_decryption_material("bucket", "object", &metadata).await {
+            Ok(_) => panic!("provider unavailability must fail managed decryption"),
+            Err(error) => error,
+        };
+        assert_eq!(error.code, S3ErrorCode::ServiceUnavailable);
+
+        reset_sse_dek_provider();
+    }
+
     #[tokio::test]
     async fn test_kms_sse_dek_provider_uses_latest_reconfigured_service() {
         use rustfs_kms::config::KmsConfig;
@@ -4159,6 +4424,24 @@ mod tests {
             .expect("provider should resolve the latest reconfigured service");
 
         manager.stop().await.expect("kms service should stop cleanly");
+        let generate_error = provider
+            .generate_sse_dek(&context, "second-key")
+            .await
+            .expect_err("stopped KMS must reject data-key generation");
+        assert_eq!(generate_error.code, S3ErrorCode::ServiceUnavailable);
+        let decrypt_error = provider
+            .decrypt_sse_dek(b"{}", "second-key", &context)
+            .await
+            .expect_err("stopped KMS must reject data-key decryption");
+        assert_eq!(decrypt_error.code, S3ErrorCode::ServiceUnavailable);
+        #[cfg(feature = "rio-v2")]
+        {
+            let legacy_error = provider
+                .decrypt_legacy_sse_dek(b"{}", "second-key", &context)
+                .await
+                .expect_err("stopped KMS must reject legacy data-key decryption");
+            assert_eq!(legacy_error.code, S3ErrorCode::ServiceUnavailable);
+        }
     }
 
     #[test]