fix(sse): separate SSE-S3 and KMS key providers

crates/e2e_test/src/multipart_auth_test.rs

@@ -1068,7 +1068,8 @@ async fn test_anonymous_post_object_uses_bucket_default_sse_s3() -> Result<(), B
 
 #[tokio::test]
 #[serial]
-async fn test_anonymous_post_object_uses_bucket_default_sse_kms() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+async fn test_anonymous_post_object_bucket_default_sse_kms_requires_kms() -> Result<(), Box<dyn std::error::Error + Send + Sync>>
+{
     init_logging();
 
     let mut env = RustFSTestEnvironment::new().await?;
@@ -1128,14 +1129,23 @@ async fn test_anonymous_post_object_uses_bucket_default_sse_kms() -> Result<(),
         .send()
         .await?;
 
-    assert_eq!(post_resp.status(), reqwest::StatusCode::NO_CONTENT);
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.server_side_encryption().map(|value| value.as_str()), Some("aws:kms"));
-
-    let uploaded = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = uploaded.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+    assert_eq!(status, reqwest::StatusCode::BAD_REQUEST);
+    assert!(
+        response_body.contains("configured KMS service"),
+        "unexpected response body: {response_body}"
+    );
+    assert!(
+        admin_client
+            .head_object()
+            .bucket(bucket)
+            .key(object_key)
+            .send()
+            .await
+            .is_err(),
+        "failed SSE-KMS POST must not create an object"
+    );
 
     Ok(())
 }

crates/ecstore/src/api/mod.rs

@@ -398,6 +398,10 @@ pub mod set_disk {
     pub use crate::set_disk::{DEFAULT_READ_BUFFER_SIZE, SetDisks, get_lock_acquire_timeout, is_valid_storage_class};
 }
 
+pub mod sse {
+    pub use crate::sse::{ManagedDekProvider, ManagedSseScheme, managed_dek_provider};
+}
+
 pub mod store_list {
     pub use crate::store::list_objects::{ListPathOptions, max_keys_plus_one};
 }

crates/ecstore/src/lib.rs

@@ -49,6 +49,7 @@ mod object_api;
 mod runtime;
 mod services;
 mod set_disk;
+mod sse;
 mod storage_api_contracts;
 mod store;
 

