diff --git a/crates/e2e_test/src/kms/bucket_default_encryption_test.rs b/crates/e2e_test/src/kms/bucket_default_encryption_test.rs index c0648c1ca..965b33d02 100644 --- a/crates/e2e_test/src/kms/bucket_default_encryption_test.rs +++ b/crates/e2e_test/src/kms/bucket_default_encryption_test.rs @@ -21,9 +21,12 @@ use super::common::LocalKMSTestEnvironment; use crate::common::{TEST_BUCKET, init_logging}; +use aws_sdk_s3::primitives::ByteStream; use aws_sdk_s3::types::{ - ServerSideEncryption, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, + ChecksumAlgorithm, ChecksumMode, CompletedMultipartUpload, CompletedPart, ServerSideEncryption, + ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, }; +use rustfs_rio::{Checksum, ChecksumType}; use serial_test::serial; use tracing::{debug, info, warn}; @@ -273,7 +276,7 @@ async fn test_bucket_default_sse_kms_put_object() -> Result<(), Box Result<(), Box> { +async fn test_bucket_default_sse_kms_multipart_crc32() -> Result<(), Box> { init_logging(); info!("Testing bucket default encryption impact on create_multipart_upload"); @@ -309,15 +312,16 @@ async fn test_bucket_default_encryption_multipart_upload() -> Result<(), Box Result<(), Box = (0..PART_SIZE).map(|i| (i % 251) as u8).collect(); + let part2: Vec = (0..1024 * 1024).map(|i| ((i + 17) % 251) as u8).collect(); + let expected_body: Vec = part1.iter().chain(&part2).copied().collect(); - // Upload part 1 - let upload_part_response = s3_client - .upload_part() - .bucket(TEST_BUCKET) - .key(test_key) - .upload_id(upload_id) - .part_number(1) - .body(test_data.to_vec().into()) - .send() - .await - .expect("Failed to upload part"); + let upload_part = |part_number: i32, body: Vec| { + s3_client + .upload_part() + .bucket(TEST_BUCKET) + .key(test_key) + .upload_id(upload_id) + .part_number(part_number) + .checksum_algorithm(ChecksumAlgorithm::Crc32) + .body(ByteStream::from(body)) + .send() + }; - let etag = upload_part_response.e_tag().unwrap().to_string(); + let expected_part1_crc32 = Checksum::new_from_data(ChecksumType::CRC32, &part1) + .expect("calculate part 1 CRC32") + .encoded; + let upload1 = upload_part(1, part1).await.expect("Failed to upload part 1 with CRC32"); + assert_eq!( + upload1.checksum_crc32(), + Some(expected_part1_crc32.as_str()), + "UploadPart must return the CRC32 calculated over plaintext" + ); + + let expected_part2_crc32 = Checksum::new_from_data(ChecksumType::CRC32, &part2) + .expect("calculate part 2 CRC32") + .encoded; + let upload2 = upload_part(2, part2).await.expect("Failed to upload part 2 with CRC32"); + assert_eq!( + upload2.checksum_crc32(), + Some(expected_part2_crc32.as_str()), + "UploadPart must return the CRC32 calculated over plaintext" + ); // Complete multipart upload - let completed_part = aws_sdk_s3::types::CompletedPart::builder() - .part_number(1) - .e_tag(&etag) + let completed_upload = CompletedMultipartUpload::builder() + .parts( + CompletedPart::builder() + .part_number(1) + .e_tag(upload1.e_tag().expect("No ETag for part 1")) + .checksum_crc32(upload1.checksum_crc32().expect("No CRC32 for part 1")) + .build(), + ) + .parts( + CompletedPart::builder() + .part_number(2) + .e_tag(upload2.e_tag().expect("No ETag for part 2")) + .checksum_crc32(upload2.checksum_crc32().expect("No CRC32 for part 2")) + .build(), + ) .build(); let complete_multipart_response = s3_client @@ -372,11 +409,7 @@ async fn test_bucket_default_encryption_multipart_upload() -> Result<(), Box Result<(), Box Result<(), Box Result<(), Box std::io::Result { + // Transformations create new HashReaders around the plaintext reader. Keep + // the request checksum metadata on the final reader for multipart/single + // PUT persistence, but leave verification to the plaintext reader. + let checksum = reader.content_hash().clone(); + let trailer = reader.get_trailer().cloned(); + let encrypted = self.encryption.is_some(); if let Some(algorithm) = self.compression { reader = HashReader::from_reader( @@ -438,6 +444,12 @@ impl WritePlan { }; } + // `ignore_value` deliberately avoids a second hasher over compressed or + // encrypted bytes. The inner reader still validates the plaintext request + // checksum while this outer reader exposes the request checksum context. + reader.add_non_trailing_checksum(checksum, true)?; + reader.set_trailer(trailer); + Ok(reader) } } @@ -445,10 +457,73 @@ impl WritePlan { #[cfg(test)] mod tests { use super::*; + use http::{HeaderMap, HeaderValue}; + use rustfs_rio::{Checksum, ChecksumType}; use rustfs_utils::CompressionAlgorithm; use std::io::Cursor; use tokio::io::AsyncReadExt; + async fn assert_non_trailing_checksum_survives(plan: WritePlan) { + let plaintext = b"checksum-context-through-write-plan".repeat(256); + let actual_size = plaintext.len() as i64; + let checksum = Checksum::new_from_data(ChecksumType::CRC32, &plaintext).expect("create CRC32 checksum"); + let mut reader = HashReader::from_stream(Cursor::new(plaintext), actual_size, actual_size, None, None, false) + .expect("create hash reader"); + reader + .add_non_trailing_checksum(Some(checksum.clone()), false) + .expect("attach plaintext checksum"); + + let mut transformed = plan.apply(reader, actual_size).expect("apply write plan"); + assert_eq!(transformed.content_crc_type(), Some(ChecksumType::CRC32)); + + let mut transformed_bytes = Vec::new(); + transformed + .read_to_end(&mut transformed_bytes) + .await + .expect("stream transformed data without rehashing ciphertext"); + + assert!(!transformed_bytes.is_empty()); + assert_eq!(transformed.content_crc().get("CRC32"), Some(&checksum.encoded)); + } + + #[tokio::test] + async fn write_plan_preserves_non_trailing_checksum_context_across_transforms() { + assert_non_trailing_checksum_survives(WritePlan::new().with_compression(CompressionAlgorithm::default())).await; + assert_non_trailing_checksum_survives( + WritePlan::new().with_encryption(WriteEncryption::singlepart([0x5Au8; 32], [0xA5u8; 12])), + ) + .await; + assert_non_trailing_checksum_survives( + WritePlan::new() + .with_compression(CompressionAlgorithm::default()) + .with_encryption(WriteEncryption::singlepart([0x5Au8; 32], [0xA5u8; 12])), + ) + .await; + } + + #[tokio::test] + async fn write_plan_preserves_trailing_checksum_type_across_transforms() { + let plaintext = b"trailing-checksum-context".to_vec(); + let actual_size = plaintext.len() as i64; + let mut reader = HashReader::from_stream(Cursor::new(plaintext), actual_size, actual_size, None, None, false) + .expect("create hash reader"); + let mut headers = HeaderMap::new(); + headers.insert("x-amz-trailer", HeaderValue::from_static("x-amz-checksum-crc32")); + reader + .add_checksum_from_s3s(&headers, None, false) + .expect("attach trailing checksum metadata"); + + let transformed = WritePlan::new() + .with_encryption(WriteEncryption::singlepart([0x5Au8; 32], [0xA5u8; 12])) + .apply(reader, actual_size) + .expect("apply encryption plan"); + + assert_eq!( + transformed.content_crc_type(), + Some(ChecksumType(ChecksumType::CRC32.0 | ChecksumType::TRAILING.0)) + ); + } + #[cfg(feature = "rio-v2")] fn s2_chunk_types(stream: &[u8]) -> Vec { let mut chunk_types = Vec::new();