fix(s3): align encrypted checksums and multipart completion (#7025)

* fix(s3): align encrypted checksum handling

* test(s3): align multipart SSE-C completion

* fix(ecstore): scope startup helper to tests

crates/e2e_test/src/checksum_upload_test.rs

@@ -22,7 +22,7 @@ mod tests {
     use aws_sdk_s3::config::{Credentials, Region, RequestChecksumCalculation};
     use aws_sdk_s3::error::ProvideErrorMetadata;
     use aws_sdk_s3::primitives::ByteStream;
-    use aws_sdk_s3::types::{ChecksumAlgorithm, ChecksumMode, CompletedMultipartUpload, CompletedPart};
+    use aws_sdk_s3::types::{ChecksumAlgorithm, ChecksumMode, CompletedMultipartUpload, CompletedPart, ServerSideEncryption};
     use aws_smithy_http_client::Builder as SmithyHttpClientBuilder;
     use md5::{Digest as Md5Digest, Md5};
     use rustfs_rio::{Checksum, ChecksumType as RioChecksumType};
@@ -260,6 +260,117 @@ mod tests {
         info!("PASSED: HeadObject returns stored SHA256 digest");
     }
 
+    #[tokio::test]
+    async fn test_head_object_returns_sse_s3_checksum() {
+        init_logging();
+
+        let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
+        env.start_rustfs_server_with_env(
+            vec![],
+            &[
+                ("RUSTFS_SSE_S3_MASTER_KEY", "MTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0NTY3ODkwMTI="),
+                ("RUSTFS_CONSOLE_ENABLE", "false"),
+            ],
+        )
+        .await
+        .expect("Failed to start RustFS");
+
+        let client = create_s3_client(&env);
+        let bucket = "test-sse-s3-checksum-head";
+        create_bucket(&client, bucket).await.expect("Failed to create bucket");
+
+        let put = client
+            .put_object()
+            .bucket(bucket)
+            .key("encrypted.txt")
+            .body(ByteStream::from_static(b"encrypted checksum"))
+            .server_side_encryption(ServerSideEncryption::Aes256)
+            .checksum_algorithm(ChecksumAlgorithm::Crc32)
+            .send()
+            .await
+            .expect("SSE-S3 PutObject with CRC32 failed");
+        let expected = put.checksum_crc32().expect("PutObject must return CRC32");
+
+        let head = client
+            .head_object()
+            .bucket(bucket)
+            .key("encrypted.txt")
+            .checksum_mode(ChecksumMode::Enabled)
+            .send()
+            .await
+            .expect("SSE-S3 HeadObject failed");
+
+        assert_eq!(head.checksum_crc32(), Some(expected));
+
+        client
+            .copy_object()
+            .bucket(bucket)
+            .key("encrypted-copy.txt")
+            .copy_source(format!("{bucket}/encrypted.txt"))
+            .server_side_encryption(ServerSideEncryption::Aes256)
+            .send()
+            .await
+            .expect("SSE-S3 CopyObject failed");
+        let copy_head = client
+            .head_object()
+            .bucket(bucket)
+            .key("encrypted-copy.txt")
+            .checksum_mode(ChecksumMode::Enabled)
+            .send()
+            .await
+            .expect("SSE-S3 copied HeadObject failed");
+
+        assert_eq!(copy_head.checksum_crc32(), Some(expected));
+
+        let multipart_key = "encrypted-multipart.txt";
+        let create = client
+            .create_multipart_upload()
+            .bucket(bucket)
+            .key(multipart_key)
+            .server_side_encryption(ServerSideEncryption::Aes256)
+            .checksum_algorithm(ChecksumAlgorithm::Crc32)
+            .send()
+            .await
+            .expect("SSE-S3 CreateMultipartUpload with CRC32 failed");
+        let upload_id = create.upload_id().expect("CreateMultipartUpload must return an upload ID");
+        let part = client
+            .upload_part()
+            .bucket(bucket)
+            .key(multipart_key)
+            .upload_id(upload_id)
+            .part_number(1)
+            .body(ByteStream::from_static(b"encrypted multipart checksum"))
+            .checksum_algorithm(ChecksumAlgorithm::Crc32)
+            .send()
+            .await
+            .expect("SSE-S3 UploadPart with CRC32 failed");
+        let completed_part = CompletedPart::builder()
+            .part_number(1)
+            .e_tag(part.e_tag().expect("UploadPart must return an ETag"))
+            .checksum_crc32(part.checksum_crc32().expect("UploadPart must return CRC32"))
+            .build();
+        let complete = client
+            .complete_multipart_upload()
+            .bucket(bucket)
+            .key(multipart_key)
+            .upload_id(upload_id)
+            .multipart_upload(CompletedMultipartUpload::builder().parts(completed_part).build())
+            .send()
+            .await
+            .expect("SSE-S3 CompleteMultipartUpload with CRC32 failed");
+        let expected_multipart = complete.checksum_crc32().expect("CompleteMultipartUpload must return CRC32");
+        let multipart_head = client
+            .head_object()
+            .bucket(bucket)
+            .key(multipart_key)
+            .checksum_mode(ChecksumMode::Enabled)
+            .send()
+            .await
+            .expect("SSE-S3 multipart HeadObject failed");
+
+        assert_eq!(multipart_head.checksum_crc32(), Some(expected_multipart));
+    }
+
     /// Multipart upload with checksum: CreateMultipartUpload, UploadPart(s) with checksum_sha256, CompleteMultipartUpload; then GetObject verifies content.
     /// Uses part size >= 5MB (server minimum) for two parts.
     #[tokio::test]

crates/e2e_test/src/kms/multipart_encryption_test.rs

@@ -560,18 +560,12 @@ async fn test_multipart_encryption_type(
         .set_parts(Some(completed_parts))
         .build();
 
-    let mut complete_request = s3_client
+    let complete_request = s3_client
         .complete_multipart_upload()
         .bucket(bucket)
         .key(object_key)
         .upload_id(upload_id)
         .multipart_upload(completed_multipart_upload);
-    if matches!(encryption_type, EncryptionType::SSEC) {
-        complete_request = complete_request
-            .sse_customer_algorithm("AES256")
-            .sse_customer_key(sse_c_key.as_ref().unwrap())
-            .sse_customer_key_md5(sse_c_md5.as_ref().unwrap());
-    }
     let _complete_output = complete_request.send().await?;
 
     // Download and verify

crates/ecstore/src/object_api/mod.rs

@@ -37,7 +37,9 @@ use rustfs_filemeta::{FileInfo, MetaCacheEntriesSorted, ObjectPartInfo, RestoreS
 use rustfs_rio::Checksum;
 use rustfs_utils::CompressionAlgorithm;
 use rustfs_utils::http::headers::AMZ_OBJECT_TAGGING;
-use rustfs_utils::http::{AMZ_BUCKET_REPLICATION_STATUS, AMZ_RESTORE, AMZ_STORAGE_CLASS};
+use rustfs_utils::http::{
+    AMZ_BUCKET_REPLICATION_STATUS, AMZ_RESTORE, AMZ_STORAGE_CLASS, SUFFIX_PLAINTEXT_CHECKSUM, get_consistent_str,
+};
 use rustfs_utils::path::decode_dir_object;
 use std::collections::HashMap;
 use std::fmt::Debug;

crates/ecstore/src/object_api/types.rs

@@ -1771,9 +1771,10 @@ impl ObjectInfo {
         }
 
         if let Some(data) = &self.checksum {
-            if self.is_encrypted() {
+            if self.is_encrypted() && get_consistent_str(&self.user_defined, SUFFIX_PLAINTEXT_CHECKSUM) != Some("true") {
                 // Object-level encrypted checksum bytes require SSE decrypt material,
-                // so do not expose them as plaintext checksum headers here. The
+                // unless RustFS marked the stored bytes as plaintext. Do not expose
+                // unmarked bytes as checksum headers here. The
                 // `false` multipart flag feeds the response-path COMPOSITE
                 // fallback; callers that need accurate multipart routing must
                 // consult `is_multipart()` instead of this value.
@@ -2479,6 +2480,31 @@ mod tests {
         assert!(checksums.is_empty());
     }
 
+    #[test]
+    fn decrypt_checksums_reads_marked_rustfs_encrypted_object_checksum() {
+        let checksum = rustfs_rio::Checksum::new_from_data(rustfs_rio::ChecksumType::CRC32, b"encrypted-object")
+            .expect("test checksum should be valid");
+        let checksum_key = checksum.checksum_type.to_string();
+        let expected_checksum = checksum.encoded.clone();
+        let mut user_defined =
+            HashMap::from([(rustfs_utils::http::headers::AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "AES256".to_string())]);
+        rustfs_utils::http::insert_str(&mut user_defined, SUFFIX_PLAINTEXT_CHECKSUM, "true".to_string());
+        assert_eq!(user_defined.get("x-rustfs-internal-plaintext-checksum").map(String::as_str), Some("true"));
+        assert_eq!(user_defined.get("x-minio-internal-plaintext-checksum").map(String::as_str), Some("true"));
+        let info = ObjectInfo {
+            checksum: Some(checksum.to_bytes(&[])),
+            user_defined: Arc::new(user_defined),
+            ..Default::default()
+        };
+
+        let (checksums, is_multipart) = info
+            .decrypt_checksums(0, &HeaderMap::new())
+            .expect("marked RustFS checksum should decode");
+
+        assert!(!is_multipart);
+        assert_eq!(checksums.get(&checksum_key), Some(&expected_checksum));
+    }
+
     #[test]
     fn decrypt_checksums_keeps_encrypted_multipart_flag_false_for_response_paths() {
         let checksum = rustfs_rio::Checksum::new_from_data(rustfs_rio::ChecksumType::CRC32, b"encrypted-object")

crates/utils/src/http/metadata_compat.rs

@@ -50,6 +50,11 @@ pub const SUFFIX_ACTUAL_OBJECT_SIZE: &str = "actual-object-size";
 /// Used by replication; key stored with capital A
 pub const SUFFIX_ACTUAL_OBJECT_SIZE_CAP: &str = "Actual-Object-Size";
 pub const SUFFIX_CRC: &str = "crc";
+/// Marks checksum bytes produced by RustFS as plaintext checksum metadata.
+///
+/// MinIO encrypts the same on-disk field for SSE objects, so readers must not
+/// decode encrypted-object checksums unless this marker is present.
+pub const SUFFIX_PLAINTEXT_CHECKSUM: &str = "plaintext-checksum";
 /// JSON-encoded per-part S3 checksum maps retained across raw data movement.
 pub const SUFFIX_PART_CHECKSUMS: &str = "part-checksums";
 pub const SUFFIX_TRANSITION_STATUS: &str = "transition-status";

rustfs/src/app/multipart_usecase.rs

@@ -101,7 +101,7 @@ use rustfs_utils::CompressionAlgorithm;
 #[cfg(test)]
 use rustfs_utils::http::insert_header;
 use rustfs_utils::http::{
-    SUFFIX_MAX_TOTAL_OBJECT_SIZE, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, SUFFIX_REPLICATION_STATUS,
+    SUFFIX_MAX_TOTAL_OBJECT_SIZE, SUFFIX_PLAINTEXT_CHECKSUM, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, SUFFIX_REPLICATION_STATUS,
     SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST, contains_key_str, get_consistent_str, get_header,
     get_source_scheme,
     headers::{AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING, AMZ_STORAGE_CLASS},
@@ -653,7 +653,7 @@ impl DefaultMultipartUsecase {
                 content_size: 0,
                 principal: None,
             }
-            .validate_multipart_ssec(&multipart_info.user_defined)?;
+            .validate_complete_multipart_ssec(&multipart_info.user_defined)?;
         }
         let cache_adapter = self.object_data_cache();
         let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &key).await;
@@ -1023,6 +1023,9 @@ impl DefaultMultipartUsecase {
                 ..Default::default()
             });
         }
+        if effective_sse.is_some() && opts.want_checksum.is_some() {
+            insert_str(&mut opts.user_defined, SUFFIX_PLAINTEXT_CHECKSUM, "true".to_string());
+        }
 
         let MultipartUploadResult {
             upload_id,

rustfs/src/app/object/copy.rs

@@ -575,6 +575,7 @@ impl DefaultObjectUsecase {
         }
 
         strip_managed_encryption_metadata(&mut user_defined);
+        remove_str(&mut user_defined, SUFFIX_PLAINTEXT_CHECKSUM);
 
         let destination_storage_class = storage_class
             .as_ref()
@@ -665,6 +666,7 @@ impl DefaultObjectUsecase {
         // none is requested, carry the source object's stored checksum over unchanged — the copy
         // does not alter the plaintext, so re-hashing would be wasted work and would flatten a
         // multipart composite value.
+        let destination_has_checksum = requested_checksum_type.is_some() || src_checksum.is_some();
         match requested_checksum_type {
             Some(checksum_type) => {
                 reader.add_calculated_checksum(checksum_type).map_err(ApiError::from)?;
@@ -696,6 +698,9 @@ impl DefaultObjectUsecase {
             write_plan = write_plan.with_encryption(material.write_encryption(None));
 
             user_defined.extend(encryption_material_to_metadata(&material)?);
+            if destination_has_checksum {
+                insert_str(&mut user_defined, SUFFIX_PLAINTEXT_CHECKSUM, "true".to_string());
+            }
         }
 
         reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;

rustfs/src/app/object/mod.rs

@@ -145,9 +145,9 @@ use rustfs_utils::http::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_K
 use rustfs_utils::http::insert_header;
 use rustfs_utils::http::{
     AMZ_BUCKET_REPLICATION_STATUS, AMZ_CHECKSUM_MODE, AMZ_CHECKSUM_TYPE, AMZ_WEBSITE_REDIRECT_LOCATION, CONTENT_TYPE,
-    SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICA_STATUS, SUFFIX_REPLICA_TIMESTAMP,
-    SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_RESTORE_OPERATION_ID, SUFFIX_SOURCE_REPLICATION_CHECK,
-    SUFFIX_SOURCE_REPLICATION_REQUEST, get_header,
+    SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_PLAINTEXT_CHECKSUM, SUFFIX_REPLICA_STATUS,
+    SUFFIX_REPLICA_TIMESTAMP, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_RESTORE_OPERATION_ID,
+    SUFFIX_SOURCE_REPLICATION_CHECK, SUFFIX_SOURCE_REPLICATION_REQUEST, get_header,
     headers::{
         AMZ_CONTENT_SHA256, AMZ_DECODED_CONTENT_LENGTH, AMZ_MINIO_SNOWBALL_IGNORE_DIRS, AMZ_MINIO_SNOWBALL_IGNORE_ERRORS,
         AMZ_MINIO_SNOWBALL_PREFIX, AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE,

rustfs/src/app/object/put.rs

@@ -1447,6 +1447,10 @@ impl DefaultObjectUsecase {
             let encryption_metadata = encryption_material_to_metadata(&material)?;
             metadata.extend(encryption_metadata.clone());
             opts.user_defined.extend(encryption_metadata);
+            if opts.want_checksum.is_some() {
+                insert_str(&mut metadata, SUFFIX_PLAINTEXT_CHECKSUM, "true".to_string());
+                insert_str(&mut opts.user_defined, SUFFIX_PLAINTEXT_CHECKSUM, "true".to_string());
+            }
         }
 
         reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;

rustfs/src/storage/sse.rs

@@ -460,6 +460,28 @@ pub struct EncryptionRequest<'a> {
 }
 
 impl EncryptionRequest<'_> {
+    pub fn validate_complete_multipart_ssec(&self, user_defined: &HashMap<String, String>) -> Result<(), ApiError> {
+        let request_uses_ssec =
+            self.sse_customer_algorithm.is_some() || self.sse_customer_key.is_some() || self.sse_customer_key_md5.is_some();
+        if !request_uses_ssec {
+            let stored_algorithm = user_defined.get("x-amz-server-side-encryption-customer-algorithm");
+            let stored_key_md5 = user_defined.get("x-amz-server-side-encryption-customer-key-md5");
+            return match (stored_algorithm, stored_key_md5) {
+                (None, None) => Ok(()),
+                (Some(algorithm), Some(key_md5))
+                    if algorithm == DEFAULT_SSE_ALGORITHM
+                        && BASE64_STANDARD
+                            .decode_to_vec(key_md5)
+                            .is_ok_and(|decoded| decoded.len() == 16) =>
+                {
+                    Ok(())
+                }
+                _ => Err(ssec_invalid_request("The multipart upload contains invalid SSE-C metadata.")),
+            };
+        }
+        self.validate_multipart_ssec(user_defined)
+    }
+
     pub fn validate_multipart_ssec(&self, user_defined: &HashMap<String, String>) -> Result<(), ApiError> {
         let stored_algorithm = user_defined.get("x-amz-server-side-encryption-customer-algorithm");
         let stored_key_md5 = user_defined.get("x-amz-server-side-encryption-customer-key-md5");
@@ -6374,6 +6396,61 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_validate_complete_multipart_ssec_allows_omitted_parameters() {
+        let no_ssec = EncryptionRequest {
+            sse_customer_algorithm: None,
+            sse_customer_key: None,
+            sse_customer_key_md5: None,
+            ..multipart_ssec_request(42)
+        };
+
+        assert!(no_ssec.validate_complete_multipart_ssec(&multipart_ssec_metadata(42)).is_ok());
+        assert!(no_ssec.validate_complete_multipart_ssec(&HashMap::new()).is_ok());
+
+        for invalid_metadata in [
+            HashMap::from([("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string())]),
+            HashMap::from([
+                ("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES128".to_string()),
+                ("x-amz-server-side-encryption-customer-key-md5".to_string(), md5_base64([42u8; 32])),
+            ]),
+            HashMap::from([
+                ("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string()),
+                ("x-amz-server-side-encryption-customer-key-md5".to_string(), "invalid".to_string()),
+            ]),
+        ] {
+            assert_eq!(
+                no_ssec
+                    .validate_complete_multipart_ssec(&invalid_metadata)
+                    .expect_err("corrupt SSE-C session metadata must fail closed")
+                    .code,
+                S3ErrorCode::InvalidRequest
+            );
+        }
+    }
+
+    #[test]
+    fn test_validate_complete_multipart_ssec_still_validates_present_parameters() {
+        let partial = EncryptionRequest {
+            sse_customer_key: None,
+            ..multipart_ssec_request(42)
+        };
+        assert_eq!(
+            partial
+                .validate_complete_multipart_ssec(&multipart_ssec_metadata(42))
+                .expect_err("partial SSE-C parameters must fail")
+                .code,
+            S3ErrorCode::InvalidRequest
+        );
+        assert_eq!(
+            multipart_ssec_request(43)
+                .validate_complete_multipart_ssec(&multipart_ssec_metadata(42))
+                .expect_err("wrong SSE-C parameters must fail")
+                .code,
+            S3ErrorCode::InvalidRequest
+        );
+    }
+
     #[test]
     fn test_validate_multipart_ssec_rejects_wrong_key_without_leaking_it() {
         let request = multipart_ssec_request(43);

scripts/s3-tests/patches/0003-complete-multipart-ssec-headers.patch

@@ -0,0 +1,28 @@
+diff --git a/s3tests/functional/test_s3.py b/s3tests/functional/test_s3.py
+--- a/s3tests/functional/test_s3.py
++++ b/s3tests/functional/test_s3.py
+@@ -20631,13 +20631,6 @@ def _test_copy_part_enc(file_size, source_mode_key, dest_mode_key, source_sc=Non
+     })
+ 
+     if dest_mode_key == 'sse-c':
+-        # make sure api is verifying the SSE-C headers
+-        e = assert_raises(ClientError, client.complete_multipart_upload,
+-                          Bucket=dest_bucket_name, Key='testobj2',
+-                          UploadId=upload_id, MultipartUpload={'Parts': parts})
+-        status, _ = _get_status_and_error_code(e.response)
+-        assert status == 400
+-
+         # and the key would be the same as the one used in upload part
+         # use the source key to complete the upload
+         # this is not allowed, so we expect an error
+@@ -20649,6 +20642,10 @@ def _test_copy_part_enc(file_size, source_mode_key, dest_mode_key, source_sc=Non
+         status, _ = _get_status_and_error_code(e.response)
+         assert status == 400
+ 
++        # CompleteMultipartUpload does not require SSE-C headers. The upload
++        # metadata already identifies the key used for the uploaded parts.
++        complete_args = {}
++
+     # complete the multipart upload
+     response = client.complete_multipart_upload(
+         Bucket=dest_bucket_name,