crates/ecstore/src/object_api/readers.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use super::*;
+use crate::sse::{ManagedDekProvider, ManagedSseScheme, managed_dek_provider as classify_managed_dek_provider};
 #[cfg(feature = "rio-v2")]
 use aes_gcm::aead::Payload;
 use aes_gcm::{
@@ -26,7 +27,11 @@ use chacha20poly1305::ChaCha20Poly1305;
 use hmac::{Hmac, Mac};
 use md5::{Digest, Md5};
 use rustfs_kms::types::ObjectEncryptionContext;
-use rustfs_utils::http::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
+#[cfg(feature = "rio-v2")]
+use rustfs_kms::{MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER, RUSTFS_ENCRYPTION_CONTEXT_HEADER, decode_managed_kms_context};
+use rustfs_utils::http::{
+    AMZ_SERVER_SIDE_ENCRYPTION, SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER, get_consistent_metadata_value,
+};
 use rustfs_utils::path::path_join_buf;
 #[cfg(feature = "rio-v2")]
 use serde::Deserialize;
@@ -39,11 +44,11 @@ use crate::io_support::rio::Index;
 
 const INTERNAL_ENCRYPTION_KEY_ID_HEADER: &str = "x-rustfs-encryption-key-id";
 const INTERNAL_ENCRYPTION_KEY_HEADER: &str = "x-rustfs-encryption-key";
-const INTERNAL_ENCRYPTION_CONTEXT_HEADER: &str = "x-rustfs-encryption-context";
 const INTERNAL_ENCRYPTION_IV_HEADER: &str = "x-rustfs-encryption-iv";
 const INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER: &str = "x-rustfs-encryption-original-size";
 const SSEC_ORIGINAL_SIZE_HEADER: &str = "x-amz-server-side-encryption-customer-original-size";
 const DEFAULT_SSE_ALGORITHM: &str = "AES256";
+const SSE_KMS_ALGORITHM: &str = "aws:kms";
 #[cfg(feature = "rio-v2")]
 const DARE_PAYLOAD_SIZE: i64 = 64 * 1024;
 #[cfg(feature = "rio-v2")]
@@ -60,8 +65,6 @@ const MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER: &str = "X-Minio-Internal-Serv
 #[cfg(feature = "rio-v2")]
 const MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER: &str = "X-Minio-Internal-Server-Side-Encryption-S3-Kms-Sealed-Key";
 #[cfg(feature = "rio-v2")]
-const MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER: &str = "X-Minio-Internal-Server-Side-Encryption-Context";
-#[cfg(feature = "rio-v2")]
 const MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER: &str = "X-Minio-Internal-Server-Side-Encryption-Sealed-Key";
 #[cfg(feature = "rio-v2")]
 const MINIO_INTERNAL_ENCRYPTION_SEAL_ALGORITHM: &str = "DAREv2-HMAC-SHA256";
@@ -1381,31 +1384,30 @@ async fn resolve_managed_material(bucket: &str, object: &str, metadata: &HashMap
 
     let kms_key_id = metadata_get(&normalized_metadata, INTERNAL_ENCRYPTION_KEY_ID_HEADER).unwrap_or("default");
     #[cfg(feature = "rio-v2")]
-    let kms_context = metadata_get(&normalized_metadata, INTERNAL_ENCRYPTION_CONTEXT_HEADER)
-        .map(|value| {
-            serde_json::from_str::<HashMap<String, String>>(value)
-                .map_err(|e| Error::other(format!("failed to parse managed KMS context: {e}")))
-        })
-        .transpose()?;
+    let kms_context = decode_managed_kms_context(metadata).map_err(|err| Error::other(err.to_string()))?;
     #[cfg(not(feature = "rio-v2"))]
     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 {
-        #[cfg(feature = "rio-v2")]
-        let data_key = if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
-            service.decrypt_legacy_data_key(&encrypted_dek).await
-        } else {
-            service.decrypt_data_key(&encrypted_dek, &object_context).await
-        };
-        #[cfg(not(feature = "rio-v2"))]
-        let data_key = service.decrypt_data_key(&encrypted_dek, &object_context).await;
+    let decrypted_key = match managed_dek_provider(metadata, &encrypted_dek)? {
+        ManagedDekProvider::LocalSseS3 => decrypt_local_sse_dek(&encrypted_dek, kms_key_id, &object_context)?,
+        ManagedDekProvider::Kms => {
+            let service = crate::runtime::sources::object_encryption_service()
+                .await
+                .ok_or_else(|| Error::other("KMS encryption service is required to decrypt this object"))?;
+            #[cfg(feature = "rio-v2")]
+            let data_key = if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
+                service.decrypt_legacy_data_key(&encrypted_dek).await
+            } else {
+                service.decrypt_data_key(&encrypted_dek, &object_context).await
+            };
+            #[cfg(not(feature = "rio-v2"))]
+            let data_key = service.decrypt_data_key(&encrypted_dek, &object_context).await;
 
-        data_key
-            .map_err(|e| Error::other(format!("failed to decrypt managed data key: {e}")))?
-            .plaintext_key
-    } else {
-        decrypt_local_sse_dek(&encrypted_dek, kms_key_id, &object_context)?
+            data_key
+                .map_err(|e| Error::other(format!("failed to decrypt managed data key: {e}")))?
+                .plaintext_key
+        }
     };
 
     #[cfg(feature = "rio-v2")]
@@ -1436,6 +1438,19 @@ async fn resolve_managed_material(bucket: &str, object: &str, metadata: &HashMap
     })
 }
 
