diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 5873f822c..7ca4688e5 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -1453,10 +1453,6 @@ where } } -async fn wait_for_replication_failure_event(response: reqwest::Response, expected_key: &str) -> TestResult { - wait_for_replication_failure_event_stream(response.bytes_stream(), expected_key, Duration::from_secs(30)).await -} - fn target_history_contains_key(output: &ListObjectVersionsOutput, key: &str) -> bool { output.versions().iter().any(|version| version.key() == Some(key)) || output.delete_markers().iter().any(|marker| marker.key() == Some(key)) @@ -1513,29 +1509,6 @@ async fn assert_failed_replication_stays_absent_for( } } -async fn subscribe_to_replication_failure( - env: &RustFSTestEnvironment, - bucket: &str, - key: &str, -) -> Result> { - let url = format!( - "{}/{bucket}?events={}&prefix={}&ping=1", - env.url, - urlencoding::encode(REPLICATION_FAILED_EVENT), - urlencoding::encode(key) - ); - let response = timeout( - Duration::from_secs(30), - signed_request(http::Method::GET, &url, &env.access_key, &env.secret_key, None, None), - ) - .await - .map_err(|_| "replication failure event subscription did not respond within 30 seconds")??; - if response.status() != StatusCode::OK { - return Err(format!("failed to subscribe to replication failure events: {}", response.status()).into()); - } - Ok(response) -} - async fn build_sse_replication_pair( label: &str, source_kms: bool, @@ -4447,9 +4420,11 @@ async fn test_repl17_failure_observation_helpers() -> TestResult { Ok(()) } -/// backlog#1147 repl-17: SSE-C currently fails replication explicitly. Pin the -/// observed contract: the source remains decryptable with its customer key, -/// reports FAILED, emits the standard failure event, and leaves no target data. +/// backlog#1147 repl-17 / backlog#1783: SSE-C objects replicate as ciphertext +/// passthrough — the source cannot decrypt them (no customer key server-side), +/// so the stored ciphertext and its encryption metadata travel verbatim and +/// the replica is decryptable only with the original customer key. The +/// backlog#1291 property still holds: never a silent plaintext replica. #[tokio::test] #[serial] async fn test_bucket_replication_sse_c_contract() -> TestResult { @@ -4462,7 +4437,6 @@ async fn test_bucket_replication_sse_c_contract() -> TestResult { let body = b"repl-17 SSE-C payload"; let customer_key = BASE64_STANDARD.encode(REPL17_SSEC_KEY); let customer_key_md5 = sse_customer_key_md5_base64(REPL17_SSEC_KEY); - let failure_events = subscribe_to_replication_failure(&source_env, &source_bucket, key).await?; source_client .put_object() @@ -4484,41 +4458,157 @@ async fn test_bucket_replication_sse_c_contract() -> TestResult { .sse_customer_key_md5(&customer_key_md5) .send() .await?; + let source_etag = source.e_tag().map(str::to_string); assert_eq!(source.body.collect().await?.into_bytes().as_ref(), body); - wait_for_replication_failure_event(failure_events, key).await?; - wait_for_source_replication_status(&source_client, &source_bucket, key, "FAILED", true).await?; - assert_failed_replication_stays_absent_for( - &source_client, - &source_bucket, - &target_client, - &target_bucket, - key, - true, - Duration::from_secs(5), - ) - .await?; + wait_for_source_replication_status(&source_client, &source_bucket, key, "COMPLETED", true).await?; - target_client - .put_object() + // The replica is readable only with the original customer key. + let replica = target_client + .get_object() .bucket(&target_bucket) .key(key) - .body(ByteStream::from_static(b"observer negative-path fixture")) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) .send() .await?; - target_client.delete_object().bucket(&target_bucket).key(key).send().await?; - let history_error = assert_failed_replication_stays_absent_for( - &source_client, - &source_bucket, - &target_client, - &target_bucket, - key, - true, - Duration::ZERO, - ) - .await - .expect_err("target history must violate the failed replication contract"); - assert!(history_error.to_string().contains("created target history")); + assert_eq!(replica.sse_customer_algorithm(), Some("AES256")); + let replica_etag = replica.e_tag().map(str::to_string); + assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body); + assert_eq!(replica_etag, source_etag, "replica ETag must match the source ETag"); + + // Without the customer key the replica must not be readable — the direct + // detection point for a silent-plaintext replica (backlog#1291). + let plain_read = target_client.get_object().bucket(&target_bucket).key(key).send().await; + assert!(plain_read.is_err(), "SSE-C replica must not be readable without the customer key"); + + // A wrong customer key must fail too. + let wrong_key = BASE64_STANDARD.encode("99999999999999999999999999999999"); + let wrong_key_md5 = sse_customer_key_md5_base64("99999999999999999999999999999999"); + let wrong_read = target_client + .get_object() + .bucket(&target_bucket) + .key(key) + .sse_customer_algorithm("AES256") + .sse_customer_key(&wrong_key) + .sse_customer_key_md5(&wrong_key_md5) + .send() + .await; + assert!(wrong_read.is_err(), "SSE-C replica must reject a wrong customer key"); + + Ok(()) +} + +/// backlog#1783: SSE-C multipart objects pass through as ciphertext part by +/// part — part boundaries and the encrypted-multipart marker survive so the +/// replica decrypts each part with its part-derived nonce. +#[tokio::test] +#[serial] +async fn test_bucket_replication_sse_c_multipart_passthrough() -> TestResult { + init_logging(); + + const PART_SIZE: usize = 5 * 1024 * 1024; + const PART_COUNT: usize = 3; + + let (source_env, target_env, source_bucket, target_bucket) = build_sse_replication_pair("ssec-mp", false, false).await?; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + let key = "ssec-mp-contract.bin"; + let customer_key = BASE64_STANDARD.encode(REPL17_SSEC_KEY); + let customer_key_md5 = sse_customer_key_md5_base64(REPL17_SSEC_KEY); + + let created = source_client + .create_multipart_upload() + .bucket(&source_bucket) + .key(key) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + let upload_id = created.upload_id().ok_or("missing multipart upload id")?.to_string(); + + let mut completed_parts = Vec::with_capacity(PART_COUNT); + let mut payload = Vec::with_capacity(PART_SIZE * PART_COUNT); + for part_number in 1..=PART_COUNT { + let part = vec![u8::try_from(part_number)?; PART_SIZE]; + payload.extend_from_slice(&part); + let uploaded = source_client + .upload_part() + .bucket(&source_bucket) + .key(key) + .upload_id(&upload_id) + .part_number(i32::try_from(part_number)?) + .body(ByteStream::from(part)) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + completed_parts.push( + CompletedPart::builder() + .part_number(i32::try_from(part_number)?) + .set_e_tag(uploaded.e_tag().map(str::to_string)) + .build(), + ); + } + source_client + .complete_multipart_upload() + .bucket(&source_bucket) + .key(key) + .upload_id(&upload_id) + .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build()) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + + wait_for_source_replication_status(&source_client, &source_bucket, key, "COMPLETED", true).await?; + + let source_head = source_client + .head_object() + .bucket(&source_bucket) + .key(key) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + let replica = target_client + .get_object() + .bucket(&target_bucket) + .key(key) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + // The replica must carry the SSE-C marker and the source's multipart ETag + // (its -N suffix also pins that the part structure survived). + assert_eq!(replica.sse_customer_algorithm(), Some("AES256")); + assert_eq!(replica.e_tag(), source_head.e_tag(), "replica must keep the source multipart ETag"); + let replica_version_id = replica.version_id().map(str::to_string); + assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), payload.as_slice()); + + let plain_read = target_client.get_object().bucket(&target_bucket).key(key).send().await; + assert!( + plain_read.is_err(), + "SSE-C multipart replica must not be readable without the customer key" + ); + + // Stability across scanner cycles: convergence must hold for passthrough. + sleep(Duration::from_secs(5)).await; + let versions = target_client + .list_object_versions() + .bucket(&target_bucket) + .prefix(key) + .send() + .await?; + let replica_versions: Vec<_> = versions.versions().iter().filter(|v| v.key() == Some(key)).collect(); + assert_eq!(replica_versions.len(), 1, "SSE-C replica must not accumulate versions"); + assert_eq!(replica_versions[0].version_id().map(str::to_string), replica_version_id); Ok(()) } diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index b663d5db3..1cf1ce833 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -1832,12 +1832,27 @@ impl TargetClient { object: &str, version_id: Option, ) -> Result> { + // Announce the replication check so a RustFS target returns SSE-C + // object metadata (etag/size) without the customer key the replication + // worker cannot hold; otherwise SSE-C replicas never converge on HEAD. + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_CHECK, "true"); match self .client .head_object() .bucket(bucket) .key(object) .set_version_id(version_id) + .customize() + .map_request(move |mut req| { + for (k, v) in headers.clone().into_iter() { + if let Some(key_str) = k.map(|k| k.as_str().to_string()) { + let value_str = v.to_str().unwrap_or("").to_string(); + req.headers_mut().insert(key_str, value_str); + } + } + Result::<_, std::convert::Infallible>::Ok(req) + }) .send() .await { diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index dbdcdd476..9350f9c0d 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -2390,6 +2390,9 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { version_suspended, versioned, replication_request: true, + // SSE-C passthrough reads the stored ciphertext verbatim; the + // decrypting reader cannot serve it (no customer key server-side). + raw_data_movement_read: self.ssec, ..Default::default() }; @@ -2451,6 +2454,9 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } }; + // SSE-C passthrough sends the stored ciphertext; the wire length is + // the stored size while rinfo keeps the logical size for metering. + let transfer_size = if self.ssec { object_info.size } else { size }; if tgt_client.bucket.is_empty() { debug!( @@ -2597,7 +2603,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { gr.stream = wrap_with_bandwidth_monitor(gr.stream, &put_opts, &bucket, &rinfo.arn); let byte_stream = async_read_to_bytestream(gr.stream); let result = tgt_client - .put_object(&tgt_client.bucket, &object, size, byte_stream, &put_opts) + .put_object(&tgt_client.bucket, &object, transfer_size, byte_stream, &put_opts) .await .map_err(|e| std::io::Error::other(e.to_string())); record_proxy_request(&bucket, "PutObject", result.is_err()).await; @@ -2682,6 +2688,9 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { version_suspended, versioned, replication_request: true, + // SSE-C passthrough reads the stored ciphertext verbatim; the + // decrypting reader cannot serve it (no customer key server-side). + raw_data_movement_read: self.ssec, ..Default::default() }; @@ -2742,6 +2751,9 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } }; + // SSE-C passthrough sends the stored ciphertext; the wire length is + // the stored size while rinfo keeps the logical size for metering. + let transfer_size = if self.ssec { object_info.size } else { size }; if tgt_client.bucket.is_empty() { debug!( @@ -2998,7 +3010,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { gr.stream = wrap_with_bandwidth_monitor(gr.stream, &put_opts, &bucket, &rinfo.arn); let byte_stream = async_read_to_bytestream(gr.stream); let result = tgt_client - .put_object(&tgt_client.bucket, &object, size, byte_stream, &put_opts) + .put_object(&tgt_client.bucket, &object, transfer_size, byte_stream, &put_opts) .await .map_err(|e| std::io::Error::other(e.to_string())); record_proxy_request(&bucket, "PutObject", result.is_err()).await; @@ -3137,10 +3149,17 @@ async fn replicate_object_with_multipart(ctx: MultipartR let mut header_size = replication_put_object_header_size(&put_opts); let mut offset: i64 = 0; for part_info in object_info.parts.iter() { + // Ciphertext passthrough (raw read) ranges over the stored part + // bytes; decrypted reads range over the logical plaintext parts. + let part_size = if obj_opts.raw_data_movement_read { + part_info.size as i64 + } else { + part_info.actual_size + }; let part_plan = replication_multipart_part_plan(ReplicationMultipartPartInput { offset, part_number: part_info.number, - part_size: part_info.actual_size, + part_size, }) .map_err(|err| std::io::Error::other(err.to_string()))?; let range_spec = HTTPRangeSpec { diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 108888e5d..138b08b91 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -253,6 +253,11 @@ pub struct ObjectOptions { /// fence avoids recursively acquiring the read lock behind a queued writer. pub bucket_lifecycle_lock_fence: Option, pub replication_request: bool, + /// Authorized SSE-C replication passthrough: the body is already + /// ciphertext, so the write path must not encrypt or compress it and + /// stores the restored encryption metadata verbatim. Only the + /// replication-authorized options builders may set this. + pub preserve_ciphertext: bool, pub delete_marker: bool, pub synthetic_version_id: bool, diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index a52e76677..74d7a45bb 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -1897,6 +1897,13 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } } + // The SSE-C passthrough session marker is upload-scoped; drop it from + // the completed object's metadata. + rustfs_utils::http::metadata_compat::remove_str( + &mut fi.metadata, + rustfs_utils::http::SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, + ); + if checksum_type.is_set() { checksum_type .merge(rustfs_rio::ChecksumType::MULTIPART) diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index de4bed37d..a7858041d 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -1233,6 +1233,16 @@ impl SetDisks { } } + // SSE-C replication carries the source object's sealed checksum + // out of band; store it verbatim like the multipart path does. + if let Some(cssum) = + rustfs_utils::http::get_header_map(&user_defined, rustfs_utils::http::SUFFIX_REPLICATION_SSEC_CRC) + && !cssum.is_empty() + { + fi.checksum = base64_simd::STANDARD.decode_to_vec(&cssum).ok().map(bytes::Bytes::from); + rustfs_utils::http::remove_header_map(&mut user_defined, rustfs_utils::http::SUFFIX_REPLICATION_SSEC_CRC); + } + if fi.checksum.is_none() && let Some(content_hash) = data.as_hash_reader().content_hash() { diff --git a/crates/utils/src/http/metadata_compat.rs b/crates/utils/src/http/metadata_compat.rs index 464a5ed89..ffec691ba 100644 --- a/crates/utils/src/http/metadata_compat.rs +++ b/crates/utils/src/http/metadata_compat.rs @@ -28,6 +28,9 @@ pub const SUFFIX_DATA_MOV: &str = "data-mov"; /// Transient flag for healing pub const SUFFIX_HEALING: &str = "healing"; pub const SUFFIX_COMPRESSION: &str = "compression"; +/// Session marker for SSE-C replication passthrough multipart uploads: parts +/// arrive as ciphertext and must be stored without re-encryption. +pub const SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT: &str = "replication-preserve-ciphertext"; pub const SUFFIX_COMPRESSION_SIZE: &str = "compression-size"; pub const SUFFIX_ACTUAL_SIZE: &str = "actual-size"; pub const SUFFIX_ACTUAL_OBJECT_SIZE: &str = "actual-object-size"; diff --git a/crates/utils/src/http/object_encryption_keys.rs b/crates/utils/src/http/object_encryption_keys.rs index c08459c7a..394572921 100644 --- a/crates/utils/src/http/object_encryption_keys.rs +++ b/crates/utils/src/http/object_encryption_keys.rs @@ -22,7 +22,10 @@ //! ciphertext passthrough; every other encryption key must be stripped from //! outbound replication metadata via [`is_replication_stripped_encryption_key`]. -use super::headers::{AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5}; +// The lowercase stored forms, matching exactly what encryption_material_to_metadata +// persists. The read-path SSE-C check is case-sensitive, so restoring under any +// other casing would classify the replica as managed-SSE and reject SSE-C GETs. +use super::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_MD5_HEADER}; pub const INTERNAL_ENCRYPTION_KEY_ID_HEADER: &str = "x-rustfs-encryption-key-id"; pub const INTERNAL_ENCRYPTION_KEY_HEADER: &str = "x-rustfs-encryption-key"; @@ -56,8 +59,8 @@ pub const REPLICATION_ENCRYPTED_MULTIPART_HEADER: &str = "X-Rustfs-Replication-E /// Source keys must match what `encryption_material_to_metadata` persists; the /// reconciliation test in `rustfs::storage::sse` pins that correspondence. pub const SSEC_REPLICATION_TRANSPORT_HEADERS: &[(&str, &str)] = &[ - (AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, REPLICATION_SSEC_ALGORITHM_HEADER), - (AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5, REPLICATION_SSEC_KEY_MD5_HEADER), + (SSEC_ALGORITHM_HEADER, REPLICATION_SSEC_ALGORITHM_HEADER), + (SSEC_KEY_MD5_HEADER, REPLICATION_SSEC_KEY_MD5_HEADER), (SSEC_ORIGINAL_SIZE_HEADER, REPLICATION_SSEC_ORIGINAL_SIZE_HEADER), (INTERNAL_ENCRYPTION_IV_HEADER, REPLICATION_ENCRYPTION_IV_HEADER), (MINIO_INTERNAL_ENCRYPTION_IV_HEADER, REPLICATION_SSE_IV_HEADER), @@ -74,6 +77,41 @@ pub const REPLICATION_SSE_TRANSPORT_PREFIXES: &[&str] = &[ "x-rustfs-replication-ssec-", ]; +/// Returns true when the request carries any SSE-C replication transport +/// header — the signal that an authorized replication PUT is a ciphertext +/// passthrough and the receiver must not re-encrypt or compress the body. +pub fn has_ssec_transport_headers(headers: &http::HeaderMap) -> bool { + headers.keys().any(|name| { + let name = name.as_str(); + REPLICATION_SSE_TRANSPORT_PREFIXES + .iter() + .any(|prefix| super::starts_with_ignore_ascii_case(name, prefix)) + || name.eq_ignore_ascii_case(REPLICATION_ENCRYPTED_MULTIPART_HEADER) + }) +} + +/// Restores the stored SSE-C metadata keys from their replication transport +/// names. Returns None when the request carries no transport headers. When the +/// customer algorithm is present, the AES256 SSE marker is re-added so the +/// restored metadata matches the shape `encryption_material_to_metadata` +/// persists (SSE-C Direct writes both IV twins; each travels under its own +/// transport name, so the 1:1 reverse mapping restores the dual-key pair). +pub fn ssec_transport_to_stored_metadata(headers: &http::HeaderMap) -> Option> { + let mut restored = std::collections::HashMap::new(); + for (stored, transport) in SSEC_REPLICATION_TRANSPORT_HEADERS { + if let Some(value) = headers.get(*transport).and_then(|value| value.to_str().ok()) { + restored.insert((*stored).to_string(), value.to_string()); + } + } + if restored.is_empty() { + return None; + } + if restored.contains_key(SSEC_ALGORITHM_HEADER) { + restored.insert("x-amz-server-side-encryption".to_string(), "AES256".to_string()); + } + Some(restored) +} + /// Maps a stored SSE-C metadata key to its replication transport name. pub fn ssec_replication_transport_header(stored_key: &str) -> Option<&'static str> { SSEC_REPLICATION_TRANSPORT_HEADERS @@ -100,6 +138,45 @@ pub fn is_replication_stripped_encryption_key(key: &str) -> bool { mod tests { use super::*; + #[test] + fn transport_metadata_roundtrip_restores_stored_keys() { + let mut headers = http::HeaderMap::new(); + headers.insert( + http::HeaderName::from_static("x-rustfs-replication-ssec-algorithm"), + http::HeaderValue::from_static("AES256"), + ); + headers.insert( + http::HeaderName::from_static("x-rustfs-replication-encryption-iv"), + http::HeaderValue::from_static("iv-direct"), + ); + headers.insert( + http::HeaderName::from_static("x-rustfs-replication-server-side-encryption-iv"), + http::HeaderValue::from_static("iv-minio"), + ); + + assert!(has_ssec_transport_headers(&headers)); + let restored = ssec_transport_to_stored_metadata(&headers).expect("transport headers must restore"); + // Restore MUST use the exact lowercase stored key: the read-path SSE-C + // check is case-sensitive, so a TitleCase key would classify the + // replica as managed-SSE and reject SSE-C GETs. + assert_eq!( + restored + .get("x-amz-server-side-encryption-customer-algorithm") + .map(String::as_str), + Some("AES256") + ); + assert!(!restored.keys().any(|k| k != "x-amz-server-side-encryption-customer-algorithm" + && k.eq_ignore_ascii_case("x-amz-server-side-encryption-customer-algorithm"))); + assert_eq!(restored.get(INTERNAL_ENCRYPTION_IV_HEADER).map(String::as_str), Some("iv-direct")); + assert_eq!(restored.get(MINIO_INTERNAL_ENCRYPTION_IV_HEADER).map(String::as_str), Some("iv-minio")); + // The SSE marker is re-added to match the stored SSE-C shape. + assert_eq!(restored.get("x-amz-server-side-encryption").map(String::as_str), Some("AES256")); + + let plain = http::HeaderMap::new(); + assert!(!has_ssec_transport_headers(&plain)); + assert!(ssec_transport_to_stored_metadata(&plain).is_none()); + } + #[test] fn transport_lookup_is_case_insensitive() { assert_eq!( @@ -137,7 +214,7 @@ mod tests { // SSE intent headers, including the KMS key id. assert!(is_replication_stripped_encryption_key("x-amz-server-side-encryption")); assert!(is_replication_stripped_encryption_key("x-amz-server-side-encryption-aws-kms-key-id")); - assert!(is_replication_stripped_encryption_key(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM)); + assert!(is_replication_stripped_encryption_key(SSEC_ALGORITHM_HEADER)); // is_sse_header does not cover the SSE-C original-size key; the // predicate must add it explicitly. assert!(is_replication_stripped_encryption_key(SSEC_ORIGINAL_SIZE_HEADER)); diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 1aeb732e3..bcbd27bf4 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -81,7 +81,8 @@ use rustfs_s3_ops::S3Operation; use rustfs_targets::EventName; use rustfs_utils::CompressionAlgorithm; use rustfs_utils::http::{ - SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, get_source_scheme, + SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, + SUFFIX_SOURCE_REPLICATION_REQUEST, contains_key_str, get_header, get_source_scheme, headers::{AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING, AMZ_STORAGE_CLASS}, insert_str, }; @@ -504,19 +505,24 @@ impl DefaultMultipartUsecase { .get_multipart_info(&bucket, &key, &upload_id, &opts) .await .map_err(ApiError::from)?; - EncryptionRequest { - bucket: &bucket, - key: &key, - server_side_encryption: None, - ssekms_key_id: None, - ssekms_context: None, - sse_customer_algorithm, - sse_customer_key, - sse_customer_key_md5, - content_size: 0, - principal: None, + // A ciphertext-passthrough session stores encrypted parts verbatim and + // completes without the customer key (the replication client has none), + // so the SSE-C completion check must be skipped for it. + if !contains_key_str(&multipart_info.user_defined, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT) { + EncryptionRequest { + bucket: &bucket, + key: &key, + server_side_encryption: None, + ssekms_key_id: None, + ssekms_context: None, + sse_customer_algorithm, + sse_customer_key, + sse_customer_key_md5, + content_size: 0, + principal: None, + } + .validate_multipart_ssec(&multipart_info.user_defined)?; } - .validate_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; @@ -759,7 +765,22 @@ impl DefaultMultipartUsecase { principal: session_principal.as_ref(), }; - let (effective_sse, effective_kms_key_id) = match sse_prepare_encryption(encryption_request).await? { + // SSE-C ciphertext passthrough: parts are already encrypted, so no + // session DEK is prepared; a session marker tells UploadPart to store + // the ciphertext verbatim instead of recovering encryption material. + let ciphertext_passthrough = replication_authorized + && get_header(&req.headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true") + && rustfs_utils::http::ssec_transport_to_stored_metadata(&req.headers).is_some(); + if ciphertext_passthrough { + insert_str(&mut metadata, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, "true".to_string()); + } + + let prepared_material = if ciphertext_passthrough { + None + } else { + sse_prepare_encryption(encryption_request).await? + }; + let (effective_sse, effective_kms_key_id) = match prepared_material { Some(material) => { let server_side_encryption = Some(material.server_side_encryption.clone()); let ssekms_key_id = material.kms_key_id.clone(); @@ -951,10 +972,14 @@ impl DefaultMultipartUsecase { } opts.want_checksum = reader.checksum(); - let has_ssec = fi - .user_defined - .contains_key("x-amz-server-side-encryption-customer-algorithm"); - let (server_side_encryption, ssekms_key_id) = if has_ssec { + // An SSE-C passthrough session stores ciphertext parts verbatim: no + // material recovery, no validation against the (absent) customer key. + let preserve_ciphertext = contains_key_str(&fi.user_defined, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT); + let has_ssec = !preserve_ciphertext + && fi + .user_defined + .contains_key("x-amz-server-side-encryption-customer-algorithm"); + let (server_side_encryption, ssekms_key_id) = if has_ssec || preserve_ciphertext { (None, None) } else { let sse = fi @@ -974,19 +999,21 @@ impl DefaultMultipartUsecase { }; (sse, key_id) }; - EncryptionRequest { - bucket: &bucket, - key: &key, - server_side_encryption: server_side_encryption.clone(), - ssekms_key_id: ssekms_key_id.clone(), - ssekms_context: None, - sse_customer_algorithm: sse_customer_algorithm.clone(), - sse_customer_key: sse_customer_key.clone(), - sse_customer_key_md5: sse_customer_key_md5.clone(), - content_size: actual_size, - principal: None, + if !preserve_ciphertext { + EncryptionRequest { + bucket: &bucket, + key: &key, + server_side_encryption: server_side_encryption.clone(), + ssekms_key_id: ssekms_key_id.clone(), + ssekms_context: None, + sse_customer_algorithm: sse_customer_algorithm.clone(), + sse_customer_key: sse_customer_key.clone(), + sse_customer_key_md5: sse_customer_key_md5.clone(), + content_size: actual_size, + principal: None, + } + .validate_multipart_ssec(&fi.user_defined)?; } - .validate_multipart_ssec(&fi.user_defined)?; let (requested_sse, requested_kms_key_id) = if has_ssec { let ssec_material = sse_decryption(DecryptionRequest { bucket: &bucket, diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 64f02bf57..d3213d632 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -137,8 +137,8 @@ use rustfs_utils::CompressionAlgorithm; 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_REQUEST, - get_header, + 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, @@ -5393,6 +5393,11 @@ impl DefaultObjectUsecase { if is_put_object_extract_requested(&req.headers) && !inbound_replication_put { return Box::pin(self.execute_put_object_extract(req)).await; } + // SSE-C ciphertext passthrough (authorized replication only): the body + // is already ciphertext and must be stored verbatim — no compression, + // no bucket-default encryption. + let ciphertext_passthrough = + inbound_replication_put && rustfs_utils::http::ssec_transport_to_stored_metadata(&req.headers).is_some(); let input = std::mem::take(&mut req.input); @@ -5465,7 +5470,8 @@ impl DefaultObjectUsecase { self.check_bucket_quota(&bucket, quota_operation, size as u64).await?; let ingress_stage_start = std::time::Instant::now(); - let should_compress = is_disk_compressible(&req.headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64; + let should_compress = + is_disk_compressible(&req.headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64 && !ciphertext_passthrough; let server_side_encryption_requested = server_side_encryption.is_some() || sse_customer_algorithm.is_some() || ssekms_key_id.is_some(); @@ -5575,6 +5581,13 @@ impl DefaultObjectUsecase { }) }); + if ciphertext_passthrough { + // The replica keeps the source's SSE-C metadata; the bucket + // default must not claim managed encryption on it. + effective_sse = None; + effective_kms_key_id = None; + } + // Validate SSE-C headers early: reject partial/invalid combinations per S3 spec validate_sse_headers_for_write( effective_sse.as_ref(), @@ -5778,12 +5791,20 @@ impl DefaultObjectUsecase { principal: write_principal.as_ref(), }; - let encryption_material = match sse_encryption(encryption_request).await { - Ok(material) => material, - Err(err) => { - let result = Err(err.into()); - let _ = helper.complete(&result); - return result; + // SSE-C ciphertext passthrough must skip sse_encryption entirely: an + // explicit guard is required because prepare_sse_configuration inside + // it falls back to the bucket default encryption config and would + // double-encrypt the already-encrypted body. + let encryption_material = if opts.preserve_ciphertext { + None + } else { + match sse_encryption(encryption_request).await { + Ok(material) => material, + Err(err) => { + let result = Err(err.into()); + let _ = helper.complete(&result); + return result; + } } }; @@ -8319,15 +8340,22 @@ impl DefaultObjectUsecase { { return Err(S3Error::new(S3ErrorCode::PreconditionFailed)); } - validate_sse_headers_for_read(&info.user_defined, &req.headers)?; + // An authorized replication convergence check only needs etag/size/mtime + // to compare source and replica; it holds no customer key, so the SSE-C + // read validation is skipped for it (and only it). + let replication_check = replication_request_authorized(&req) + && get_header(&req.headers, SUFFIX_SOURCE_REPLICATION_CHECK).as_deref() == Some("true"); + if !replication_check { + validate_sse_headers_for_read(&info.user_defined, &req.headers)?; - // Validate SSE-C: if the object was encrypted with a customer-provided key, - // the caller must supply the matching key even for HEAD requests (per S3 spec). - validate_ssec_for_read( - &info.user_defined, - req.input.sse_customer_key.as_ref(), - req.input.sse_customer_key_md5.as_ref(), - )?; + // Validate SSE-C: if the object was encrypted with a customer-provided key, + // the caller must supply the matching key even for HEAD requests (per S3 spec). + validate_ssec_for_read( + &info.user_defined, + req.input.sse_customer_key.as_ref(), + req.input.sse_customer_key_md5.as_ref(), + )?; + } // Compute x-amz-expiration header from lifecycle prediction (before info is partially moved) let expiration_header = resolve_put_object_expiration(&bucket, &info).await; diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index f6152bf0f..f0f8b9033 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -87,6 +87,7 @@ fn has_replication_only_put_headers(headers: &HeaderMap) -> bool { || get_header(headers, SUFFIX_SOURCE_REPLICATION_CHECK).is_some() || get_header(headers, SUFFIX_SOURCE_REPLICATION_REQUEST).is_some() || get_header(headers, SUFFIX_SOURCE_VERSION_ID).is_some() + || rustfs_utils::http::has_ssec_transport_headers(headers) } async fn authorize_replication_only_put_headers(req: &mut S3Request) -> S3Result<()> { @@ -2182,6 +2183,16 @@ impl S3Access for FS { req_info.object = Some(req.input.key.clone()); req_info.version_id = req.input.version_id.clone(); + // A replication convergence check HEADs the replica to compare + // etag/size/mtime. For SSE-C replicas the worker holds no customer key, + // so authorize it as a replication action and let the handler skip the + // SSE-C read validation. + if get_header(&req.headers, SUFFIX_SOURCE_REPLICATION_CHECK).as_deref() == Some("true") { + authorize_request(req, Action::S3Action(S3Action::ReplicateObjectAction)).await?; + req_info_mut(req)?.replication_request_authorized = true; + return Ok(()); + } + authorize_request(req, Action::S3Action(S3Action::GetObjectAction)).await } diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index 20f9a5f89..39825963b 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -427,6 +427,16 @@ pub fn put_opts_from_headers_with_replication_authorization( opts.replication_request = true; opts.mod_time = replication_source_mtime(headers); opts.preserve_etag = replication_source_etag(headers); + // SSE-C ciphertext passthrough: restore the stored encryption metadata + // from the transport headers and mark the body as already encrypted so + // the write path stores it verbatim. + if let Some(restored) = rustfs_utils::http::ssec_transport_to_stored_metadata(headers) { + opts.user_defined.extend(restored); + opts.preserve_ciphertext = true; + } + if let Some(crc) = get_header(headers, SUFFIX_REPLICATION_SSEC_CRC) { + insert_header_map(&mut opts.user_defined, SUFFIX_REPLICATION_SSEC_CRC, crc.into_owned()); + } } Ok(opts) } @@ -1470,6 +1480,56 @@ mod tests { assert!(opts.preserve_etag.is_none()); } + #[test] + fn test_put_opts_from_headers_gates_ssec_passthrough_on_authorization() { + use rustfs_utils::http::object_encryption_keys::{ + INTERNAL_ENCRYPTION_IV_HEADER, REPLICATION_ENCRYPTION_IV_HEADER, REPLICATION_SSEC_ALGORITHM_HEADER, + }; + + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); + headers.insert( + REPLICATION_SSEC_ALGORITHM_HEADER.parse::().unwrap(), + HeaderValue::from_static("AES256"), + ); + headers.insert( + REPLICATION_ENCRYPTION_IV_HEADER.parse::().unwrap(), + HeaderValue::from_static("iv-value"), + ); + + // Unauthorized: the transport headers must be inert — no restored + // encryption metadata, no ciphertext-passthrough flag. + let untrusted = put_opts_from_headers(&headers, HashMap::new()).expect("ordinary PUT options should be created"); + assert!(!untrusted.preserve_ciphertext); + assert!(!untrusted.user_defined.contains_key(INTERNAL_ENCRYPTION_IV_HEADER)); + assert!( + !untrusted + .user_defined + .contains_key("x-amz-server-side-encryption-customer-algorithm") + ); + + // Authorized: the stored keys are restored and the write path is told + // the body is already ciphertext. + let trusted = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true) + .expect("authorized replication request should parse"); + assert!(trusted.preserve_ciphertext); + assert_eq!( + trusted.user_defined.get(INTERNAL_ENCRYPTION_IV_HEADER).map(String::as_str), + Some("iv-value") + ); + assert_eq!( + trusted + .user_defined + .get("x-amz-server-side-encryption-customer-algorithm") + .map(String::as_str), + Some("AES256") + ); + assert_eq!( + trusted.user_defined.get("x-amz-server-side-encryption").map(String::as_str), + Some("AES256") + ); + } + #[test] fn test_put_opts_from_headers_accepts_replication_request_after_authorization() { let mut headers = HeaderMap::new();