diff --git a/.config/nextest.toml b/.config/nextest.toml index bc9a4d5b3..ee5e72657 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -280,10 +280,12 @@ slow-timeout = { period = "60s", terminate-after = 2, grace-period = "10s" } # tests that are unfit for the per-PR e2e-smoke gate: # # * 2 remote-target TLS validation tests. -# * 13 bucket-replication data-plane/helper tests — they PUT/delete objects +# * 15 bucket-replication data-plane/helper tests — they PUT/delete objects # and poll until source and target converge; two replicate over HTTPS, -# four pin active SSE fail-closed contracts (SSE-C, SSE-S3, SSE-KMS, and -# the SSE-S3 resync path), and one guards event/history observers. +# six pin SSE replication contracts (managed SSE-S3/SSE-KMS re-encrypt on +# the target incl. multipart and the resync path, SSE-C and +# target-without-KMS stay fail-closed), and one guards event/history +# observers. # * 12 `_real_dual_node` site-replication tests — each spawns TWO full rustfs # servers and drives the cross-process site-replication control plane. # * 1 `_real_three_node` site-replication test. diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 1c0e64517..5690c1a9e 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -1532,16 +1532,22 @@ async fn subscribe_to_replication_failure( async fn build_sse_replication_pair( label: &str, - enable_kms: bool, + source_kms: bool, + target_kms: bool, ) -> Result<(RustFSTestEnvironment, RustFSTestEnvironment, String, String), Box> { let mut source_env = RustFSTestEnvironment::new().await?; let mut target_env = RustFSTestEnvironment::new().await?; let source_kms_key_dir = format!("{}/kms-keys", source_env.temp_dir); let target_kms_key_dir = format!("{}/kms-keys", target_env.temp_dir); - if enable_kms { + if source_kms { fs::create_dir_all(&source_kms_key_dir).await?; - fs::create_dir_all(&target_kms_key_dir).await?; create_key_with_specific_id(&source_kms_key_dir, REPL17_KMS_KEY_ID).await?; + } + // The two sites share a key id but never key material: each side generates + // its own key, which is exactly the independent-KMS topology managed-SSE + // replication must survive (target re-encrypts with its own envelope). + if target_kms { + fs::create_dir_all(&target_kms_key_dir).await?; create_key_with_specific_id(&target_kms_key_dir, REPL17_KMS_KEY_ID).await?; } @@ -1549,7 +1555,7 @@ async fn build_sse_replication_pair( source_process_env.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); source_process_env.extend_from_slice(FAST_SCANNER_ENV); source_process_env.extend_from_slice(&[("NO_PROXY", "127.0.0.1,localhost"), ("HTTP_PROXY", ""), ("HTTPS_PROXY", "")]); - if enable_kms { + if source_kms { source_process_env.extend_from_slice(&[ ("RUSTFS_KMS_ENABLE", "true"), ("RUSTFS_KMS_BACKEND", "local"), @@ -1557,15 +1563,15 @@ async fn build_sse_replication_pair( ("RUSTFS_KMS_DEFAULT_KEY_ID", REPL17_KMS_KEY_ID), ("RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS", "true"), // Per-key KMS authorization is on so this contract is pinned in the - // configuration replication will eventually ship with: the replication - // worker carries no request identity and must stay exempt. + // configuration replication ships with: the replication worker + // carries no request identity and must stay exempt. ("RUSTFS_KMS_ENFORCE_SSE_KEY_POLICY", "true"), ]); } source_env.start_rustfs_server_with_env(vec![], &source_process_env).await?; let mut target_process_env = vec![("NO_PROXY", "127.0.0.1,localhost"), ("HTTP_PROXY", ""), ("HTTPS_PROXY", "")]; - if enable_kms { + if target_kms { target_process_env.extend_from_slice(&[ ("RUSTFS_KMS_ENABLE", "true"), ("RUSTFS_KMS_BACKEND", "local"), @@ -1593,13 +1599,12 @@ async fn build_sse_replication_pair( Ok((source_env, target_env, source_bucket, target_bucket)) } -async fn assert_managed_sse_replication_fails_explicitly(label: &str, kms: bool) -> TestResult { - let (source_env, target_env, source_bucket, target_bucket) = build_sse_replication_pair(label, true).await?; +async fn assert_managed_sse_replicates_and_reencrypts(label: &str, kms: bool) -> TestResult { + let (source_env, target_env, source_bucket, target_bucket) = build_sse_replication_pair(label, true, true).await?; let source_client = source_env.create_s3_client(); let target_client = target_env.create_s3_client(); let key = format!("{label}-contract.txt"); let body = format!("repl-17 {label} payload").into_bytes(); - let failure_events = subscribe_to_replication_failure(&source_env, &source_bucket, &key).await?; let encryption = if kms { ServerSideEncryption::AwsKms @@ -1621,20 +1626,41 @@ async fn assert_managed_sse_replication_fails_explicitly(label: &str, kms: bool) let source = source_client.get_object().bucket(&source_bucket).key(&key).send().await?; assert_eq!(source.server_side_encryption(), Some(&encryption)); + let source_etag = source.e_tag().map(str::to_string); assert_eq!(source.body.collect().await?.into_bytes().as_ref(), body.as_slice()); - wait_for_replication_failure_event(failure_events, &key).await?; - wait_for_source_replication_status(&source_client, &source_bucket, &key, "FAILED", false).await?; - assert_failed_replication_stays_absent_for( - &source_client, - &source_bucket, - &target_client, - &target_bucket, - &key, - false, - Duration::from_secs(5), - ) - .await?; + wait_for_source_replication_status(&source_client, &source_bucket, &key, "COMPLETED", false).await?; + + // The target sits on an independent KMS (same key id, different material), + // so a successful plain GET proves the replica's envelope belongs to the + // target's KMS: a forwarded source envelope could never unwrap here. + let replica = target_client.get_object().bucket(&target_bucket).key(&key).send().await?; + assert_eq!(replica.server_side_encryption(), Some(&encryption)); + let replica_version_id = replica.version_id().map(str::to_string); + let replica_etag = replica.e_tag().map(str::to_string); + assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body.as_slice()); + + // The replica must keep the source ETag; otherwise every replication HEAD + // comparison sees a mismatch and re-replicates the object forever. + assert_eq!(replica_etag, source_etag, "replica ETag must match the source ETag"); + + // Spanning several fast-scanner cycles, the replica must stay the same + // version: a second version appearing here means the ETag comparison did + // not converge and the scanner is re-driving the object. + 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.as_str())).collect(); + assert_eq!(replica_versions.len(), 1, "replica must not accumulate versions from re-replication"); + assert_eq!( + replica_versions[0].version_id().map(str::to_string), + replica_version_id, + "replica version must stay stable across scanner cycles" + ); Ok(()) } @@ -4294,6 +4320,8 @@ async fn test_single_bucket_multipart_replication_fans_out_to_multiple_targets() .create_multipart_upload() .bucket(source_bucket) .key(object_key) + .content_type("application/x-fanout") + .metadata("app", "fanout") .send() .await?; let upload_id = created.upload_id().ok_or("missing multipart upload id")?.to_string(); @@ -4341,15 +4369,17 @@ async fn test_single_bucket_multipart_replication_fans_out_to_multiple_targets() wait_for_replicated_sha256(&target_client_b, target_bucket_b, object_key, expected_sha256), )?; - let target_etag_a = target_client_a + let target_head_a = target_client_a .head_object() .bucket(target_bucket_a) .key(object_key) .send() - .await? - .e_tag() - .ok_or("first target omitted ETag")? - .to_string(); + .await?; + // Multipart replicas carry their metadata through CreateMultipartUpload; + // this pins the plaintext side of the multipart header fix. + assert_eq!(target_head_a.content_type(), Some("application/x-fanout")); + assert_eq!(target_head_a.metadata().and_then(|m| m.get("app").map(String::as_str)), Some("fanout")); + let target_etag_a = target_head_a.e_tag().ok_or("first target omitted ETag")?.to_string(); let target_etag_b = target_client_b .head_object() .bucket(target_bucket_b) @@ -4419,7 +4449,7 @@ async fn test_repl17_failure_observation_helpers() -> TestResult { async fn test_bucket_replication_sse_c_contract() -> TestResult { init_logging(); - let (source_env, target_env, source_bucket, target_bucket) = build_sse_replication_pair("ssec", false).await?; + let (source_env, target_env, source_bucket, target_bucket) = build_sse_replication_pair("ssec", false, false).await?; let source_client = source_env.create_s3_client(); let target_client = target_env.create_s3_client(); let key = "ssec-contract.txt"; @@ -4487,34 +4517,73 @@ async fn test_bucket_replication_sse_c_contract() -> TestResult { Ok(()) } -/// backlog#1147 repl-17 / backlog#1291: SSE-S3 must fail closed until managed -/// encryption is supported on the target. The silent plaintext replication -/// that originally kept this test ignored was fixed by the fail-closed gate in -/// `crates/ecstore/src/bucket/replication/replication_target_boundary.rs` -/// (all replication modes route through it), so this now pins the current -/// fail-closed contract: FAILED status, failure event, readable source, and a -/// stable absence of all target versions. +/// backlog#1147 repl-17 / backlog#1783: SSE-S3 objects replicate by decrypting +/// at the source and re-encrypting on the target with the target's own KMS. +/// The property backlog#1291 pinned — never a silent plaintext replica — still +/// holds, but the expectation flips from FAILED to a converged, decryptable +/// replica: COMPLETED status, byte-identical plain GET on the target +/// (independent KMS, so success proves target-owned envelopes), preserved +/// source ETag, and a version that stays stable across scanner cycles. #[tokio::test] #[serial] async fn test_bucket_replication_sse_s3_contract() -> TestResult { init_logging(); - assert_managed_sse_replication_fails_explicitly("sse-s3", false).await + assert_managed_sse_replicates_and_reencrypts("sse-s3", false).await } -/// P1-22 stage 0: the existing-object resync path must fail closed for -/// managed-SSE objects exactly like inline replication (which -/// `test_bucket_replication_sse_s3_contract` pins, including the scanner heal -/// re-drive). Resync re-drives every object version through the same -/// fail-closed target boundary, so a resync over an encrypted bucket must -/// terminate without ever materializing a plaintext (or unreadable) replica; -/// the post-resync stays-absent window also spans further fast-scanner heal -/// cycles. +/// backlog#1783: when the target site has no KMS, managed-SSE replication must +/// fail closed — replication FAILED, and no plaintext (or any) replica ever +/// materializes on the target. #[tokio::test] #[serial] -async fn test_bucket_replication_sse_s3_resync_stays_fail_closed() -> TestResult { +async fn test_bucket_replication_sse_s3_fails_closed_without_target_kms() -> TestResult { init_logging(); - let (source_env, target_env, source_bucket, target_bucket) = build_sse_replication_pair("sse-resync", true).await?; + let (source_env, target_env, source_bucket, target_bucket) = build_sse_replication_pair("sse-nokms", true, false).await?; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + let key = "sse-nokms-contract.txt"; + let body = b"repl-17 sse target-without-kms payload".to_vec(); + + source_client + .put_object() + .bucket(&source_bucket) + .key(key) + .body(ByteStream::from(body.clone())) + .server_side_encryption(ServerSideEncryption::Aes256) + .send() + .await?; + + wait_for_source_replication_status(&source_client, &source_bucket, key, "FAILED", false).await?; + assert_failed_replication_stays_absent_for( + &source_client, + &source_bucket, + &target_client, + &target_bucket, + key, + false, + Duration::from_secs(5), + ) + .await?; + + let source = source_client.get_object().bucket(&source_bucket).key(key).send().await?; + assert_eq!(source.server_side_encryption(), Some(&ServerSideEncryption::Aes256)); + assert_eq!(source.body.collect().await?.into_bytes().as_ref(), body.as_slice()); + + Ok(()) +} + +/// P1-22 stage 0 → backlog#1783: the existing-object resync path re-drives +/// managed-SSE objects through the same target boundary as live replication. +/// After the live pass completes, a resync over the bucket must converge — +/// the ETag comparison sees the preserved source ETag on the replica and does +/// not rewrite it, so the replica's version stays stable through the resync. +#[tokio::test] +#[serial] +async fn test_bucket_replication_sse_s3_resync_converges() -> TestResult { + init_logging(); + + let (source_env, target_env, source_bucket, target_bucket) = build_sse_replication_pair("sse-resync", true, true).await?; let source_client = source_env.create_s3_client(); let target_client = target_env.create_s3_client(); let key = "sse-resync-contract.txt"; @@ -4528,29 +4597,33 @@ async fn test_bucket_replication_sse_s3_resync_stays_fail_closed() -> TestResult .server_side_encryption(ServerSideEncryption::Aes256) .send() .await?; - wait_for_source_replication_status(&source_client, &source_bucket, key, "FAILED", false).await?; + wait_for_source_replication_status(&source_client, &source_bucket, key, "COMPLETED", false).await?; - // Resync: drive the existing-object resync path over the failed object. + let replica = target_client.get_object().bucket(&target_bucket).key(key).send().await?; + assert_eq!(replica.server_side_encryption(), Some(&ServerSideEncryption::Aes256)); + let replica_version_id = replica.version_id().map(str::to_string); + assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body.as_slice()); + + // Resync: drive the existing-object resync path over the replicated object. let (target_arn, reset_id) = start_bucket_replication_reset(&source_env, &source_bucket).await?; let terminal = wait_for_replication_reset_target(&source_env, &source_bucket, &target_arn, |target| { target.reset_id == reset_id && matches!(target.status.as_str(), "Completed" | "Failed") }) .await?; assert_eq!(terminal.reset_id, reset_id); + assert_eq!(terminal.status, "Completed", "resync over a managed-SSE bucket must complete"); + + // The resync pass must not have rewritten the converged replica. + 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, "resync must not create additional replica versions"); + assert_eq!(replica_versions[0].version_id().map(str::to_string), replica_version_id); - // The resync pass must have failed closed: still no target version (the - // window also spans further scanner heal cycles), and the source object - // stays readable and encrypted. - assert_failed_replication_stays_absent_for( - &source_client, - &source_bucket, - &target_client, - &target_bucket, - key, - false, - Duration::from_secs(5), - ) - .await?; let source = source_client.get_object().bucket(&source_bucket).key(key).send().await?; assert_eq!(source.server_side_encryption(), Some(&ServerSideEncryption::Aes256)); assert_eq!(source.body.collect().await?.into_bytes().as_ref(), body.as_slice()); @@ -4558,14 +4631,106 @@ async fn test_bucket_replication_sse_s3_resync_stays_fail_closed() -> TestResult Ok(()) } -/// backlog#1147 repl-17: SSE-KMS currently fails closed rather than creating an -/// unreadable replica; the shared helper verifies FAILED, the failure event, -/// source readability, and a stable absence of all target versions. +/// backlog#1147 repl-17 / backlog#1783: SSE-KMS replicates like SSE-S3 — the +/// source key id never crosses sites (only the aws:kms intent), and the target +/// re-encrypts under its own default key. The independent-KMS pair proves the +/// replica's envelope is target-owned. #[tokio::test] #[serial] -async fn test_bucket_replication_sse_kms_failure_contract() -> TestResult { +async fn test_bucket_replication_sse_kms_contract() -> TestResult { init_logging(); - assert_managed_sse_replication_fails_explicitly("sse-kms", true).await + assert_managed_sse_replicates_and_reencrypts("sse-kms", true).await +} + +/// backlog#1783: managed-SSE multipart objects keep their part structure and +/// their metadata through replication. CreateMultipartUpload on the target +/// carries the full header set (SSE intent, content-type, user metadata) and +/// the completed replica preserves the source's multipart ETag. +#[tokio::test] +#[serial] +async fn test_bucket_replication_sse_s3_multipart_reencrypts() -> 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("sse-mp", true, true).await?; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + let key = "sse-mp-contract.bin"; + + let created = source_client + .create_multipart_upload() + .bucket(&source_bucket) + .key(key) + .content_type("application/x-repl17") + .metadata("app", "repl17") + .server_side_encryption(ServerSideEncryption::Aes256) + .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)) + .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()) + .send() + .await?; + + wait_for_source_replication_status(&source_client, &source_bucket, key, "COMPLETED", false).await?; + + let source_head = source_client.head_object().bucket(&source_bucket).key(key).send().await?; + let replica = target_client.get_object().bucket(&target_bucket).key(key).send().await?; + assert_eq!(replica.server_side_encryption(), Some(&ServerSideEncryption::Aes256)); + assert_eq!(replica.e_tag(), source_head.e_tag(), "replica must keep the source multipart ETag"); + assert_eq!( + replica.last_modified(), + source_head.last_modified(), + "replica must keep the source mtime or the multipart HEAD comparison never converges" + ); + assert_eq!(replica.content_type(), Some("application/x-repl17")); + assert_eq!(replica.metadata().and_then(|m| m.get("app").map(String::as_str)), Some("repl17")); + let replica_version_id = replica.version_id().map(str::to_string); + assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), payload.as_slice()); + + // The multipart replica must also stay stable across scanner cycles: a + // rewritten or additional version means ETag/mtime convergence failed and + // the scanner keeps re-driving the object. + 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, "multipart replica must not accumulate versions"); + assert_eq!(replica_versions[0].version_id().map(str::to_string), replica_version_id); + + Ok(()) } /// backlog#1147 repl-5, scenario (a) — target outage + recovery (rustfs#3421 / #2071). diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 10ffb0e20..b663d5db3 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -1922,14 +1922,11 @@ impl TargetClient { object: &str, opts: &PutObjectOptions, ) -> Result { - let mut headers = HeaderMap::new(); + // Object metadata belongs to CreateMultipartUpload in S3 semantics; + // building only the source-version headers here used to drop user + // metadata, content-type, and the SSE intent for multipart replicas. + let headers = opts.header(); let version_id = opts.internal.source_version_id.clone(); - if !version_id.is_empty() { - insert_header(&mut headers, SUFFIX_SOURCE_VERSION_ID, &version_id); - } - if opts.internal.replication_request { - insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); - } // The remote version of a multipart replication is decided at initiate // time; CompleteMultipartUpload does not read a versionId. let api_version_id = resolve_put_api_version_id(&version_id).map(ToOwned::to_owned); diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 67402340d..3433aa297 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -2756,8 +2756,6 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { } }; - // TODO: SSE - if tgt_client.bucket.is_empty() { debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, @@ -3204,7 +3202,7 @@ async fn replicate_object_with_multipart(ctx: MultipartR object, &upload_id, uploaded_parts, - &replication_complete_multipart_options(actual_size), + &replication_complete_multipart_options(actual_size, object_info.etag.clone().unwrap_or_default(), object_info.mod_time), ) .await .map_err(|e| std::io::Error::other(e.to_string()))?; diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index 993e97d7a..156c94cd7 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -62,7 +62,6 @@ static STANDARD_HEADERS: &[&str] = &[ AMZ_SERVER_SIDE_ENCRYPTION, ]; -const ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED: &str = "managed SSE replication requires target encryption support"; const ERR_REPLICATION_ENCRYPTION_METADATA_UNSUPPORTED: &str = "replication source contains unsupported encryption metadata"; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -160,14 +159,8 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo) let source_encryption = classify_replication_source_encryption(&object_info.user_defined); let is_ssec = matches!(source_encryption, ReplicationSourceEncryption::SseC); - match source_encryption { - ReplicationSourceEncryption::Plaintext | ReplicationSourceEncryption::SseC => {} - ReplicationSourceEncryption::SseS3 | ReplicationSourceEncryption::SseKms => { - return Err(Error::other(ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED)); - } - ReplicationSourceEncryption::Unsupported => { - return Err(Error::other(ERR_REPLICATION_ENCRYPTION_METADATA_UNSUPPORTED)); - } + if matches!(source_encryption, ReplicationSourceEncryption::Unsupported) { + return Err(Error::other(ERR_REPLICATION_ENCRYPTION_METADATA_UNSUPPORTED)); } for (key, value) in object_info.user_defined.iter() { @@ -190,6 +183,16 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo) meta.insert(key.to_string(), value.to_string()); } + // Managed SSE replicates as plaintext (the replication reader decrypts via + // the object-encryption resolver) and re-encrypts on the target with the + // target's own KMS. Send only the encryption intent — never the source + // key id, whose meaning is local to the source site's KMS. + if matches!(source_encryption, ReplicationSourceEncryption::SseS3) { + meta.insert(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "AES256".to_string()); + } else if matches!(source_encryption, ReplicationSourceEncryption::SseKms) { + meta.insert(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "aws:kms".to_string()); + } + let mut is_multipart = object_info.is_multipart(); if let Some(checksum_data) = &object_info.checksum @@ -402,13 +405,22 @@ pub(crate) fn replication_force_delete_remove_options() -> RemoveObjectOptions { } } -pub(crate) fn replication_complete_multipart_options(actual_size: String) -> PutObjectOptions { +pub(crate) fn replication_complete_multipart_options( + actual_size: String, + source_etag: String, + source_mtime: Option, +) -> PutObjectOptions { let mut user_metadata = HashMap::new(); insert_header_map(&mut user_metadata, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, actual_size); PutObjectOptions { user_metadata, internal: AdvancedPutOptions { + source_etag, + // AdvancedPutOptions::default() stamps now_utc(); an absent source + // mtime must degrade to epoch so header() suppresses the header + // instead of asserting the replication time as the object's mtime. + source_mtime: source_mtime.unwrap_or(OffsetDateTime::UNIX_EPOCH), replication_status: ReplicationStatusType::Replica, replication_request: true, ..Default::default() @@ -573,7 +585,21 @@ mod tests { #[test] fn replication_complete_multipart_options_sets_actual_size() { - let options = replication_complete_multipart_options("1024".to_string()); + let source_mtime = OffsetDateTime::from_unix_timestamp(1_716_170_000).expect("valid test timestamp"); + let options = replication_complete_multipart_options( + "1024".to_string(), + "0123456789abcdef0123456789abcdef-3".to_string(), + Some(source_mtime), + ); + assert_eq!(options.internal.source_etag, "0123456789abcdef0123456789abcdef-3"); + assert_eq!(options.internal.source_mtime, source_mtime); + + // Absent source mtime must degrade to epoch (header suppressed), not + // the AdvancedPutOptions default of now_utc() — that default would + // stamp the replication time as the replica's mtime and break the + // multipart HEAD convergence. + let options_no_mtime = replication_complete_multipart_options("1024".to_string(), String::new(), None); + assert_eq!(options_no_mtime.internal.source_mtime.unix_timestamp(), 0); assert_eq!( get_header_map(&options.user_metadata, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE).as_deref(), @@ -809,36 +835,75 @@ mod tests { } #[test] - fn replication_put_options_rejects_sse_s3_until_target_encryption_is_supported() { - let object_info = ObjectInfo { - user_defined: Arc::new(HashMap::from([(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "AES256".to_string())])), - ..Default::default() + fn replication_put_options_sends_sse_s3_intent_without_source_material() { + use rustfs_utils::http::object_encryption_keys::{ + INTERNAL_ENCRYPTION_ALGORITHM_HEADER, INTERNAL_ENCRYPTION_IV_HEADER, INTERNAL_ENCRYPTION_KEY_HEADER, + INTERNAL_ENCRYPTION_KEY_ID_HEADER, INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER, }; - let err = match replication_put_object_options("", &object_info) { - Ok(_) => panic!("SSE-S3 replication should fail closed until target encryption headers are supported"), - Err(err) => err, - }; - - assert!(err.to_string().contains(ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED)); - } - - #[test] - fn replication_put_options_rejects_sse_kms_until_target_encryption_is_supported() { + // The stored shape of a managed SSE-S3 object per + // encryption_material_to_metadata: SSE marker plus envelope material. let object_info = ObjectInfo { user_defined: Arc::new(HashMap::from([ - (AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "aws:kms".to_string()), - (AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID.to_string(), "key-1".to_string()), + (AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "AES256".to_string()), + (INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "default".to_string()), + (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), "sealed-envelope".to_string()), + (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), "iv".to_string()), + (INTERNAL_ENCRYPTION_ALGORITHM_HEADER.to_string(), "AES256-GCM".to_string()), + (INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER.to_string(), "1024".to_string()), + ("x-user-meta".to_string(), "value".to_string()), ])), ..Default::default() }; - let err = match replication_put_object_options("", &object_info) { - Ok(_) => panic!("SSE-KMS replication should fail closed until target encryption headers are supported"), - Err(err) => err, + let (options, _) = replication_put_object_options("", &object_info).expect("managed SSE-S3 must build put options"); + + assert_eq!(options.user_metadata.get(AMZ_SERVER_SIDE_ENCRYPTION), Some(&"AES256".to_string())); + assert_eq!(options.user_metadata.get("x-user-meta"), Some(&"value".to_string())); + // No envelope material and no key id may leave the source. + assert!(!options.user_metadata.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)); + assert!(!options.user_metadata.contains_key(INTERNAL_ENCRYPTION_KEY_ID_HEADER)); + assert!(!options.user_metadata.contains_key(INTERNAL_ENCRYPTION_IV_HEADER)); + assert!( + !options.user_metadata.values().any(|value| value.contains("sealed-envelope")), + "source envelope material must never leave the source site" + ); + } + + #[test] + fn replication_put_options_sends_sse_kms_intent_without_source_key_id() { + use rustfs_utils::http::object_encryption_keys::{ + INTERNAL_ENCRYPTION_KEY_HEADER, MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER, }; - assert!(err.to_string().contains(ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED)); + let object_info = ObjectInfo { + user_defined: Arc::new(HashMap::from([ + (AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "aws:kms".to_string()), + (AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID.to_string(), "source-key-1".to_string()), + (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), "sealed-envelope".to_string()), + (MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER.to_string(), "ctx".to_string()), + ])), + ..Default::default() + }; + + let (options, _) = replication_put_object_options("", &object_info).expect("managed SSE-KMS must build put options"); + + // Intent only: the target encrypts with its own default KMS key. + assert_eq!(options.user_metadata.get(AMZ_SERVER_SIDE_ENCRYPTION), Some(&"aws:kms".to_string())); + assert!(!options.user_metadata.contains_key(AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID)); + assert!(!options.user_metadata.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)); + assert!( + !options + .user_metadata + .contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER) + ); + assert!( + !options + .user_metadata + .values() + .any(|value| value.contains("sealed-envelope") || value.contains("source-key-1")), + "source KMS identifiers and envelopes must never leave the source site" + ); } #[test] diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index b5b9f136b..a52e76677 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -1919,13 +1919,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } // etag - let etag = { - if let Some(etag) = opts.user_defined.get("etag") { - etag.clone() - } else { - get_complete_multipart_md5(&uploaded_parts) - } - }; + let etag = resolve_complete_etag(opts, &uploaded_parts); fi.metadata.insert("etag".to_owned(), etag); @@ -2167,6 +2161,21 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } } +/// Final ETag for a completed multipart object. An authorized replication +/// request preserves the source ETag so the replication HEAD comparison +/// converges even when the source ETag is not derivable from the uploaded +/// parts (foreign-origin objects, ciphertext-derived ETags); the internal +/// metadata override comes next; otherwise the ETag is computed from parts. +fn resolve_complete_etag(opts: &ObjectOptions, uploaded_parts: &[CompletePart]) -> String { + if let Some(etag) = opts.preserve_etag.as_ref().filter(|etag| !etag.is_empty()) { + return etag.clone(); + } + if let Some(etag) = opts.user_defined.get("etag") { + return etag.clone(); + } + get_complete_multipart_md5(uploaded_parts) +} + #[cfg(test)] mod tests { use super::*; @@ -5190,4 +5199,28 @@ mod tests { assert_eq!(body_after, new, "reclaiming the leftover upload must not disturb the committed object"); } } + + #[test] + fn resolve_complete_etag_prefers_preserved_source_etag() { + // A replication-preserved ETag that no part combination can derive + // (foreign-origin object) must win over the computed md5-of-parts. + let foreign_etag = "11111111111111111111111111111111-7".to_string(); + let opts = ObjectOptions { + preserve_etag: Some(foreign_etag.clone()), + ..Default::default() + }; + assert_eq!(resolve_complete_etag(&opts, &[]), foreign_etag); + + // Empty preserve value degrades to the next source. + let opts_empty = ObjectOptions { + preserve_etag: Some(String::new()), + user_defined: std::collections::HashMap::from([("etag".to_string(), "override-etag".to_string())]), + ..Default::default() + }; + assert_eq!(resolve_complete_etag(&opts_empty, &[]), "override-etag"); + + // Without either source the ETag is computed from the parts. + let computed = resolve_complete_etag(&ObjectOptions::default(), &[]); + assert_eq!(computed, get_complete_multipart_md5(&[])); + } } diff --git a/docs/testing/e2e-suite-inventory.md b/docs/testing/e2e-suite-inventory.md index c6dcb30b3..3fb476988 100644 --- a/docs/testing/e2e-suite-inventory.md +++ b/docs/testing/e2e-suite-inventory.md @@ -12,8 +12,9 @@ > lane and 28 slow / `_real_dual_node` / `_real_three_node` / `_real_single_node` tests into the > nightly lane (`.github/workflows/e2e-replication-nightly.yml`). > Note: counts exclude `#[ignore]`d tests (nextest lists them separately). -> The SSE-S3 replication contract is ignored under backlog#1291 until its -> plaintext downgrade is fixed. +> Managed-SSE (SSE-S3/SSE-KMS) replication contracts assert successful +> re-encryption on the target (backlog#1783); SSE-C replication still pins a +> fail-closed FAILED contract until ciphertext passthrough lands. | module | tests | PR smoke | |---|---|---| @@ -75,7 +76,7 @@ | quota_test | 14 | | | reliability_disk_fault_test | 3 | | | reliant | 24 | 18 ✅ | -| replication_extension_test | 48 | 20 ✅ +28 🌙 | +| replication_extension_test | 50 | 20 ✅ +30 🌙 | | security_boundary_test | 4 | | | ssec_copy_test | 2 | ✅ | | server_startup_failfast_test | 1 | | @@ -88,4 +89,4 @@ | tls_hot_reload_test | 1 | ✅ | | version_id_regression_test | 10 | ✅ | -**Total listed: 528 tests across 70 modules · PR smoke subset: 148 tests / 33 modules** (31 full modules + 18 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 28 tests** · generated 2026-08-04. +**Total listed: 530 tests across 70 modules · PR smoke subset: 148 tests / 33 modules** (31 full modules + 18 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 30 tests** · updated 2026-08-09. diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index e7fdd7120..20f9a5f89 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -18,8 +18,8 @@ use http::header::{IF_MATCH, IF_NONE_MATCH}; use http::{HeaderMap, HeaderValue}; use rustfs_utils::http::{ AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, - SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, get_header, - insert_header_map, + SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_REQUEST, + SUFFIX_SOURCE_VERSION_ID, get_header, insert_header_map, metadata_compat::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX}, }; use rustfs_utils::http::{ @@ -352,9 +352,11 @@ pub fn get_complete_multipart_upload_opts_with_replication_authorization( let mut replication_request = false; let mut mod_time = None; + let mut preserve_etag = None; if replication_request_authorized && get_header(headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true") { replication_request = true; mod_time = replication_source_mtime(headers); + preserve_etag = replication_source_etag(headers); if let Some(actual_size_str) = get_header(headers, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE) { rustfs_utils::http::insert_str( &mut user_defined, @@ -375,6 +377,7 @@ pub fn get_complete_multipart_upload_opts_with_replication_authorization( user_defined, replication_request, mod_time, + preserve_etag, ..Default::default() }; apply_replica_status_from_headers(headers, &mut opts, replication_request_authorized); @@ -423,10 +426,20 @@ pub fn put_opts_from_headers_with_replication_authorization( if replication_request_authorized && get_header(headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true") { opts.replication_request = true; opts.mod_time = replication_source_mtime(headers); + opts.preserve_etag = replication_source_etag(headers); } Ok(opts) } +/// Replicas must keep the source object's ETag: managed-SSE replication +/// re-encrypts on the target, so a recomputed ETag would differ from the +/// source and every HEAD comparison would re-schedule the object forever. +fn replication_source_etag(headers: &HeaderMap) -> Option { + let value = get_header(headers, SUFFIX_SOURCE_ETAG)?; + let value = value.trim().trim_matches('"'); + (!value.is_empty()).then(|| value.to_string()) +} + fn replication_source_mtime(headers: &HeaderMap) -> Option { let value = get_header(headers, SUFFIX_SOURCE_MTIME)?; let value = value.trim(); @@ -1026,8 +1039,8 @@ mod tests { use http::{HeaderMap, HeaderValue}; use rustfs_utils::http::{ AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, - AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_MTIME, - SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header, + AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, + SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header, }; use s3s::S3ErrorCode; use s3s::dto::{BucketVersioningStatus, ExcludedPrefix, VersioningConfiguration}; @@ -1443,6 +1456,7 @@ mod tests { insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); let valid_mtime = "2024-05-20T10:30:00+08:00"; insert_header(&mut headers, SUFFIX_SOURCE_MTIME, valid_mtime); + insert_header(&mut headers, SUFFIX_SOURCE_ETAG, "0123456789abcdef0123456789abcdef"); let metadata = HashMap::new(); @@ -1453,6 +1467,7 @@ mod tests { assert!(!opts.replication_request); assert!(opts.mod_time.is_none()); + assert!(opts.preserve_etag.is_none()); } #[test] @@ -1461,11 +1476,15 @@ mod tests { insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); let valid_mtime = "2024-05-20T10:30:00+08:00"; insert_header(&mut headers, SUFFIX_SOURCE_MTIME, valid_mtime); + insert_header(&mut headers, SUFFIX_SOURCE_ETAG, "\"0123456789abcdef0123456789abcdef-3\""); let opts = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true) .expect("authorized replication request should parse"); assert!(opts.replication_request); + // The replica keeps the source ETag verbatim (quotes trimmed) so the + // replication HEAD comparison converges after target-side re-encryption. + assert_eq!(opts.preserve_etag.as_deref(), Some("0123456789abcdef0123456789abcdef-3")); let expected_mtime = time::OffsetDateTime::parse(valid_mtime, &time::format_description::well_known::Rfc3339).unwrap(); assert_eq!(opts.mod_time, Some(expected_mtime)); @@ -1578,11 +1597,13 @@ mod tests { let mut headers = HeaderMap::new(); insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); insert_header(&mut headers, SUFFIX_SOURCE_MTIME, source_mtime); + insert_header(&mut headers, SUFFIX_SOURCE_ETAG, "\"0123456789abcdef0123456789abcdef-3\""); let untrusted = get_complete_multipart_upload_opts(&headers) .expect("ordinary multipart completion options should ignore replication headers"); assert!(!untrusted.replication_request); assert!(untrusted.mod_time.is_none()); + assert!(untrusted.preserve_etag.is_none()); let authorized = get_complete_multipart_upload_opts_with_replication_authorization(&headers, true) .expect("authorized multipart replication options should parse"); @@ -1590,6 +1611,7 @@ mod tests { .expect("test source mtime should be valid"); assert!(authorized.replication_request); assert_eq!(authorized.mod_time, Some(expected)); + assert_eq!(authorized.preserve_etag.as_deref(), Some("0123456789abcdef0123456789abcdef-3")); insert_header(&mut headers, SUFFIX_SOURCE_MTIME, "invalid-time"); let invalid = get_complete_multipart_upload_opts_with_replication_authorization(&headers, true)