+fn managed_dek_provider(metadata: &HashMap<String, String>, encrypted_dek: &[u8]) -> Result<ManagedDekProvider> {
+    let algorithm = get_consistent_metadata_value(metadata, AMZ_SERVER_SIDE_ENCRYPTION)
+        .map_err(|_| Error::other(format!("conflicting managed encryption metadata for {AMZ_SERVER_SIDE_ENCRYPTION}")))?;
+    let scheme = match algorithm {
+        Some(SSE_KMS_ALGORITHM) => ManagedSseScheme::SseKms,
+        Some(DEFAULT_SSE_ALGORITHM) | None => ManagedSseScheme::SseS3,
+        Some(algorithm) => return Err(Error::other(format!("unsupported stored server-side encryption {algorithm}"))),
+    };
+    // RUSTFS_COMPAT_TODO(rustfs-5063): Keep legacy SSE-S3 KMS envelopes readable. Remove after SSE-S3 migration rewraps every referenced legacy DEK.
+    let has_kms_envelope = rustfs_kms::is_data_key_envelope(encrypted_dek);
+    Ok(classify_managed_dek_provider(scheme, has_kms_envelope))
+}
+
 fn normalize_managed_metadata(metadata: &HashMap<String, String>) -> HashMap<String, String> {
     #[cfg(feature = "rio-v2")]
     {
@@ -1460,13 +1475,13 @@ fn normalize_managed_metadata(metadata: &HashMap<String, String>) -> HashMap<Str
             normalized.insert(INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), value.to_string());
         }
 
-        if metadata_get(&normalized, INTERNAL_ENCRYPTION_CONTEXT_HEADER).is_none()
+        if metadata_get(&normalized, RUSTFS_ENCRYPTION_CONTEXT_HEADER).is_none()
             && let Some(value) = metadata_get(metadata, MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER)
             && let Ok(decoded) = BASE64_STANDARD.decode(value)
             && let Ok(context) = serde_json::from_slice::<HashMap<String, String>>(&decoded)
             && let Ok(encoded) = serde_json::to_string(&context)
         {
-            normalized.insert(INTERNAL_ENCRYPTION_CONTEXT_HEADER.to_string(), encoded);
+            normalized.insert(RUSTFS_ENCRYPTION_CONTEXT_HEADER.to_string(), encoded);
         }
 
         normalized
