fix(storage): restore legacy SSE-S3 read compatibility (#3584)

* Update .gitignore

* Fix. fixed SSE-S3 compatibility issues in large-scale testing

* fix

* fix(ecstore): reject whitespace bucket names

* Update replication_extension_test.rs

* style(ecstore): format bucket whitespace test

---------

Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: cxymds <cxymds@gmail.com>

crates/e2e_test/src/replication_extension_test.rs

@@ -1617,6 +1617,46 @@ async fn test_delete_bucket_replication_removes_remote_target() -> Result<(), Bo
     Ok(())
 }
 
+#[tokio::test]
+#[serial]
+async fn test_bucket_replication_replicates_put_object_issue_2539() -> Result<(), Box<dyn Error + Send + Sync>> {
+    init_logging();
+
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    source_env.start_rustfs_server(vec![]).await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env.start_rustfs_server_without_cleanup(vec![]).await?;
+
+    let source_bucket = "issue-2539-src";
+    let target_bucket = "issue-2539-dst";
+    let object_key = "put-object.txt";
+    let body = "bucket replication should copy PutObject payload";
+
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+
+    source_client.create_bucket().bucket(source_bucket).send().await?;
+    target_client.create_bucket().bucket(target_bucket).send().await?;
+    enable_bucket_versioning(&source_env, source_bucket).await?;
+    enable_bucket_versioning(&target_env, target_bucket).await?;
+
+    let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
+    put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
+
+    source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(object_key)
+        .body(ByteStream::from(body.as_bytes().to_vec()))
+        .send()
+        .await?;
+
+    wait_for_replicated_object(&target_client, target_bucket, object_key, body).await?;
+
+    Ok(())
+}
+
 #[tokio::test]
 #[serial]
 async fn test_single_bucket_replication_fans_out_to_multiple_targets() -> Result<(), Box<dyn Error + Send + Sync>> {

crates/ecstore/src/bucket/utils.rs

@@ -32,7 +32,7 @@ lazy_static::lazy_static! {
 pub fn check_bucket_name_common(bucket_name: &str, strict: bool) -> Result<()> {
     let bucket_name_trimmed = bucket_name.trim();
 
-    if strict && bucket_name_trimmed != bucket_name {
+    if bucket_name_trimmed != bucket_name {
         return Err(Error::other("Bucket name cannot contain leading or trailing whitespace"));
     }
     if bucket_name_trimmed.is_empty() {
@@ -456,10 +456,23 @@ mod tests {
     }
 
     #[test]
-    fn test_strict_bucket_name_rejects_surrounding_whitespace() {
-        assert!(check_valid_bucket_name_strict(" valid-bucket").is_err());
-        assert!(check_valid_bucket_name_strict("valid-bucket ").is_err());
-        assert!(check_valid_bucket_name_strict("\u{c}valid-bucket\u{c}").is_err());
+    fn test_check_bucket_name_rejects_leading_and_trailing_whitespace() {
+        for bucket in [
+            " valid-bucket",
+            "valid-bucket ",
+            "valid-bucket\n",
+            "valid-bucket\u{b}",
+            "\u{c}valid-bucket\u{c}",
+        ] {
+            assert!(
+                check_valid_bucket_name_strict(bucket).is_err(),
+                "bucket name with leading or trailing whitespace must be rejected: {bucket:?}"
+            );
+            assert!(
+                check_valid_bucket_name(bucket).is_err(),
+                "legacy bucket validation must reject leading or trailing whitespace: {bucket:?}"
+            );
+        }
     }
 
     #[test]

crates/ecstore/src/object_api/readers.rs

@@ -97,6 +97,14 @@ fn build_object_encryption_context(
     object_context
 }
 
+#[cfg(feature = "rio-v2")]
+fn is_legacy_rustfs_managed_metadata(metadata: &HashMap<String, String>) -> bool {
+    metadata_get(metadata, INTERNAL_ENCRYPTION_KEY_HEADER).is_some()
+        && metadata_get(metadata, INTERNAL_ENCRYPTION_IV_HEADER).is_some()
+        && metadata_get(metadata, MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER).is_none()
+        && metadata_get(metadata, MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER).is_none()
+}
+
 fn part_plaintext_size(part: &ObjectPartInfo) -> i64 {
     if part.actual_size > 0 {
         part.actual_size
@@ -1301,9 +1309,16 @@ async fn resolve_managed_material(bucket: &str, object: &str, metadata: &HashMap
     let object_context = build_object_encryption_context(bucket, object, kms_context.as_ref());
 
     let decrypted_key = if let Some(service) = get_global_encryption_service().await {
-        service
-            .decrypt_data_key(&encrypted_dek, &object_context)
-            .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;
+
+        data_key
             .map_err(|e| Error::other(format!("failed to decrypt managed data key: {e}")))?
             .plaintext_key
     } else {
@@ -1601,6 +1616,24 @@ mod tests {
         headers
     }
 
+    #[cfg(feature = "rio-v2")]
+    #[test]
+    fn test_legacy_managed_metadata_excludes_sealed_keys() {
+        let legacy_metadata = HashMap::from([
+            (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), "encrypted-dek".to_string()),
+            (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), "nonce".to_string()),
+        ]);
+        assert!(is_legacy_rustfs_managed_metadata(&legacy_metadata));
+
+        let sealed_metadata = HashMap::from([
+            (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), "encrypted-dek".to_string()),
+            (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), "nonce".to_string()),
+            (MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER.to_string(), "sealed-key".to_string()),
+        ]);
+
+        assert!(!is_legacy_rustfs_managed_metadata(&sealed_metadata));
+    }
+
     #[cfg(feature = "rio-v2")]
     fn seal_ssec_object_key_for_test(
         bucket: &str,

crates/kms/src/service.rs

@@ -232,9 +232,26 @@ impl ObjectEncryptionService {
     /// DataKey with decrypted key
     ///
     pub async fn decrypt_data_key(&self, encrypted_key: &[u8], context: &ObjectEncryptionContext) -> Result<DataKey> {
+        self.decrypt_data_key_with_context(encrypted_key, request_encryption_context(context))
+            .await
+    }
+
+    /// Decrypt a data key written by legacy RustFS versions that reused KMS data
+    /// keys across objects with different encryption contexts.
+    ///
+    /// Callers must restrict this to positively identified legacy object metadata.
+    pub async fn decrypt_legacy_data_key(&self, encrypted_key: &[u8]) -> Result<DataKey> {
+        self.decrypt_data_key_with_context(encrypted_key, HashMap::new()).await
+    }
+
+    async fn decrypt_data_key_with_context(
+        &self,
+        encrypted_key: &[u8],
+        encryption_context: HashMap<String, String>,
+    ) -> Result<DataKey> {
         let decrypt_request = DecryptRequest {
             ciphertext: encrypted_key.to_vec(),
-            encryption_context: request_encryption_context(context),
+            encryption_context,
             grant_tokens: Vec::new(),
         };
 
@@ -928,5 +945,11 @@ mod tests {
             .await
             .expect("decrypt should accept matching KMS context");
         assert_ne!(decrypted.plaintext_key, [0u8; 32]);
+
+        let legacy_decrypted = service
+            .decrypt_legacy_data_key(&encrypted_key)
+            .await
+            .expect("legacy decrypt should use the backend compatibility path");
+        assert_eq!(legacy_decrypted.plaintext_key, decrypted.plaintext_key);
     }
 }