fix(kms): unify persisted SSE data key envelopes (#5343)

* feat(kms): implement secure handling of static KMS secret keys and enhance encryption context validation

* feat: enhance local SSE DEK handling with JSON envelope format and versioning

rustfs/src/admin/handlers/kms_dynamic.rs

@@ -36,6 +36,8 @@ use tracing::{error, info, instrument, warn};
 
 /// Path to store KMS configuration in the cluster metadata
 const KMS_CONFIG_PATH: &str = "config/kms_config.json";
+const STATIC_KMS_LOCAL_CONFIG_REQUIRED: &str =
+    "Static KMS must be configured through RUSTFS_KMS_STATIC_SECRET_KEY or RUSTFS_KMS_STATIC_SECRET_KEY_FILE";
 const LOG_COMPONENT_ADMIN: &str = "admin";
 const LOG_SUBSYSTEM_KMS: &str = "kms";
 const EVENT_ADMIN_KMS_DYNAMIC_STATE: &str = "admin_kms_dynamic_state";
@@ -106,9 +108,25 @@ fn normalize_configure_request_auth(
     Ok(())
 }
 
+fn ensure_kms_config_persistable(config: &KmsConfig) -> Result<(), String> {
+    if matches!(&config.backend_config, rustfs_kms::BackendConfig::Static(_)) {
+        return Err(STATIC_KMS_LOCAL_CONFIG_REQUIRED.to_string());
+    }
+    Ok(())
+}
+
+fn ensure_kms_request_persistable(request: &ConfigureKmsRequest) -> Result<(), String> {
+    if matches!(request, ConfigureKmsRequest::Static(_)) {
+        return Err(STATIC_KMS_LOCAL_CONFIG_REQUIRED.to_string());
+    }
+    Ok(())
+}
+
 /// Save KMS configuration to cluster storage
 #[instrument(skip(config))]
 async fn save_kms_config(config: &KmsConfig) -> Result<(), String> {
+    ensure_kms_config_persistable(config)?;
+
     let context = current_app_context();
     let Some(store) = current_object_store_handle_for_context(context.as_deref()) else {
         return Err("Storage layer not initialized".to_string());
@@ -332,12 +350,16 @@ impl Operation for ConfigureKmsHandler {
         );
 
         let service_manager = kms_service_manager_from_context();
-        let existing_config = service_manager.get_config().await;
+        let existing_config = service_manager.get_redacted_config().await;
 
         if let Err(e) = normalize_configure_request_auth(&mut configure_request, existing_config.as_ref()) {
             return Ok(S3Response::new((StatusCode::BAD_REQUEST, Body::from(e))));
         }
 
+        if let Err(e) = ensure_kms_request_persistable(&configure_request) {
+            return Ok(S3Response::new((StatusCode::BAD_REQUEST, Body::from(e))));
+        }
+
         // Convert request to KmsConfig
         let kms_config = configure_request.to_kms_config();
 
@@ -753,7 +775,7 @@ impl Operation for GetKmsStatusHandler {
         let service_manager = kms_service_manager_from_context();
 
         let status = service_manager.get_status().await;
-        let config = service_manager.get_config().await;
+        let config = service_manager.get_redacted_config().await;
 
         // Get backend type and health status
         let backend_type = config.as_ref().map(|c| c.backend.clone());
@@ -873,12 +895,16 @@ impl Operation for ReconfigureKmsHandler {
         );
 
         let service_manager = kms_service_manager_from_context();
-        let existing_config = service_manager.get_config().await;
+        let existing_config = service_manager.get_redacted_config().await;
 
         if let Err(e) = normalize_configure_request_auth(&mut configure_request, existing_config.as_ref()) {
             return Ok(S3Response::new((StatusCode::BAD_REQUEST, Body::from(e))));
         }
 
+        if let Err(e) = ensure_kms_request_persistable(&configure_request) {
+            return Ok(S3Response::new((StatusCode::BAD_REQUEST, Body::from(e))));
+        }
+
         // Convert request to KmsConfig
         let kms_config = configure_request.to_kms_config();
 
@@ -960,7 +986,7 @@ impl Operation for ReconfigureKmsHandler {
 
 #[cfg(test)]
 mod tests {
-    use super::{decode_persisted_kms_config, kms_configure_actions, kms_service_control_actions};
+    use super::{decode_persisted_kms_config, ensure_kms_config_persistable, kms_configure_actions, kms_service_control_actions};
     use rustfs_policy::policy::action::{Action, AdminAction, KmsAction};
     use tempfile::TempDir;
 
@@ -1046,4 +1072,16 @@ mod tests {
         assert!(!config.allow_insecure_dev_defaults);
         assert!(config.validate().is_ok());
     }
+
+    #[test]
+    fn static_kms_config_is_not_persisted_with_cluster_configuration() {
+        use base64::Engine as _;
+
+        let config = rustfs_kms::KmsConfig::static_kms(
+            "static-key".to_string(),
+            base64::engine::general_purpose::STANDARD.encode([0x5au8; 32]),
+        );
+
+        assert!(ensure_kms_config_persistable(&config).is_err());
+    }
 }

rustfs/src/admin/handlers/kms_management.rs

@@ -193,7 +193,7 @@ impl Operation for KmsStatusHandler {
             hit_count: hits,
             miss_count: misses,
         });
-        let config = kms_service_manager_from_context().get_config().await;
+        let config = kms_service_manager_from_context().get_redacted_config().await;
 
         let response = KmsStatusResponse {
             backend_type: config
@@ -243,7 +243,7 @@ impl Operation for KmsConfigHandler {
         };
 
         let config = kms_service_manager_from_context()
-            .get_config()
+            .get_redacted_config()
             .await
             .ok_or_else(|| s3_error!(InternalError, "KMS config not available"))?;
 

rustfs/src/error.rs

@@ -245,6 +245,21 @@ impl From<StorageError> for ApiError {
             };
         }
 
+        if let StorageError::Io(ref io_err) = err
+            && matches!(
+                io_err
+                    .get_ref()
+                    .and_then(|inner| inner.downcast_ref::<rustfs_kms::KmsError>()),
+                Some(rustfs_kms::KmsError::BackendError { .. })
+            )
+        {
+            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,
@@ -487,6 +502,14 @@ mod tests {
         assert_eq!(api_error.message, "The service is unavailable. Please retry.");
     }
 
+    #[test]
+    fn test_kms_backend_unavailable_maps_to_retryable_error() {
+        let api_error = ApiError::from(StorageError::other(rustfs_kms::KmsError::backend_error("Vault connection refused")));
+
+        assert_eq!(api_error.code, S3ErrorCode::ServiceUnavailable);
+        assert_eq!(api_error.message, "The service is unavailable. Please retry.");
+    }
+
     #[test]
     fn test_unknown_authoritative_quota_usage_maps_to_retryable_error() {
         let api_error = ApiError::from(QuotaError::UsageUnavailable {
@@ -497,6 +520,16 @@ mod tests {
         assert_eq!(api_error.message, "The service is unavailable. Please retry.");
     }
 
+    #[test]
+    fn test_kms_cryptographic_error_is_not_retryable() {
+        let api_error = ApiError::from(StorageError::other(rustfs_kms::KmsError::cryptographic_error(
+            "decrypt",
+            "authentication failed",
+        )));
+
+        assert_eq!(api_error.code, S3ErrorCode::InternalError);
+    }
+
     #[test]
     fn test_api_error_from_storage_error_mappings() {
         let test_cases = vec![

rustfs/src/storage/sse.rs

@@ -91,6 +91,7 @@ use rustfs_kms::{DataKey, KmsUnavailableError, is_data_key_envelope, types::Obje
 use rustfs_utils::get_env_opt_str;
 use s3s::S3ErrorCode;
 use s3s::dto::ServerSideEncryption;
+use serde::{Deserialize, Serialize};
 #[cfg(feature = "rio-v2")]
 use sha2::Sha256;
 use std::collections::HashMap;
@@ -1885,6 +1886,10 @@ struct KmsSseDekProvider {
     service_manager: Option<Arc<rustfs_kms::KmsServiceManager>>,
 }
 
+fn kms_operation_error(error: rustfs_kms::KmsError) -> ApiError {
+    ApiError::from(StorageError::other(error))
+}
+
 impl KmsSseDekProvider {
     /// Create a new KMS-backed provider
     pub async fn new() -> Result<Self, ApiError> {
@@ -1936,7 +1941,7 @@ impl SseDekProvider for KmsSseDekProvider {
         let (data_key, encrypted_data_key) = service
             .create_data_key(&kms_key_option, context)
             .await
-            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to create data key: {}", e))))?;
+            .map_err(kms_operation_error)?;
 
         Ok((data_key, encrypted_data_key))
     }
@@ -1954,7 +1959,7 @@ impl SseDekProvider for KmsSseDekProvider {
         let data_key = service
             .decrypt_data_key(encrypted_dek, context)
             .await
-            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decrypt data key: {}", e))))?;
+            .map_err(kms_operation_error)?;
 
         Ok(data_key.plaintext_key)
     }
@@ -1973,7 +1978,7 @@ impl SseDekProvider for KmsSseDekProvider {
         let data_key = service
             .decrypt_legacy_data_key(encrypted_dek)
             .await
-            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decrypt legacy data key: {e}"))))?;
+            .map_err(kms_operation_error)?;
 
         Ok(data_key.plaintext_key)
     }
@@ -1998,6 +2003,16 @@ pub(crate) struct LocalSseDekProvider {
     master_key: [u8; 32],
 }
 
+const LOCAL_SSE_DEK_FORMAT_VERSION: u8 = 1;
+
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+struct LocalSseDekEnvelope<'a> {
+    version: u8,
+    nonce: &'a str,
+    ciphertext: &'a str,
+}
+
 /// Test-only alias so existing test code that references `TestSseDekProvider`
 /// continues to compile without changes.
 #[cfg(test)]
@@ -2099,22 +2114,51 @@ impl LocalSseDekProvider {
             .encrypt(&nonce, dek.as_slice())
             .map_err(|_| ApiError::from(StorageError::other("Failed to encrypt DEK")))?;
 
-        // nonce:ciphertext
-        Ok(format!("{}:{}", BASE64_STANDARD.encode(nonce), BASE64_STANDARD.encode(ciphertext)))
+        let nonce = BASE64_STANDARD.encode(nonce);
+        let ciphertext = BASE64_STANDARD.encode(ciphertext);
+        serde_json::to_string(&LocalSseDekEnvelope {
+            version: LOCAL_SSE_DEK_FORMAT_VERSION,
+            nonce: &nonce,
+            ciphertext: &ciphertext,
+        })
+        .map_err(|e| ApiError::from(StorageError::other(format!("Failed to serialize encrypted DEK: {e}"))))
     }
 
     // Simple decryption of DEK
     pub(crate) fn decrypt_dek(encrypted_dek: &str, cmk_value: [u8; 32]) -> Result<[u8; 32], ApiError> {
-        let parts: Vec<&str> = encrypted_dek.split(':').collect();
-        if parts.len() != 2 {
-            return Err(ApiError::from(StorageError::other("Invalid encrypted DEK format")));
-        }
-
+        let envelope = serde_json::from_str::<LocalSseDekEnvelope<'_>>(encrypted_dek);
+        let (nonce, ciphertext) = match envelope {
+            Ok(envelope) => {
+                if envelope.version != LOCAL_SSE_DEK_FORMAT_VERSION {
+                    return Err(ApiError::from(StorageError::other(format!(
+                        "Unsupported encrypted DEK format version: {}",
+                        envelope.version
+                    ))));
+                }
+                (envelope.nonce, envelope.ciphertext)
+            }
+            Err(json_error) if encrypted_dek.trim_start().starts_with('{') => {
+                return Err(ApiError::from(StorageError::other(format!(
+                    "Invalid encrypted DEK JSON format: {json_error}"
+                ))));
+            }
+            Err(_) => {
+                // DEPRECATED: read-only compatibility for persisted colon-delimited DEKs.
+                // RUSTFS_COMPAT_TODO(sse-local-dek-json-v1): Remove after all supported upgrades have rewritten legacy DEKs.
+                let Some((nonce, ciphertext)) = encrypted_dek.split_once(':') else {
+                    return Err(ApiError::from(StorageError::other("Invalid encrypted DEK format")));
+                };
+                if ciphertext.contains(':') {
+                    return Err(ApiError::from(StorageError::other("Invalid encrypted DEK format")));
+                }
+                (nonce, ciphertext)
+            }
+        };
         let nonce_vec = BASE64_STANDARD
-            .decode(parts[0])
+            .decode(nonce)
             .map_err(|_| ApiError::from(StorageError::other("Invalid nonce format")))?;
         let ciphertext = BASE64_STANDARD
-            .decode(parts[1])
+            .decode(ciphertext)
             .map_err(|_| ApiError::from(StorageError::other("Invalid ciphertext format")))?;
 
         let key = Key::<Aes256Gcm>::from(cmk_value);
@@ -2596,10 +2640,10 @@ mod tests {
         SseDekProvider, SsecParams, StorageError, TestSseDekProvider, apply_managed_decryption_material,
         apply_managed_encryption_material, 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,
+        kms_operation_error, 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::{
@@ -2622,6 +2666,15 @@ mod tests {
         assert!(super::parse_simple_sse_cmk(&zero).is_err());
     }
 
+    #[test]
+    fn kms_operation_errors_preserve_retryability_classification() {
+        let unavailable = kms_operation_error(rustfs_kms::KmsError::backend_error("connection refused"));
+        let corrupt = kms_operation_error(rustfs_kms::KmsError::cryptographic_error("decrypt", "authentication failed"));
+
+        assert_eq!(unavailable.code, S3ErrorCode::ServiceUnavailable);
+        assert_eq!(corrupt.code, S3ErrorCode::InternalError);
+    }
+
     #[test]
     fn parse_simple_sse_cmk_accepts_valid_32_byte_key() {
         let mut key = [0u8; 32];
@@ -4023,17 +4076,25 @@ mod tests {
     }
 
     #[test]
-    fn test_encrypt_dek_uses_random_nonce_prefixes() {
+    fn test_encrypt_dek_writes_json_with_random_nonces() {
         let dek = [0x11u8; 32];
         let cmk = [0x22u8; 32];
 
         let encrypted_a = TestSseDekProvider::encrypt_dek(dek, cmk).expect("first DEK wrap should succeed");
         let encrypted_b = TestSseDekProvider::encrypt_dek(dek, cmk).expect("second DEK wrap should succeed");
 
-        let nonce_a = encrypted_a.split(':').next().expect("wrapped DEK should contain nonce");
-        let nonce_b = encrypted_b.split(':').next().expect("wrapped DEK should contain nonce");
+        let envelope_a: super::LocalSseDekEnvelope =
+            serde_json::from_str(&encrypted_a).expect("first wrapped DEK should be a JSON envelope");
+        let envelope_b: super::LocalSseDekEnvelope =
+            serde_json::from_str(&encrypted_b).expect("second wrapped DEK should be a JSON envelope");
 
-        assert_ne!(nonce_a, nonce_b, "each DEK wrap should use a distinct nonce prefix");
+        assert_eq!(envelope_a.version, super::LOCAL_SSE_DEK_FORMAT_VERSION);
+        assert_eq!(envelope_b.version, super::LOCAL_SSE_DEK_FORMAT_VERSION);
+        assert_ne!(envelope_a.nonce, envelope_b.nonce, "each DEK wrap should use a distinct nonce");
+        assert_eq!(
+            TestSseDekProvider::decrypt_dek(&encrypted_a, cmk).expect("first JSON envelope should decrypt"),
+            dek
+        );
     }
 
     #[test]
@@ -4051,6 +4112,20 @@ mod tests {
         assert_eq!(decrypted, dek);
     }
 
+    #[test]
+    fn test_decrypt_dek_rejects_unknown_json_version() {
+        let envelope = serde_json::json!({
+            "version": super::LOCAL_SSE_DEK_FORMAT_VERSION + 1,
+            "nonce": BASE64_STANDARD.encode([0u8; 12]),
+            "ciphertext": BASE64_STANDARD.encode([0u8; 48]),
+        })
+        .to_string();
+
+        let error = TestSseDekProvider::decrypt_dek(&envelope, [0x55u8; 32])
+            .expect_err("unknown JSON envelope versions must fail closed");
+        assert!(error.message.contains("Unsupported encrypted DEK format version"));
+    }
+
     #[tokio::test]
     async fn test_sse_encryption_fails_closed_without_local_sse_master_key() {
         let _guard = lock_sse_test_state().await;