@@ -1624,15 +1639,21 @@ fn marshal_minio_kms_context(context: &HashMap<String, String>) -> Vec<u8> {
 }
 
 fn local_sse_master_key() -> Result<[u8; 32]> {
+    #[cfg(test)]
     if let Some(key) = decode_master_key_env("__RUSTFS_SSE_SIMPLE_CMK")? {
         return Ok(key);
     }
 
     if let Some(key) = decode_master_key_env("RUSTFS_SSE_S3_MASTER_KEY")? {
+        if key == [0u8; 32] {
+            return Err(Error::other("RUSTFS_SSE_S3_MASTER_KEY must not be an all-zero key"));
+        }
         return Ok(key);
     }
 
-    Ok([0u8; 32])
+    Err(Error::other(
+        "SSE-S3 requires RUSTFS_SSE_S3_MASTER_KEY to decrypt locally managed objects",
+    ))
 }
 
 fn decode_master_key_env(name: &str) -> Result<Option<[u8; 32]>> {
@@ -2108,6 +2129,57 @@ mod tests {
         format!("{}:{}", BASE64_STANDARD.encode(nonce), BASE64_STANDARD.encode(ciphertext))
     }
 
+    #[test]
+    fn managed_dek_provider_routes_by_algorithm_and_persisted_envelope() {
+        let local_dek = encrypt_managed_dek_for_test([0x24; 32], [0x42; 32]);
+        let kms_dek = serde_json::to_vec(&serde_json::json!({
+            "key_id": "legacy-data-key",
+            "master_key_id": "legacy-master-key",
+            "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"
+        }))
+        .expect("legacy KMS envelope should serialize");
+
+        assert_eq!(
+            managed_dek_provider(
+                &HashMap::from([(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), DEFAULT_SSE_ALGORITHM.to_string())]),
+                local_dek.as_bytes(),
+            )
+            .expect("SSE-S3 local DEK should be classified"),
+            ManagedDekProvider::LocalSseS3
+        );
+        assert_eq!(
+            managed_dek_provider(
+                &HashMap::from([(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), SSE_KMS_ALGORITHM.to_string())]),
+                local_dek.as_bytes(),
+            )
+            .expect("SSE-KMS should be classified by its stored algorithm"),
+            ManagedDekProvider::Kms
+        );
+        assert_eq!(
+            managed_dek_provider(
+                &HashMap::from([(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), DEFAULT_SSE_ALGORITHM.to_string())]),
+                &kms_dek,
+            )
+            .expect("legacy SSE-S3 KMS envelope should be classified"),
+            ManagedDekProvider::Kms
+        );
+        assert_eq!(
+            managed_dek_provider(&HashMap::new(), &kms_dek).expect("legacy KMS envelope without algorithm should be classified"),
+            ManagedDekProvider::Kms
+        );
+        assert!(
+            managed_dek_provider(
+                &HashMap::from([(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "unsupported".to_string())]),
+                local_dek.as_bytes(),
+            )
+            .is_err()
+        );
+    }
+
     #[cfg(feature = "rio-v2")]
     fn seal_managed_s3_object_key_for_test(
         bucket: &str,
@@ -2622,12 +2694,12 @@ mod tests {
         async_with_vars(
             [
                 ("__RUSTFS_SSE_SIMPLE_CMK", None::<String>),
-                ("RUSTFS_SSE_S3_MASTER_KEY", Some(BASE64_STANDARD.encode([0u8; 32]))),
+                ("RUSTFS_SSE_S3_MASTER_KEY", Some(BASE64_STANDARD.encode([0x33u8; 32]))),
             ],
             async {
                 let plaintext = b"managed-local-fallback".to_vec();
                 let data_key = [0x22; 32];
-                let encrypted_dek = encrypt_managed_dek_for_test(data_key, [0u8; 32]);
+                let encrypted_dek = encrypt_managed_dek_for_test(data_key, [0x33; 32]);
                 let bucket = "bucket";
                 let object = "managed-local-fallback";
 

crates/ecstore/src/sse/mod.rs

@@ -0,0 +1,46 @@
+// 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.
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ManagedSseScheme {
+    SseS3,
+    SseKms,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ManagedDekProvider {
+    LocalSseS3,
+    Kms,
+}
+
+pub fn managed_dek_provider(scheme: ManagedSseScheme, has_kms_envelope: bool) -> ManagedDekProvider {
+    match scheme {
+        ManagedSseScheme::SseKms => ManagedDekProvider::Kms,
+        ManagedSseScheme::SseS3 if has_kms_envelope => ManagedDekProvider::Kms,
+        ManagedSseScheme::SseS3 => ManagedDekProvider::LocalSseS3,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn managed_sse_routing_is_determined_by_scheme_and_envelope() {
+        assert_eq!(managed_dek_provider(ManagedSseScheme::SseS3, false), ManagedDekProvider::LocalSseS3);
+        assert_eq!(managed_dek_provider(ManagedSseScheme::SseS3, true), ManagedDekProvider::Kms);
+        assert_eq!(managed_dek_provider(ManagedSseScheme::SseKms, false), ManagedDekProvider::Kms);
+        assert_eq!(managed_dek_provider(ManagedSseScheme::SseKms, true), ManagedDekProvider::Kms);
+    }
+}

crates/ecstore/tests/minio_generated_read_test.rs

@@ -133,8 +133,8 @@ async fn read_fixture_plaintext(encrypted: Vec<u8>, object_info: ObjectInfo, kms
 
     async_with_vars(
         [
-            ("__RUSTFS_SSE_SIMPLE_CMK", Some(kms_key_b64)),
-            ("RUSTFS_SSE_S3_MASTER_KEY", None::<String>),
+            ("RUSTFS_SSE_S3_MASTER_KEY", Some(kms_key_b64)),
+            ("__RUSTFS_SSE_SIMPLE_CMK", None::<String>),
         ],
         async move {
             let (mut reader, offset, length) = GetObjectReader::new(

crates/kms/Cargo.toml

@@ -56,7 +56,7 @@ moka = { workspace = true, features = ["future"] }
 # Additional dependencies
 md5 = { workspace = true }
 arc-swap = { workspace = true }
-rustfs-utils = { workspace = true }
+rustfs-utils = { workspace = true, features = ["http"] }
 rustfs-security-governance = { workspace = true }
 
 # HTTP client for Vault

crates/kms/src/encryption/dek.rs

@@ -44,6 +44,23 @@ pub struct DataKeyEnvelope {
     pub created_at: Zoned,
 }
 
+/// Return whether bytes contain a complete RustFS KMS data-key envelope.
+pub fn is_data_key_envelope(ciphertext: &[u8]) -> bool {
+    const MAX_ENVELOPE_SIZE: usize = 64 * 1024;
+
+    if ciphertext.is_empty() || ciphertext.len() > MAX_ENVELOPE_SIZE {
+        return false;
+    }
+
+    serde_json::from_slice::<DataKeyEnvelope>(ciphertext).is_ok_and(|envelope| {
+        !envelope.key_id.trim().is_empty()
+            && !envelope.master_key_id.trim().is_empty()
+            && envelope.key_spec == "AES_256"
+            && !envelope.encrypted_key.is_empty()
+            && (envelope.nonce.is_empty() || envelope.nonce.len() == 12)
+    })
+}
+
 /// Trait for encrypting and decrypting data encryption keys (DEK)
 ///
 /// This trait abstracts the encryption operations used to protect
@@ -293,6 +310,24 @@ mod tests {
         assert_eq!(deserialized.key_id, envelope.key_id);
         assert_eq!(deserialized.master_key_id, envelope.master_key_id);
         assert_eq!(deserialized.encrypted_key, envelope.encrypted_key);
+        assert!(is_data_key_envelope(&serialized));
+        assert!(!is_data_key_envelope(b"not-a-kms-envelope"));
+        let mut boundary_envelope = serialized;
+        boundary_envelope.resize(64 * 1024, b' ');
+        assert!(is_data_key_envelope(&boundary_envelope));
+        boundary_envelope.push(b' ');
+        assert!(!is_data_key_envelope(&boundary_envelope));
+
+        let mut invalid = serde_json::to_value(&envelope).expect("Envelope should convert to JSON");
+        invalid["key_spec"] = serde_json::Value::String("AES_128".to_string());
+        assert!(!is_data_key_envelope(
+            &serde_json::to_vec(&invalid).expect("Invalid envelope should serialize")
+        ));
+        invalid["key_spec"] = serde_json::Value::String("AES_256".to_string());
+        invalid["unknown"] = serde_json::Value::Bool(true);
+        assert!(is_data_key_envelope(
+            &serde_json::to_vec(&invalid).expect("Envelope with unknown field should serialize")
+        ));
     }
 
     #[tokio::test]

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/lib.rs

@@ -70,6 +70,7 @@ mod cache;
 pub mod config;
 mod encryption;
 mod error;
+mod managed_context;
 pub mod manager;
 pub mod service;
 pub mod service_manager;
@@ -83,7 +84,11 @@ pub use api_types::{
     TagKeyRequest, TagKeyResponse, UntagKeyRequest, UntagKeyResponse, UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse,
 };
 pub use config::*;
+pub use encryption::is_data_key_envelope;
 pub use error::{KmsError, Result};
+pub use managed_context::{
+    MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER, RUSTFS_ENCRYPTION_CONTEXT_HEADER, decode_managed_kms_context,
+};
 pub use manager::KmsManager;
 pub use service::{DataKey, ObjectEncryptionService};
 pub use service_manager::{

crates/kms/src/managed_context.rs

@@ -0,0 +1,110 @@
+// 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.
+
+use crate::{KmsError, Result};
+use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD};
+use rustfs_utils::http::get_consistent_metadata_value;
+use std::collections::HashMap;
+
+pub const RUSTFS_ENCRYPTION_CONTEXT_HEADER: &str = "x-rustfs-encryption-context";
+pub const MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER: &str = "X-Minio-Internal-Server-Side-Encryption-Context";
+
+pub fn decode_managed_kms_context(metadata: &HashMap<String, String>) -> Result<Option<HashMap<String, String>>> {
+    let minio_context = consistent_value(metadata, MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER)?
+        .map(|context| {
+            let decoded = BASE64_STANDARD
+                .decode(context)
+                .map_err(|err| KmsError::serialization_error(format!("Failed to decode MinIO KMS context: {err}")))?;
+            serde_json::from_slice(&decoded)
+                .map_err(|err| KmsError::serialization_error(format!("Failed to parse MinIO KMS context: {err}")))
+        })
+        .transpose()?;
+    let rustfs_context = consistent_value(metadata, RUSTFS_ENCRYPTION_CONTEXT_HEADER)?
+        .map(|context| {
+            serde_json::from_str(context)
+                .map_err(|err| KmsError::serialization_error(format!("Failed to parse RustFS KMS context: {err}")))
+        })
+        .transpose()?;
+
+    match (minio_context, rustfs_context) {
+        (Some(minio), Some(rustfs)) if minio != rustfs => {
+            Err(KmsError::context_mismatch("Conflicting RustFS and MinIO KMS contexts"))
+        }
+        (Some(context), _) | (_, Some(context)) => Ok(Some(context)),
+        (None, None) => Ok(None),
+    }
+}
+
+fn consistent_value<'a>(metadata: &'a HashMap<String, String>, name: &str) -> Result<Option<&'a str>> {
+    get_consistent_metadata_value(metadata, name)
+        .map_err(|_| KmsError::validation_error(format!("Conflicting managed encryption metadata for {name}")))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn decode_context_accepts_compatible_headers_and_rejects_conflicts() {
+        let expected = HashMap::from([("tenant".to_string(), "alpha".to_string())]);
+        let metadata = HashMap::from([
+            (
+                RUSTFS_ENCRYPTION_CONTEXT_HEADER.to_string(),
+                serde_json::to_string(&expected).expect("RustFS KMS context should serialize"),
+            ),
+            (
+                MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER.to_string(),
+                BASE64_STANDARD.encode(serde_json::to_vec(&expected).expect("MinIO KMS context should serialize")),
+            ),
+        ]);
+        assert_eq!(
+            decode_managed_kms_context(&metadata).expect("matching KMS contexts should parse"),
+            Some(expected.clone())
+        );
+        assert_eq!(
+            decode_managed_kms_context(&HashMap::from([(
+                RUSTFS_ENCRYPTION_CONTEXT_HEADER.to_string(),
+                serde_json::to_string(&expected).expect("legacy RustFS KMS context should serialize"),
+            )]))
+            .expect("legacy RustFS KMS context should parse"),
+            Some(expected)
+        );
+
+        let conflicting = HashMap::from([
+            (
+                RUSTFS_ENCRYPTION_CONTEXT_HEADER.to_string(),
+                serde_json::to_string(&HashMap::from([("tenant", "alpha")])).expect("RustFS KMS context should serialize"),
+            ),
+            (
+                MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER.to_string(),
+                BASE64_STANDARD.encode(
+                    serde_json::to_vec(&HashMap::from([("tenant", "beta")])).expect("MinIO KMS context should serialize"),
+                ),
+            ),
+        ]);
+        assert!(decode_managed_kms_context(&conflicting).is_err());
+
+        assert!(
+            decode_managed_kms_context(&HashMap::from([(
+                MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER.to_string(),
+                "not-base64".to_string(),
+            )]))
+            .is_err()
+        );
+        assert!(
+            decode_managed_kms_context(&HashMap::from([(RUSTFS_ENCRYPTION_CONTEXT_HEADER.to_string(), "not-json".to_string(),)]))
+                .is_err()
+        );
+    }
+}

crates/utils/src/http/header_compat.rs

@@ -20,12 +20,26 @@
 
 use http::{HeaderMap, HeaderValue};
 use std::borrow::Cow;
+use std::collections::HashMap;
+use std::error::Error;
+use std::fmt::{Display, Formatter};
 
 const RUSTFS_PREFIX: &str = "x-rustfs-";
 const MINIO_PREFIX: &str = "x-minio-";
 const MINIO_ENCRYPTION_PREFIX: &str = "x-minio-encryption-";
 const RUSTFS_ENCRYPTION_PREFIX: &str = "x-rustfs-encryption-";
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ConflictingMetadataValue;
+
+impl Display for ConflictingMetadataValue {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.write_str("metadata contains conflicting values for the same case-insensitive key")
+    }
+}
+
+impl Error for ConflictingMetadataValue {}
+
 // Suffix constants (part after x-rustfs- or x-minio-). Use with get_header/insert_header.
 pub const SUFFIX_FORCE_DELETE: &str = "force-delete";
 pub const SUFFIX_INCLUDE_DELETED: &str = "include-deleted";
@@ -84,6 +98,24 @@ pub fn get_header_map(map: &std::collections::HashMap<String, String>, suffix: &
     map.get(&rk).cloned().or_else(|| map.get(&mk).cloned())
 }
 
+/// Returns the value when every case-insensitive occurrence of `name` agrees.
+pub fn get_consistent_metadata_value<'a>(
+    metadata: &'a HashMap<String, String>,
+    name: &str,
+) -> Result<Option<&'a str>, ConflictingMetadataValue> {
+    let mut value = None;
+    for (candidate, candidate_value) in metadata {
+        if !candidate.eq_ignore_ascii_case(name) {
+            continue;
+        }
+        if value.is_some_and(|existing| existing != candidate_value) {
+            return Err(ConflictingMetadataValue);
+        }
+        value = Some(candidate_value.as_str());
+    }
+    Ok(value)
+}
+
 /// Insert into HashMap with both x-rustfs-{suffix} and x-minio-{suffix}.
 pub fn insert_header_map(map: &mut std::collections::HashMap<String, String>, suffix: &str, value: impl Into<String>) {
     let v = value.into();
@@ -120,4 +152,28 @@ mod tests {
         headers2.insert("X-Rustfs-Force-Delete", HeaderValue::from_static("true"));
         assert_eq!(get_header(&headers2, SUFFIX_FORCE_DELETE).as_deref(), Some("true"));
     }
+
+    #[test]
+    fn test_get_consistent_metadata_value() {
+        let mut metadata = HashMap::new();
+        assert_eq!(
+            get_consistent_metadata_value(&metadata, "x-test").expect("missing value should be valid"),
+            None
+        );
+
+        metadata.insert("X-Test".to_string(), "value".to_string());
+        assert_eq!(
+            get_consistent_metadata_value(&metadata, "x-test").expect("single value should be valid"),
+            Some("value")
+        );
+
+        metadata.insert("x-test".to_string(), "value".to_string());
+        assert_eq!(
+            get_consistent_metadata_value(&metadata, "x-test").expect("matching values should be valid"),
+            Some("value")
+        );
+
+        metadata.insert("x-TEST".to_string(), "other".to_string());
+        assert!(get_consistent_metadata_value(&metadata, "x-test").is_err());
+    }
 }

docs/architecture/compat-cleanup-register.md

@@ -12,6 +12,7 @@ for later deletion.
 
 ## Open Items
 
+- `rustfs-5063` legacy SSE-S3 KMS envelopes: releases before provider routing was separated could wrap an AES256 object's DEK with KMS, so readers identify the persisted KMS envelope and retain KMS decryption for those objects. Remove this compatibility path after the SSE-S3 migration has rewrapped every referenced legacy DEK with the local SSE-S3 key provider.
 - `#4648` walk-dir stream completion capability: old clients can append fallback output to an already-used metacache writer after a terminal body error, so servers emit terminal walk errors only to clients that sign the `walk_dir_stream_completion=error-v1` query capability and its request-body digest. Remove the legacy clean-EOF path after the minimum supported RustFS peer version always advertises this capability.
 - `heal-rpc-auth-v2` internode gRPC authentication: servers temporarily accept legacy prefix signatures so old peers remain available during rolling upgrades. Remove the legacy fallback after the minimum supported RustFS peer version sends v2 authentication on every internode gRPC request.
 - `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.

rustfs/src/storage/storage_api.rs

@@ -502,6 +502,10 @@ pub(crate) mod ecstore_set_disk {
     pub(crate) use rustfs_ecstore::api::set_disk::{DEFAULT_READ_BUFFER_SIZE, get_lock_acquire_timeout, is_valid_storage_class};
 }
 
+pub(crate) mod ecstore_sse {
+    pub(crate) use rustfs_ecstore::api::sse::{ManagedDekProvider, ManagedSseScheme, managed_dek_provider};
+}
+
 pub(crate) mod ecstore_storage {
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::storage::init_local_disks;