From d20476a66c2c2f09825bf04f2cb295f22faf3f24 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Tue, 18 Aug 2026 00:54:26 +0800 Subject: [PATCH] fix(replication): fail SSE-C passthrough closed on targets that drop transport headers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SSE-C ciphertext passthrough replicates via X-Rustfs-Replication-* transport headers. A MinIO/generic-S3 target silently discards them, storing bare ciphertext with no decryption material — yet the PUT succeeded, so the object reported COMPLETED with a silently unreadable replica (backlog#1675 N2). Fail-closed design: - SsecPassthroughCapability {Unknown, Supported, Unsupported} cached in BucketTargetSys per target ARN with a recording timestamp. Entries reset whenever the target is rebuilt, edited, or removed (arn_remotes_map lifecycle) and expire after SSEC_PASSTHROUGH_CAPABILITY_TTL (10 minutes): an expired verdict in either direction is re-earned through the audit, so an Unsupported target recovers automatically after an upgrade (at most one wasted PUT+HEAD audit per bad target per TTL window) and a Supported verdict cannot outlive a backend swapped behind the same endpoint. - Replication worker (replicate_object and replicate_all): fresh Unsupported targets never receive the PUT — the attempt fails immediately into the normal MRF retry channel with a "run ?replication-check to re-probe" hint. Unknown or expired verdicts are audited: after the PUT the worker HEADs the replica back through the replication-check channel (source version id mapped through resolve_read_api_version_id, so null-version objects audit correctly) and requires SSE-C evidence (the echoed customer-algorithm header); missing evidence records Unsupported and fails the attempt. Convergence HEADs are audited the same way, so a broken ciphertext replica from an earlier attempt can never launder itself into COMPLETED via an ETag match. The gate/evidence policy is pure (replication_target_boundary, staleness folded in as an input) for the M2 worker migration. - replication-check grows an SsecPassthrough probe phase: a probe PUT carrying the live transport-header shape, HEAD-back for evidence, and a machine-readable Code BucketRemoteSsecPassthroughUnsupported on failure. The probe verdict is synced into the runtime capability cache. Unlike VersionFidelity, a failed SsecPassthrough phase does NOT fail the target overall — it is a capability limit, not a broken replication contract, and a plaintext-only deployment against such a target must not turn red. - fake_s3_target: default mode now models a RustFS target (stores the transport headers, echoes SSE-C evidence); the new drop_unlisted_replication_headers mode models MinIO. The journal records whether a request carried transport headers. Receiver-echo verification: the replication-check HEAD exemption only skips SSE-C key validation; the response has always built sse-customer-algorithm from stored metadata (rustfs/src/app/object_usecase.rs), so no receiver change was needed — pinned end to end by the replication-check e2e against a real RustFS target. Rolling-upgrade constraint: RustFS targets older than the replication-check HEAD exemption (#5898) answer the audit HEAD without SSE-C evidence (or fail it outright), so SSE-C replication to such targets reports FAILED. This is deliberate — FAILED-and-retryable beats a silently undecryptable replica — and self-heals: once the target is upgraded, the next TTL expiry (or a manual ?replication-check re-probe) re-audits and records Supported. Plaintext and managed-SSE replication are unaffected. The capability cache is per-node; each node audits independently. Known limitations: - The audit judges evidence from the echoed customer-algorithm header only. A hypothetical target that preserves that one header while dropping other transport headers (partial-drop) would pass the audit; no known target behaves this way — observed targets drop the whole unknown-header family. - A mixed-version target cluster can flap the verdict between audits routed to different target nodes until the rollout completes; the TTL bounds how long each stale verdict persists. New e2e (backlog#1675 C1 + N2, red-first): fail-closed against a header-dropping fake (FAILED + no second PUT via the capability cache, journal-asserted; red run showed the old COMPLETED), replication-check reports the SsecPassthrough phase Code while the target stays OK overall, SSE-C heal convergence after a real target outage, and SSE-C existing-object resync landing a REPLICA readable with the customer key. TTL expiry in both directions is pinned at the cache and gate seams. --- crates/e2e_test/src/fake_s3_target/mod.rs | 131 +++++- .../src/replication_extension_test.rs | 421 +++++++++++++++++- crates/ecstore/src/api/mod.rs | 2 +- .../ecstore/src/bucket/bucket_target_sys.rs | 146 +++++- .../replication/replication_resyncer.rs | 231 +++++++++- .../replication_target_boundary.rs | 129 +++++- rustfs/src/admin/router.rs | 270 ++++++++++- rustfs/src/admin/storage_api.rs | 1 + 8 files changed, 1300 insertions(+), 31 deletions(-) diff --git a/crates/e2e_test/src/fake_s3_target/mod.rs b/crates/e2e_test/src/fake_s3_target/mod.rs index 658cba085..aad643043 100644 --- a/crates/e2e_test/src/fake_s3_target/mod.rs +++ b/crates/e2e_test/src/fake_s3_target/mod.rs @@ -90,6 +90,13 @@ const SOURCE_LEGALHOLD_TIMESTAMP_HEADERS: [&str; 2] = [ "x-rustfs-source-replication-legalhold-timestamp", "x-minio-source-replication-legalhold-timestamp", ]; +/// Wire prefix of the SSE-C passthrough replication transport headers +/// (`X-Rustfs-Replication-*`). In the default mode the fake stores them like a +/// RustFS target and echoes SSE-C evidence back on HEAD/GET; with +/// [`FakeS3Target::drop_unlisted_replication_headers`] it models MinIO / +/// generic S3, which silently discard unknown x-* headers. +const REPLICATION_SSE_TRANSPORT_PREFIX: &str = "x-rustfs-replication-"; +const REPLICATION_SSEC_ALGORITHM_TRANSPORT_HEADER: &str = "x-rustfs-replication-ssec-algorithm"; const RESERVED_BUCKET_PREFIXES: [&str; 3] = ["xn--", "sthree-", "amzn-s3-demo-"]; const RESERVED_BUCKET_SUFFIXES: [&str; 6] = ["-s3alias", "--ol-s3", ".mrap", "--x-s3", "--table-s3", "-an"]; @@ -166,6 +173,10 @@ pub struct ProxyHeaderSnapshot { pub ssec_algorithm: Option, pub ssec_key_present: bool, pub ssec_key_md5: Option, + /// Whether the request carried any `X-Rustfs-Replication-*` SSE-C + /// passthrough transport header, so fail-closed tests can assert the + /// sender really shipped the material a dropping target discarded. + pub ssec_transport_present: bool, } impl ProxyHeaderSnapshot { @@ -179,6 +190,9 @@ impl ProxyHeaderSnapshot { .map(bounded_journal_value), ssec_key_present: headers.contains_key("x-amz-server-side-encryption-customer-key"), ssec_key_md5: header_value(headers, &["x-amz-server-side-encryption-customer-key-md5"]).map(bounded_journal_value), + ssec_transport_present: headers + .keys() + .any(|name| name.as_str().starts_with(REPLICATION_SSE_TRANSPORT_PREFIX)), } } } @@ -213,6 +227,10 @@ struct ControlState { struct StoreState { assign_own_version_ids: bool, assign_own_multipart_version_ids: bool, + /// MinIO-like mode: silently discard non-whitelisted replication + /// transport headers instead of storing them (see + /// [`REPLICATION_SSE_TRANSPORT_PREFIX`]). + drop_unlisted_replication_headers: bool, buckets: HashMap, uploads: HashMap, total_bytes: usize, @@ -237,6 +255,9 @@ struct ObjectVersion { /// Object tags as ordered key/value pairs (PutObjectTagging replaces the /// whole set, DeleteObjectTagging clears it). tags: Vec<(String, String)>, + /// SSE-C passthrough transport headers stored with the version (RustFS + /// target behavior); empty when the drop mode discarded them. + replication_sse_headers: Vec<(String, String)>, } #[derive(Clone)] @@ -246,6 +267,7 @@ struct MultipartState { version_id: String, content_type: Option, metadata: Option>, + replication_sse_headers: Vec<(String, String)>, parts: BTreeMap, } @@ -466,6 +488,15 @@ impl FakeS3Target { /// Mint own version ids for the multipart path only — models a target /// that adopts PutObject version ids but not CreateMultipartUpload ones. + /// MinIO-like mode: silently drop every `X-Rustfs-Replication-*` SSE-C + /// passthrough transport header instead of storing it. The default (off) + /// models a RustFS target, which preserves the headers and echoes SSE-C + /// evidence (`x-amz-server-side-encryption-customer-algorithm`) on + /// HEAD/GET of the replica. + pub fn drop_unlisted_replication_headers(&self, enabled: bool) { + lock(&self.backend.store).drop_unlisted_replication_headers = enabled; + } + pub fn assign_own_multipart_version_ids(&self, enabled: bool) { lock(&self.backend.store).assign_own_multipart_version_ids = enabled; } @@ -842,6 +873,29 @@ fn new_version_id(headers: &HeaderMap, assign_own: bool) -> S3Result { Ok(version_id.to_string()) } +/// Capture the SSE-C passthrough transport headers a replication PUT carried. +/// Returns an empty set in the MinIO-like drop mode. +fn captured_replication_sse_headers(headers: &HeaderMap, drop_unlisted: bool) -> Vec<(String, String)> { + if drop_unlisted { + return Vec::new(); + } + headers + .iter() + .filter(|(name, _)| name.as_str().starts_with(REPLICATION_SSE_TRANSPORT_PREFIX)) + .filter_map(|(name, value)| Some((name.as_str().to_string(), value.to_str().ok()?.to_string()))) + .collect() +} + +/// SSE-C evidence a RustFS-like target echoes for a stored passthrough +/// replica: the customer algorithm restored from the transport headers. +fn stored_sse_customer_algorithm(version: &ObjectVersion) -> Option { + version + .replication_sse_headers + .iter() + .find(|(name, _)| name == REPLICATION_SSEC_ALGORITHM_TRANSPORT_HEADER) + .map(|(_, value)| value.clone()) +} + fn source_etag(headers: &HeaderMap) -> S3Result> { header_value(headers, &SOURCE_ETAG_HEADERS) .map(|value| validate_retained_identifier(value, "source ETag").map(|value| normalize_etag(&value))) @@ -1312,7 +1366,10 @@ impl S3 for FakeBackend { let input = req.input; let body = collect_stream(input.body, input.content_length, fault.as_ref(), &self.control).await?; validate_stored_metadata(&input.content_type, &input.metadata)?; - let assign_own = lock(&self.store).assign_own_version_ids; + let (assign_own, drop_unlisted) = { + let state = lock(&self.store); + (state.assign_own_version_ids, state.drop_unlisted_replication_headers) + }; let version_id = new_version_id(&headers, assign_own)?; let e_tag = match source_etag(&headers)? { Some(value) => value, @@ -1330,6 +1387,7 @@ impl S3 for FakeBackend { content_type: input.content_type, metadata: input.metadata, tags: Vec::new(), + replication_sse_headers: captured_replication_sse_headers(&headers, drop_unlisted), }; upsert_version(&mut lock(&self.store), &input.bucket, input.key, version)?; Ok(apply_response_fault( @@ -1350,6 +1408,7 @@ impl S3 for FakeBackend { let state = lock(&self.store); find_version(&state, &input.bucket, &input.key, input.version_id.as_deref())? }; + let sse_customer_algorithm = stored_sse_customer_algorithm(&version); Ok(apply_response_fault( S3Response::new(GetObjectOutput { body: Some(StreamingBlob::new(Body::from(version.body.clone()))), @@ -1359,6 +1418,7 @@ impl S3 for FakeBackend { e_tag: Some(ETag::Strong(version.e_tag)), last_modified: Some(version.last_modified.clone()), version_id: Some(version.version_id), + sse_customer_algorithm, ..Default::default() }), fault.as_ref(), @@ -1373,6 +1433,7 @@ impl S3 for FakeBackend { let state = lock(&self.store); find_version(&state, &input.bucket, &input.key, input.version_id.as_deref())? }; + let sse_customer_algorithm = stored_sse_customer_algorithm(&version); Ok(apply_response_fault( S3Response::new(HeadObjectOutput { content_length: Some(version.body.len() as i64), @@ -1381,6 +1442,7 @@ impl S3 for FakeBackend { e_tag: Some(ETag::Strong(version.e_tag)), last_modified: Some(version.last_modified.clone()), version_id: Some(version.version_id), + sse_customer_algorithm, ..Default::default() }), fault.as_ref(), @@ -1530,6 +1592,7 @@ impl S3 for FakeBackend { content_type: None, metadata: None, tags: Vec::new(), + replication_sse_headers: Vec::new(), }, )?; Ok(apply_response_fault( @@ -1557,9 +1620,10 @@ impl S3 for FakeBackend { ensure_upload_budget(&state)?; validate_stored_metadata(&input.content_type, &input.metadata)?; let upload_id = Uuid::new_v4().to_string(); - // Read the flag before the mutable borrow of `state.uploads` below + // Read the flags before the mutable borrow of `state.uploads` below // (and never re-lock the store: the mutex is not reentrant). let mint_own = state.assign_own_version_ids || state.assign_own_multipart_version_ids; + let drop_unlisted = state.drop_unlisted_replication_headers; let version_id = new_version_id(&headers, mint_own)?; state.uploads.insert( upload_id.clone(), @@ -1569,6 +1633,7 @@ impl S3 for FakeBackend { version_id, content_type: input.content_type, metadata: input.metadata, + replication_sse_headers: captured_replication_sse_headers(&headers, drop_unlisted), parts: BTreeMap::new(), }, ); @@ -1706,6 +1771,7 @@ impl S3 for FakeBackend { version_id: upload.version_id.clone(), content_type: upload.content_type.clone(), metadata: upload.metadata.clone(), + replication_sse_headers: upload.replication_sse_headers.clone(), parts: BTreeMap::new(), }, selected, @@ -1733,6 +1799,7 @@ impl S3 for FakeBackend { content_type: upload.content_type, metadata: upload.metadata, tags: Vec::new(), + replication_sse_headers: upload.replication_sse_headers, }; let mut state = lock(&self.store); let current = state @@ -1937,6 +2004,65 @@ mod tests { Ok(()) } + /// Default mode is RustFS-like: SSE-C passthrough transport headers are + /// stored and the customer algorithm is echoed on HEAD/GET. Drop mode is + /// MinIO-like: the headers are silently discarded, so no evidence comes + /// back — the exact difference the N2 fail-closed audit keys on. Both + /// modes journal that the sender shipped the transport headers. + #[tokio::test] + async fn ssec_passthrough_headers_echo_and_drop_modes() -> Result<(), BoxError> { + let target = FakeS3Target::start().await?; + target.create_bucket("target-bucket"); + let client = client(&target); + + let put_with_transport_headers = |key: &'static str| { + client + .put_object() + .bucket("target-bucket") + .key(key) + .body(ByteStream::from_static(b"ciphertext")) + .customize() + .map_request(move |mut request| { + let headers = request.headers_mut(); + headers.insert("x-rustfs-replication-ssec-algorithm", "AES256"); + headers.insert("x-rustfs-replication-ssec-key-md5", "AAAAAAAAAAAAAAAAAAAAAA=="); + Ok::<_, std::convert::Infallible>(request) + }) + .send() + }; + + put_with_transport_headers("kept").await?; + let head = client.head_object().bucket("target-bucket").key("kept").send().await?; + assert_eq!(head.sse_customer_algorithm(), Some("AES256")); + let get = client.get_object().bucket("target-bucket").key("kept").send().await?; + assert_eq!(get.sse_customer_algorithm(), Some("AES256")); + + target.drop_unlisted_replication_headers(true); + put_with_transport_headers("dropped").await?; + let head = client.head_object().bucket("target-bucket").key("dropped").send().await?; + assert_eq!(head.sse_customer_algorithm(), None, "drop mode must discard SSE-C evidence"); + + let requests = target.requests(); + for key in ["kept", "dropped"] { + let record = requests + .iter() + .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some(key)) + .expect("PUT must be journaled"); + assert!( + record.proxy_headers.ssec_transport_present, + "the journal must prove the sender shipped the transport headers for {key}" + ); + } + let plain_head = requests + .iter() + .find(|record| record.operation == Operation::HeadObject) + .expect("HEAD must be journaled"); + assert!(!plain_head.proxy_headers.ssec_transport_present); + + target.shutdown().await; + Ok(()) + } + macro_rules! assert_sdk_error { ($error:expr, $status:expr, $code:expr) => {{ let error = &$error; @@ -3202,6 +3328,7 @@ mod tests { version_id: index.to_string(), content_type: None, metadata: None, + replication_sse_headers: Vec::new(), parts: BTreeMap::new(), }, ); diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 2b2523c50..5da8aa0bd 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -2610,17 +2610,20 @@ async fn test_replication_check_succeeds_with_remote_target() -> Result<(), Box< assert_eq!(response.status(), StatusCode::OK); let payload: serde_json::Value = response.json().await?; - assert_eq!(payload["Status"], "OK"); + assert_eq!(payload["Status"], "OK", "{payload}"); assert_eq!(payload["ActiveMutation"], true); assert_eq!(payload["Targets"].as_array().map(Vec::len), Some(1)); - assert_eq!(payload["Targets"][0]["Status"], "OK"); - assert_eq!(payload["Targets"][0]["Phases"]["Put"]["Status"], "OK"); + assert_eq!(payload["Targets"][0]["Status"], "OK", "{payload}"); + assert_eq!(payload["Targets"][0]["Phases"]["Put"]["Status"], "OK", "{payload}"); // A RustFS target adopts the source version id, so the P1-19 // version-identity probe passes. - assert_eq!(payload["Targets"][0]["Phases"]["VersionFidelity"]["Status"], "OK"); - assert_eq!(payload["Targets"][0]["Phases"]["DeleteMarker"]["Status"], "OK"); - assert_eq!(payload["Targets"][0]["Phases"]["VersionDelete"]["Status"], "OK"); - assert_eq!(payload["Targets"][0]["Phases"]["Cleanup"]["Status"], "OK"); + assert_eq!(payload["Targets"][0]["Phases"]["VersionFidelity"]["Status"], "OK", "{payload}"); + // A RustFS target preserves the SSE-C passthrough transport headers and + // echoes the customer algorithm on the replication-check HEAD (N2). + assert_eq!(payload["Targets"][0]["Phases"]["SsecPassthrough"]["Status"], "OK", "{payload}"); + assert_eq!(payload["Targets"][0]["Phases"]["DeleteMarker"]["Status"], "OK", "{payload}"); + assert_eq!(payload["Targets"][0]["Phases"]["VersionDelete"]["Status"], "OK", "{payload}"); + assert_eq!(payload["Targets"][0]["Phases"]["Cleanup"]["Status"], "OK", "{payload}"); let target_client = target_env.create_s3_client(); let versions = target_client @@ -4649,6 +4652,410 @@ async fn test_bucket_replication_sse_c_multipart_passthrough() -> TestResult { Ok(()) } +/// N2 (backlog#1675 P1-22): SSE-C passthrough replication to a target that +/// silently drops the `X-Rustfs-Replication-*` transport headers (MinIO-like +/// behavior, modeled by the fake target's drop mode) used to report COMPLETED +/// while the replica had irrecoverably lost its decryption material — the red +/// light this test was born failing on. Fail-closed contract now under test: +/// the first attempt PUTs, HEAD-backs the replica, finds no SSE-C evidence, +/// records the target Unsupported and reports FAILED; a second SSE-C object +/// fails without any PUT reaching the target (capability cache, proven from +/// the target journal); plaintext objects still replicate COMPLETED. +#[tokio::test] +#[serial] +async fn test_ssec_replication_fails_closed_when_target_drops_passthrough_headers() -> TestResult { + init_logging(); + + let target = FakeS3Target::start().await?; + let target_bucket = "ssec-drop-dst"; + target.create_bucket(target_bucket); + target.drop_unlisted_replication_headers(true); + + let mut source_env = RustFSTestEnvironment::new().await?; + let mut env_vars = replication_fast_env(); + env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + env_vars.extend_from_slice(&[("NO_PROXY", "127.0.0.1,localhost"), ("HTTP_PROXY", ""), ("HTTPS_PROXY", "")]); + source_env.start_rustfs_server_with_env(vec![], &env_vars).await?; + + let source_bucket = "ssec-drop-src"; + let source_client = source_env.create_s3_client(); + source_client.create_bucket().bucket(source_bucket).send().await?; + enable_bucket_versioning(&source_env, source_bucket).await?; + let target_arn = set_replication_target_with_options( + &source_env, + source_bucket, + ReplicationTargetOptions { + endpoint: &target.address(), + access_key: FAKE_ACCESS_KEY, + secret_key: FAKE_SECRET_KEY, + target_bucket, + secure: false, + skip_tls_verify: false, + ca_cert_pem: None, + }, + ) + .await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + let customer_key = BASE64_STANDARD.encode(REPL17_SSEC_KEY); + let customer_key_md5 = sse_customer_key_md5_base64(REPL17_SSEC_KEY); + let put_ssec = |key: &'static str| { + source_client + .put_object() + .bucket(source_bucket) + .key(key) + .body(ByteStream::from_static(b"ssec fail-closed payload")) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + }; + + // First SSE-C object: the audit must catch the dropped material. + put_ssec("ssec-first.txt").await?; + wait_for_source_replication_status(&source_client, source_bucket, "ssec-first.txt", "FAILED", true).await?; + + let requests = target.take_requests(); + let first_put = requests + .iter() + .find(|record| record.operation == FakeTargetOperation::PutObject && record.key.as_deref() == Some("ssec-first.txt")) + .ok_or("the first SSE-C object must have been PUT (capability was Unknown)")?; + assert!( + first_put.proxy_headers.ssec_transport_present, + "the replication PUT must have shipped the SSE-C transport headers the target then dropped" + ); + assert!( + requests.iter().any(|record| { + record.operation == FakeTargetOperation::HeadObject + && record.key.as_deref() == Some("ssec-first.txt") + && record.sequence > first_put.sequence + && record.proxy_headers.replication_check.as_deref() == Some("true") + }), + "the post-PUT HEAD-back audit must have run through the replication-check channel; journal: {requests:?}" + ); + + // Second SSE-C object: the cached Unsupported verdict fails it closed + // before any PUT — including MRF retries of the first object. + put_ssec("ssec-second.txt").await?; + wait_for_source_replication_status(&source_client, source_bucket, "ssec-second.txt", "FAILED", true).await?; + assert!( + !target.requests().iter().any(|record| { + record.operation == FakeTargetOperation::PutObject + && record.key.as_deref() != Some("plain-control.txt") + && record.proxy_headers.ssec_transport_present + }), + "no further SSE-C ciphertext may reach a target recorded Unsupported; journal: {:?}", + target.requests() + ); + + // The gate is scoped to SSE-C: plaintext replication keeps working. + source_client + .put_object() + .bucket(source_bucket) + .key("plain-control.txt") + .body(ByteStream::from_static(b"plaintext control payload")) + .send() + .await?; + wait_for_source_replication_status(&source_client, source_bucket, "plain-control.txt", "COMPLETED", false).await?; + assert!(target.has_object(target_bucket, "plain-control.txt")); + + target.shutdown().await; + Ok(()) +} + +/// N2 (backlog#1675 P1-22): the admin replication-check must expose the same +/// verdict operators would otherwise only learn from failing SSE-C objects — +/// an SsecPassthrough probe phase that fails with the machine-readable +/// `BucketRemoteSsecPassthroughUnsupported` code against a header-dropping +/// target, with no probe residue left behind. The target's overall status +/// stays OK: unlike version-identity drift, dropped passthrough headers are +/// a capability limit, and a plaintext-only deployment against a MinIO-like +/// target must not turn red. +#[tokio::test] +#[serial] +async fn test_replication_check_flags_ssec_passthrough_dropping_target() -> TestResult { + init_logging(); + + let target = FakeS3Target::start().await?; + let target_bucket = "ssec-check-dst"; + target.create_bucket(target_bucket); + target.drop_unlisted_replication_headers(true); + + let mut source_env = RustFSTestEnvironment::new().await?; + let mut env_vars = replication_fast_env(); + env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + env_vars.extend_from_slice(&[("NO_PROXY", "127.0.0.1,localhost"), ("HTTP_PROXY", ""), ("HTTPS_PROXY", "")]); + source_env.start_rustfs_server_with_env(vec![], &env_vars).await?; + + let source_bucket = "ssec-check-src"; + let source_client = source_env.create_s3_client(); + source_client.create_bucket().bucket(source_bucket).send().await?; + enable_bucket_versioning(&source_env, source_bucket).await?; + let target_arn = set_replication_target_with_options( + &source_env, + source_bucket, + ReplicationTargetOptions { + endpoint: &target.address(), + access_key: FAKE_ACCESS_KEY, + secret_key: FAKE_SECRET_KEY, + target_bucket, + secure: false, + skip_tls_verify: false, + ca_cert_pem: None, + }, + ) + .await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + let response = run_replication_check(&source_env, source_bucket).await?; + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await?; + + assert_eq!( + payload["Status"], "OK", + "a capability-only SSE-C failure must not fail the check overall: {payload}" + ); + let target_report = &payload["Targets"][0]; + assert_eq!(target_report["Status"], "OK", "{payload}"); + let ssec = &target_report["Phases"]["SsecPassthrough"]; + assert_eq!(ssec["Status"], "FAILED", "SsecPassthrough phase must fail: {payload}"); + assert_eq!( + ssec["Code"], "BucketRemoteSsecPassthroughUnsupported", + "the failure must carry the machine-readable code: {payload}" + ); + // Basic replication of plaintext objects works on this target: every other + // phase passes, so the code is the discriminator operators branch on. + assert_eq!(target_report["Phases"]["Put"]["Status"], "OK", "{payload}"); + assert_eq!(target_report["Phases"]["VersionFidelity"]["Status"], "OK", "{payload}"); + assert_eq!(target_report["Phases"]["DeleteMarker"]["Status"], "OK", "{payload}"); + assert_eq!(target_report["Phases"]["VersionDelete"]["Status"], "OK", "{payload}"); + assert_eq!(target_report["Phases"]["Cleanup"]["Status"], "OK", "{payload}"); + + // The SSE-C probe PUT must have shipped the real transport header names — + // a mangled or missing header set would fail the phase for the wrong + // reason and mask a working target. + let requests = target.requests(); + assert!( + requests + .iter() + .any(|record| record.operation == FakeTargetOperation::PutObject && record.proxy_headers.ssec_transport_present), + "the SSE-C probe PUT must carry the X-Rustfs-Replication-* transport headers; journal: {requests:?}" + ); + + // No probe residue, including the SSE-C probe version. + let probe_put = requests + .into_iter() + .find(|record| record.operation == FakeTargetOperation::PutObject) + .ok_or("the probe PUT never reached the fake target")?; + let probe_key = probe_put.key.ok_or("probe PUT journal record has no key")?; + assert!( + target.stored_versions(target_bucket, &probe_key).is_empty(), + "all probe versions must be cleaned up" + ); + + target.shutdown().await; + Ok(()) +} + +/// C1 (backlog#1675 P1-22): heal-path convergence for SSE-C. An SSE-C object +/// whose live replication failed during a target outage must converge through +/// the scanner/heal compensation once the target returns — passing the N2 +/// HEAD-back audit against the recovered RustFS target — and the replica must +/// be readable with the customer key. +#[tokio::test] +#[serial] +async fn test_bucket_replication_sse_c_heals_after_target_outage() -> TestResult { + init_logging(); + + let (source_env, mut target_env, source_bucket, target_bucket) = + build_sse_replication_pair("ssec-heal", false, false).await?; + let source_client = source_env.create_s3_client(); + let key = "ssec-heal-contract.txt"; + let body = b"repl-22 ssec heal payload".to_vec(); + let customer_key = BASE64_STANDARD.encode(REPL17_SSEC_KEY); + let customer_key_md5 = sse_customer_key_md5_base64(REPL17_SSEC_KEY); + + // Target outage: the SSE-C write cannot replicate. + target_env.stop_server(); + + source_client + .put_object() + .bucket(&source_bucket) + .key(key) + .body(ByteStream::from(body.clone())) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + + // The failure is observable on the source (SSE-C HEAD needs the key). + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + let 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?; + match head.replication_status().map(|status| status.as_str()) { + Some("PENDING") | Some("FAILED") => break, + other => { + if tokio::time::Instant::now() >= deadline { + return Err(format!("source SSE-C object never reported PENDING/FAILED; last status={other:?}").into()); + } + sleep(Duration::from_millis(200)).await; + } + } + } + + // Recover the target in place; the source scanner re-drives the failure. + target_env + .restart_server_preserving_data(vec![], &[("NO_PROXY", "127.0.0.1,localhost"), ("HTTP_PROXY", ""), ("HTTPS_PROXY", "")]) + .await?; + + wait_for_source_replication_status(&source_client, &source_bucket, key, "COMPLETED", true).await?; + + // The healed replica is a REPLICA (status surfaces on HEAD) readable with + // the customer key. + let target_client = target_env.create_s3_client(); + let replica_head = target_client + .head_object() + .bucket(&target_bucket) + .key(key) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + assert_eq!( + replica_head.replication_status().map(|status| status.as_str()), + Some("REPLICA"), + "the healed copy must carry REPLICA status" + ); + 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?; + assert_eq!(replica.sse_customer_algorithm(), Some("AES256")); + assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body.as_slice()); + + Ok(()) +} + +/// C1 (backlog#1675 P1-22): existing-object resync for SSE-C. An SSE-C object +/// written BEFORE any replication config must reach the RustFS target through +/// the existing-object resync (`replicate_all` transport, N2-audited), land as +/// a REPLICA, and read back with the customer key. +#[tokio::test] +#[serial] +async fn test_bucket_replication_sse_c_existing_object_resync() -> TestResult { + init_logging(); + + let mut source_env = RustFSTestEnvironment::new().await?; + let mut source_process_env = replication_fast_env(); + 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", "")]); + source_env.start_rustfs_server_with_env(vec![], &source_process_env).await?; + + let mut target_env = RustFSTestEnvironment::new().await?; + target_env + .start_rustfs_server_without_cleanup_with_env(&[ + ("NO_PROXY", "127.0.0.1,localhost"), + ("HTTP_PROXY", ""), + ("HTTPS_PROXY", ""), + ]) + .await?; + + let source_bucket = "ssec-existing-src"; + let target_bucket = "ssec-existing-dst"; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + source_client.create_bucket().bucket(source_bucket).send().await?; + target_client.create_bucket().bucket(target_bucket).send().await?; + enable_bucket_versioning(&source_env, source_bucket).await?; + enable_bucket_versioning(&target_env, target_bucket).await?; + + // The SSE-C object exists before any replication wiring. + let key = "ssec-existing-contract.txt"; + let body = b"repl-22 ssec existing-object payload".to_vec(); + let customer_key = BASE64_STANDARD.encode(REPL17_SSEC_KEY); + let customer_key_md5 = sse_customer_key_md5_base64(REPL17_SSEC_KEY); + source_client + .put_object() + .bucket(source_bucket) + .key(key) + .body(ByteStream::from(body.clone())) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + + // Wire replication (existing-object enabled) and drive a resync. + let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + let (reset_arn, reset_id) = start_bucket_replication_reset(&source_env, source_bucket).await?; + assert_eq!(reset_arn, target_arn); + let terminal = wait_for_replication_reset_target(&source_env, source_bucket, &target_arn, |status| { + status.reset_id == reset_id && matches!(status.status.as_str(), "Completed" | "Failed") + }) + .await?; + assert_eq!(terminal.status, "Completed", "SSE-C existing-object resync must complete"); + assert!(terminal.replicated_count >= 1, "the existing SSE-C object must have been resynced"); + + // The replica is a REPLICA (status surfaces on HEAD) readable with the + // customer key. + let replica_head = target_client + .head_object() + .bucket(target_bucket) + .key(key) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + assert_eq!( + replica_head.replication_status().map(|status| status.as_str()), + Some("REPLICA"), + "the resynced copy must carry REPLICA status" + ); + 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?; + assert_eq!(replica.sse_customer_algorithm(), Some("AES256")); + assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body.as_slice()); + + // No plaintext leak: the replica stays unreadable without the key. + assert!( + target_client + .get_object() + .bucket(target_bucket) + .key(key) + .send() + .await + .is_err(), + "SSE-C replica must not be readable without the customer key" + ); + + Ok(()) +} + /// 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 diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index cafbd0ffe..949644c0d 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -32,7 +32,7 @@ pub mod bucket { pub mod bucket_target_sys { pub use crate::bucket::bucket_target_sys::{ AdvancedPutOptions, BucketTargetError, BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, - TargetClient, append_version_id_query, + SsecPassthroughCapability, TargetClient, append_version_id_query, }; } diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 480032972..3b354622f 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -299,9 +299,51 @@ struct TargetClientBuildProbe { release: Arc, } +/// Whether a replication target preserves the SSE-C passthrough transport +/// headers (`X-Rustfs-Replication-*`) end to end. +/// +/// A target that silently drops those headers (MinIO, generic S3) stores the +/// forwarded ciphertext without its decryption material — an unreadable +/// replica that used to report COMPLETED. The replication worker audits the +/// first passthrough PUT per target (HEAD-back for SSE-C evidence) and caches +/// the verdict here; a fresh `Unsupported` fails SSE-C replication closed +/// before any PUT is sent. Entries follow the `arn_remotes_map` lifecycle +/// (rebuilding or removing a target resets its capability to `Unknown`) and +/// additionally expire after [`SSEC_PASSTHROUGH_CAPABILITY_TTL`], after which +/// the next attempt re-audits. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum SsecPassthroughCapability { + #[default] + Unknown, + Supported, + Unsupported, +} + +/// How long an audited SSE-C passthrough verdict stays authoritative. +/// +/// Trade-off: without a TTL a verdict is sticky for the process lifetime — +/// an `Unsupported` target that gets upgraded (or re-probed only via +/// replication-check) would keep failing SSE-C replication forever, and the +/// fail-open twin: a `Supported` verdict would outlive a backend swapped +/// behind the same endpoint/ARN. With the TTL, a bad target costs at most +/// one wasted PUT+HEAD audit per TTL window, and a changed backend is +/// re-discovered within the same window. +pub const SSEC_PASSTHROUGH_CAPABILITY_TTL: Duration = Duration::from_secs(10 * 60); + +/// A recorded SSE-C passthrough verdict plus when it was recorded, so reads +/// can report staleness against [`SSEC_PASSTHROUGH_CAPABILITY_TTL`]. +#[derive(Debug, Clone, Copy)] +struct SsecPassthroughRecord { + capability: SsecPassthroughCapability, + recorded_at: Instant, +} + #[derive(Debug, Default)] pub struct BucketTargetSys { pub arn_remotes_map: Arc>>, + /// SSE-C passthrough capability verdicts keyed by target ARN. See + /// [`SsecPassthroughCapability`]; reset alongside `arn_remotes_map`. + ssec_passthrough_map: Arc>>, pub targets_map: Arc>>>, pub h_mutex: Arc>>, target_h_mutex: Arc>>, @@ -322,6 +364,7 @@ impl BucketTargetSys { fn new() -> Self { Self { arn_remotes_map: Arc::new(RwLock::new(HashMap::new())), + ssec_passthrough_map: Arc::new(RwLock::new(HashMap::new())), targets_map: Arc::new(RwLock::new(HashMap::new())), h_mutex: Arc::new(RwLock::new(HashMap::new())), target_h_mutex: Arc::new(RwLock::new(HashMap::new())), @@ -585,19 +628,59 @@ impl BucketTargetSys { let update_mutex = self.target_update_mutex(bucket).await; let _update_guard = update_mutex.lock().await; - // Lock order: targets_map, then arn_remotes_map, then target_h_mutex. + // Lock order: targets_map, then arn_remotes_map, then target_h_mutex, + // then ssec_passthrough_map (always last; also taken standalone by the + // capability accessors). let mut targets_map = self.targets_map.write().await; let mut arn_remotes_map = self.arn_remotes_map.write().await; let mut health_map = self.target_h_mutex.write().await; if let Some(targets) = targets_map.remove(bucket) { + let mut ssec_map = self.ssec_passthrough_map.write().await; for target in targets { arn_remotes_map.remove(&target.arn); health_map.remove(&target.arn); + ssec_map.remove(&target.arn); } } } + /// Cached SSE-C passthrough capability for a target ARN, plus whether the + /// verdict is older than [`SSEC_PASSTHROUGH_CAPABILITY_TTL`]. `(Unknown, + /// false)` when no verdict has been recorded since the target was built. + /// Staleness is computed here so the gate policy stays a pure function. + pub async fn ssec_passthrough_capability(&self, arn: &str) -> (SsecPassthroughCapability, bool) { + match self.ssec_passthrough_map.read().await.get(arn) { + Some(record) => (record.capability, record.recorded_at.elapsed() >= SSEC_PASSTHROUGH_CAPABILITY_TTL), + None => (SsecPassthroughCapability::Unknown, false), + } + } + + /// Record an audited SSE-C passthrough verdict for a target ARN. Written by + /// the replication worker's HEAD-back audit and by the replication-check + /// SsecPassthrough probe phase. + pub async fn record_ssec_passthrough_capability(&self, arn: &str, capability: SsecPassthroughCapability) { + self.ssec_passthrough_map.write().await.insert( + arn.to_string(), + SsecPassthroughRecord { + capability, + recorded_at: Instant::now(), + }, + ); + } + + /// Test hook: age an existing verdict so TTL expiry is observable without + /// waiting out the real window. + #[cfg(test)] + pub(crate) async fn backdate_ssec_passthrough_capability(&self, arn: &str, age: Duration) { + let backdated = Instant::now() + .checked_sub(age) + .expect("system uptime must exceed the backdate age"); + if let Some(record) = self.ssec_passthrough_map.write().await.get_mut(arn) { + record.recorded_at = backdated; + } + } + pub async fn set_target( &self, bucket: &str, @@ -953,15 +1036,21 @@ impl BucketTargetSys { } } - // Lock order: targets_map, then arn_remotes_map, then target_h_mutex. + // Lock order: targets_map, then arn_remotes_map, then target_h_mutex, + // then ssec_passthrough_map (always last; also taken standalone by the + // capability accessors). let mut targets_map = self.targets_map.write().await; let mut arn_remotes_map = self.arn_remotes_map.write().await; let mut health_map = self.target_h_mutex.write().await; // Remove existing targets if let Some(existing_targets) = targets_map.remove(bucket) { + let mut ssec_map = self.ssec_passthrough_map.write().await; for target in existing_targets { arn_remotes_map.remove(&target.arn); health_map.remove(&target.arn); + // A rebuilt/edited target may point at a different service: + // the SSE-C passthrough verdict must be re-audited from Unknown. + ssec_map.remove(&target.arn); self.update_bandwidth_limit(bucket, &target.arn, 0); } } @@ -1455,7 +1544,7 @@ fn resolve_put_api_version_id(source_version_id: &str) -> Option<&str> { /// RustFS represents the null version internally as the nil UUID while the S3 /// API addresses it as the literal "null" (same mapping as /// [`resolve_put_api_version_id`]); empty means "no version requested". -fn resolve_read_api_version_id(version_id: Option) -> Option { +pub(crate) fn resolve_read_api_version_id(version_id: Option) -> Option { let version_id = version_id?; let trimmed = version_id.trim(); if trimmed.is_empty() { @@ -2676,6 +2765,57 @@ mod tests { assert_eq!(health.last_online, Some(now)); } + /// N2 TTL contract, both flip directions: a recorded verdict is fresh + /// until [`SSEC_PASSTHROUGH_CAPABILITY_TTL`], then reads as expired; a + /// re-audit that records the OPPOSITE verdict replaces it as fresh. The + /// worker gate maps expired verdicts to ProceedWithAudit (pinned in + /// `replication_target_boundary`), so together this proves an Unsupported + /// target recovers to Supported through the audit once its verdict ages + /// out — and a stale Supported one is re-proven rather than trusted. + #[tokio::test] + async fn ssec_passthrough_capability_ttl_expires_and_reaudit_flips_verdict() { + let sys = BucketTargetSys::default(); + let arn = "arn:rustfs:replication:us-east-1:bucket:ssec-ttl"; + let expired_age = SSEC_PASSTHROUGH_CAPABILITY_TTL + Duration::from_secs(1); + + assert_eq!( + sys.ssec_passthrough_capability(arn).await, + (SsecPassthroughCapability::Unknown, false), + "an unrecorded target must read Unknown and never expired" + ); + + sys.record_ssec_passthrough_capability(arn, SsecPassthroughCapability::Unsupported) + .await; + assert_eq!( + sys.ssec_passthrough_capability(arn).await, + (SsecPassthroughCapability::Unsupported, false) + ); + + sys.backdate_ssec_passthrough_capability(arn, expired_age).await; + assert_eq!( + sys.ssec_passthrough_capability(arn).await, + (SsecPassthroughCapability::Unsupported, true), + "an aged-out Unsupported verdict must read expired so the gate re-audits" + ); + + // The re-audit against an upgraded target records Supported afresh. + sys.record_ssec_passthrough_capability(arn, SsecPassthroughCapability::Supported) + .await; + assert_eq!( + sys.ssec_passthrough_capability(arn).await, + (SsecPassthroughCapability::Supported, false), + "a fresh Supported verdict replaces the expired Unsupported one" + ); + + // And the fail-open twin: Supported also ages out. + sys.backdate_ssec_passthrough_capability(arn, expired_age).await; + assert_eq!( + sys.ssec_passthrough_capability(arn).await, + (SsecPassthroughCapability::Supported, true), + "an aged-out Supported verdict must read expired so the gate re-proves it" + ); + } + #[tokio::test] async fn list_targets_applies_health_stats_by_arn_and_preserves_endpoint_port() { let sys = BucketTargetSys::default(); diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 02c36179d..57bae1c90 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -49,10 +49,12 @@ use super::replication_storage_boundary::{ ReplicationDeletedObject, ReplicationObjectIO, ReplicationStorage, StatObjectOptions, StorageObjectInfoOrErr, WalkOptions, }; use super::replication_target_boundary::{ - PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, TargetClient, replication_action_for_target_head, + ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, + SsecPassthroughCapability, SsecPassthroughGate, TargetClient, replication_action_for_target_head, replication_complete_multipart_options, replication_delete_marker_purge_remove_options, replication_delete_remove_options, replication_force_delete_remove_options, replication_object_is_ssec_encrypted, replication_put_object_header_size, - replication_put_object_options, replication_target_head_is_newer_null_version, + replication_put_object_options, replication_target_head_is_newer_null_version, resolve_read_api_version_id, + ssec_passthrough_evidence_present, ssec_passthrough_gate, }; use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; @@ -279,6 +281,114 @@ async fn head_object_fallback( } } +/// Resolve the N2 fail-closed gate for an SSE-C passthrough attempt against +/// this target. Returns `Some(audit_required)` when replication may proceed; +/// on a freshly-flagged header-dropping target it settles `rinfo` as FAILED +/// (no PUT is ever sent — the object stays on the normal MRF retry channel +/// and re-audits once the verdict's TTL expires or replication-check +/// re-probes the target) and returns `None`. +async fn resolve_ssec_passthrough_gate( + ssec: bool, + tgt_client: &TargetClient, + bucket: &str, + object: &str, + rinfo: &mut ReplicatedTargetInfo, +) -> Option { + let (capability, expired) = ReplicationTargetStore::ssec_passthrough_capability(&tgt_client.arn).await; + match ssec_passthrough_gate(ssec, capability, expired) { + SsecPassthroughGate::Proceed => Some(false), + SsecPassthroughGate::ProceedWithAudit => Some(true), + SsecPassthroughGate::FailClosed => { + rinfo.replication_status = ReplicationStatusType::Failed; + rinfo.error = Some(ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED.to_string()); + warn!( + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %object, + arn = %tgt_client.arn, + operation = "ssec_passthrough_gate", + error = ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, + "Replication target operation failed" + ); + None + } + } +} + +/// Judge SSE-C passthrough evidence on a HEAD of the replica and record the +/// capability verdict for the target. Returns true when the SSE-C material +/// provably survived; otherwise records `Unsupported` and settles `rinfo` as +/// FAILED so the attempt never reports a silently unreadable COMPLETED. +async fn settle_ssec_passthrough_evidence( + head: &HeadObjectOutput, + tgt_client: &TargetClient, + bucket: &str, + object: &str, + rinfo: &mut ReplicatedTargetInfo, +) -> bool { + if ssec_passthrough_evidence_present(head) { + ReplicationTargetStore::record_ssec_passthrough_capability(&tgt_client.arn, SsecPassthroughCapability::Supported).await; + return true; + } + ReplicationTargetStore::record_ssec_passthrough_capability(&tgt_client.arn, SsecPassthroughCapability::Unsupported).await; + rinfo.replication_status = ReplicationStatusType::Failed; + rinfo.error = Some(ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED.to_string()); + warn!( + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %object, + arn = %tgt_client.arn, + endpoint = %tgt_client.endpoint, + operation = "ssec_passthrough_audit", + error = ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, + "Replication target operation failed" + ); + false +} + +/// Post-PUT HEAD-back audit for an SSE-C passthrough replica, over the worker +/// HEAD channel (replication-check exemption plus the `source-proxy-request: +/// false` suppression header, so the target answers locally without a +/// customer key). A HEAD transport failure leaves the capability `Unknown` +/// but still fails this attempt: an unverifiable SSE-C replica must not +/// report COMPLETED. +async fn audit_ssec_passthrough_replica( + tgt_client: &Arc, + bucket: &str, + object: &str, + version_id: Option, + rinfo: &mut ReplicatedTargetInfo, +) -> bool { + // Address the replica the way the PUT named it: a nil source version id + // (versioning-suspended / null-version objects) maps to the "null" + // version, so the audit HEAD does not 4xx-loop on those objects. + let version_id = resolve_read_api_version_id(version_id); + match head_object_for_worker(tgt_client.as_ref(), &tgt_client.bucket, object, version_id).await { + Ok(head) => settle_ssec_passthrough_evidence(&head, tgt_client, bucket, object, rinfo).await, + Err(e) => { + rinfo.replication_status = ReplicationStatusType::Failed; + rinfo.error = Some(format!("SSE-C passthrough audit HEAD failed: {e}")); + warn!( + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %object, + arn = %tgt_client.arn, + operation = "ssec_passthrough_audit_head", + error = %e, + "Replication target operation failed" + ); + mark_replication_target_offline_if_needed(tgt_client, &e).await; + false + } + } +} + static RESYNC_WORKER_COUNT: usize = 10; fn resync_status_duration( @@ -2872,6 +2982,21 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } + // N2 fail-closed: never PUT SSE-C ciphertext at a target known to drop + // the passthrough transport headers, and never trust a convergence HEAD + // against such a target — a previous broken replica matches by ETag. + let Some(ssec_audit_required) = resolve_ssec_passthrough_gate(self.ssec, &tgt_client, &bucket, &object, &mut rinfo).await + else { + send_local_event(EventArgs { + event_name: EventName::ObjectReplicationNotTracked.to_string(), + bucket_name: bucket.clone(), + object: self.to_object_info(), + user_agent: "Internal: [Replication]".to_string(), + ..Default::default() + }); + return rinfo; + }; + let versioned = ReplicationVersioningStore::prefix_enabled(&bucket, &object).await; let version_suspended = ReplicationVersioningStore::prefix_suspended(&bucket, &object).await; @@ -2975,6 +3100,14 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { Ok(oi) => { replication_action = replication_action_for_target_head(&object_info, &oi, self.op_type); if replication_action == ReplicationAction::None { + // An SSE-C replica only counts as converged when the same + // HEAD proves its decryption material survived; a broken + // ciphertext copy from an earlier attempt matches by ETag. + if ssec_audit_required + && !settle_ssec_passthrough_evidence(&oi, &tgt_client, &bucket, &object, &mut rinfo).await + { + return rinfo; + } rinfo.replication_status = ReplicationStatusType::Completed; rinfo.replication_resynced = true; rinfo.replication_action = ReplicationAction::None; @@ -2989,6 +3122,11 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { // Version-ID format mismatch: retry without versionId and compare ETags. match head_object_fallback(&tgt_client, &object).await { Ok(Some(oi)) if replication_etags_match(object_info.etag.as_deref(), oi.e_tag.as_deref()) => { + if ssec_audit_required + && !settle_ssec_passthrough_evidence(&oi, &tgt_client, &bucket, &object, &mut rinfo).await + { + return rinfo; + } rinfo.replication_status = ReplicationStatusType::Completed; rinfo.replication_resynced = true; rinfo.replication_action = ReplicationAction::None; @@ -3113,6 +3251,15 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } + // First SSE-C passthrough PUT against this target: verify the replica + // kept its decryption material before reporting COMPLETED. + if ssec_audit_required + && !audit_ssec_passthrough_replica(&tgt_client, &bucket, &object, self.version_id.map(|v| v.to_string()), &mut rinfo) + .await + { + return rinfo; + } + rinfo.replication_status = ReplicationStatusType::Completed; rinfo @@ -3135,6 +3282,21 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } + // N2 fail-closed: see the gate in `replicate_object` — the same policy + // applies to the metadata/existing-object transport. + let Some(ssec_audit_required) = resolve_ssec_passthrough_gate(self.ssec, &tgt_client, &bucket, &object, &mut rinfo).await + else { + send_local_event(EventArgs { + event_name: EventName::ObjectReplicationNotTracked.to_string(), + bucket_name: bucket.clone(), + object: self.to_object_info(), + user_agent: "Internal: [Replication]".to_string(), + ..Default::default() + }); + rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); + return rinfo; + }; + let versioned = ReplicationVersioningStore::prefix_enabled(&bucket, &object).await; let version_suspended = ReplicationVersioningStore::prefix_suspended(&bucket, &object).await; @@ -3173,8 +3335,19 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { let _sopts = replicate_all_stat_options(&object_info, &bucket, &tgt_client); - let Some((replication_action, object_info)) = - resolve_replicate_all_action(self, &tgt_client, &bucket, &object, object_info, start_time, &mut rinfo).await + let Some((replication_action, object_info)) = resolve_replicate_all_action( + ReplicateAllActionContext { + roi: self, + tgt_client: &tgt_client, + bucket: &bucket, + object: &object, + start_time, + ssec_audit_required, + }, + object_info, + &mut rinfo, + ) + .await else { return rinfo; }; @@ -3229,6 +3402,16 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } + // First SSE-C passthrough PUT against this target: verify the replica + // kept its decryption material before reporting COMPLETED. + if ssec_audit_required + && !audit_ssec_passthrough_replica(&tgt_client, &bucket, &object, self.version_id.map(|v| v.to_string()), &mut rinfo) + .await + { + rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); + return rinfo; + } + rinfo } @@ -3441,25 +3624,49 @@ fn apply_replication_resync_timestamp(rinfo: &mut ReplicatedTargetInfo, reset_id rinfo.replication_resynced = true; } +/// Borrowed inputs for [`resolve_replicate_all_action`]. +struct ReplicateAllActionContext<'a> { + roi: &'a ReplicateObjectInfo, + tgt_client: &'a Arc, + bucket: &'a str, + object: &'a str, + start_time: OffsetDateTime, + /// N2: the target's SSE-C passthrough capability is still `Unknown`, so a + /// converged-looking replica must additionally prove its SSE-C material + /// survived before the comparison may settle COMPLETED. + ssec_audit_required: bool, +} + /// Compare the source object against the target via HEAD and decide which /// replication action is still required. Returns `None` after fully settling /// `rinfo` when replication must stop here — either because the target already /// matches or because the comparison failed. async fn resolve_replicate_all_action( - roi: &ReplicateObjectInfo, - tgt_client: &Arc, - bucket: &str, - object: &str, + ctx: ReplicateAllActionContext<'_>, object_info: ObjectInfo, - start_time: OffsetDateTime, rinfo: &mut ReplicatedTargetInfo, ) -> Option<(ReplicationAction, ObjectInfo)> { + let ReplicateAllActionContext { + roi, + tgt_client, + bucket, + object, + start_time, + ssec_audit_required, + } = ctx; let replication_action; match head_object_for_worker(tgt_client.as_ref(), &tgt_client.bucket, object, roi.version_id.map(|v| v.to_string())).await { Ok(oi) => { replication_action = replication_action_for_target_head(&object_info, &oi, roi.op_type); rinfo.replication_status = ReplicationStatusType::Completed; if replication_action == ReplicationAction::None { + // An SSE-C replica only counts as converged when the same HEAD + // proves its decryption material survived; a broken ciphertext + // copy from an earlier attempt matches by ETag. + if ssec_audit_required && !settle_ssec_passthrough_evidence(&oi, tgt_client, bucket, object, rinfo).await { + rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); + return None; + } if roi.op_type == ReplicationType::ExistingObject && replication_target_head_is_newer_null_version(&object_info, &oi) { @@ -3509,6 +3716,12 @@ async fn resolve_replicate_all_action( match head_object_fallback(tgt_client, object).await { Ok(Some(oi)) => { replication_action = if replication_etags_match(object_info.etag.as_deref(), oi.e_tag.as_deref()) { + if ssec_audit_required + && !settle_ssec_passthrough_evidence(&oi, tgt_client, bucket, object, rinfo).await + { + rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); + return None; + } ReplicationAction::None } else { ReplicationAction::All diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index e82ab7e52..d060116d5 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -36,7 +36,8 @@ use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; pub(crate) use crate::bucket::bucket_target_sys::{ - AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient, + AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, SsecPassthroughCapability, TargetClient, + resolve_read_api_version_id, }; #[cfg(test)] pub(crate) use crate::bucket::target::BucketTarget; @@ -65,6 +66,8 @@ static STANDARD_HEADERS: &[&str] = &[ ]; const ERR_REPLICATION_ENCRYPTION_METADATA_UNSUPPORTED: &str = "replication source contains unsupported encryption metadata"; +pub(crate) const ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED: &str = "replication target does not support SSE-C passthrough: the replica would lose its decryption material \ + (run ?replication-check to re-probe)"; #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum ReplicationSourceEncryption { @@ -146,6 +149,54 @@ pub(crate) fn replication_object_is_ssec_encrypted(user_defined: &HashMap SsecPassthroughGate { + if !ssec { + return SsecPassthroughGate::Proceed; + } + // An expired verdict — Supported or Unsupported — must be re-earned: a + // stale Unsupported would otherwise stick forever after a target upgrade, + // and a stale Supported would fail open after a backend swap behind the + // same endpoint. + if expired { + return SsecPassthroughGate::ProceedWithAudit; + } + match capability { + SsecPassthroughCapability::Supported => SsecPassthroughGate::Proceed, + SsecPassthroughCapability::Unknown => SsecPassthroughGate::ProceedWithAudit, + SsecPassthroughCapability::Unsupported => SsecPassthroughGate::FailClosed, + } +} + +/// True when a replication-check HEAD of the replica proves the SSE-C +/// material survived passthrough: a RustFS target restores the transport +/// headers into the stored SSE-C keys and its HEAD echoes +/// `x-amz-server-side-encryption-customer-algorithm` (the replication-check +/// exemption skips key validation but not the metadata echo). A target that +/// dropped the headers stored a plain object and echoes nothing. +pub(crate) fn ssec_passthrough_evidence_present(head: &HeadObjectOutput) -> bool { + head.sse_customer_algorithm.as_deref().is_some_and(|algo| !algo.is_empty()) +} + pub(crate) struct ReplicationTargetStore; impl ReplicationTargetStore { @@ -165,6 +216,17 @@ impl ReplicationTargetStore { BucketTargetSys::get().mark_target_offline(target_client).await } + /// Returns the cached verdict and whether it has outlived its TTL. + pub(crate) async fn ssec_passthrough_capability(arn: &str) -> (SsecPassthroughCapability, bool) { + BucketTargetSys::get().ssec_passthrough_capability(arn).await + } + + pub(crate) async fn record_ssec_passthrough_capability(arn: &str, capability: SsecPassthroughCapability) { + BucketTargetSys::get() + .record_ssec_passthrough_capability(arn, capability) + .await + } + #[cfg(test)] pub(crate) async fn register_test_target(target_client: &Arc) { BucketTargetSys::get().arn_remotes_map.write().await.insert( @@ -898,6 +960,71 @@ mod tests { } } + /// N2 fail-closed policy: SSE-C replication may only proceed silently + /// against a target with a FRESH proof that it preserves the passthrough + /// transport headers. Unknown targets must be audited; freshly-flagged + /// dropping targets must never receive the PUT; an expired verdict in + /// EITHER direction must be re-earned through the audit — a sticky + /// Unsupported would outlive a target upgrade, and a sticky Supported + /// would fail open after a backend swap behind the same endpoint. + #[test] + fn ssec_passthrough_gate_is_fail_closed_and_ttl_bounded() { + for capability in [ + SsecPassthroughCapability::Unknown, + SsecPassthroughCapability::Supported, + SsecPassthroughCapability::Unsupported, + ] { + for expired in [false, true] { + assert_eq!( + ssec_passthrough_gate(false, capability, expired), + SsecPassthroughGate::Proceed, + "non-SSE-C objects must never be gated on the passthrough capability" + ); + } + } + assert_eq!( + ssec_passthrough_gate(true, SsecPassthroughCapability::Supported, false), + SsecPassthroughGate::Proceed + ); + assert_eq!( + ssec_passthrough_gate(true, SsecPassthroughCapability::Unknown, false), + SsecPassthroughGate::ProceedWithAudit + ); + assert_eq!( + ssec_passthrough_gate(true, SsecPassthroughCapability::Unsupported, false), + SsecPassthroughGate::FailClosed + ); + // Expiry flips both directions back to the audit. + assert_eq!( + ssec_passthrough_gate(true, SsecPassthroughCapability::Unsupported, true), + SsecPassthroughGate::ProceedWithAudit, + "an expired Unsupported verdict must allow a re-audit (upgraded target recovers without operator action)" + ); + assert_eq!( + ssec_passthrough_gate(true, SsecPassthroughCapability::Supported, true), + SsecPassthroughGate::ProceedWithAudit, + "an expired Supported verdict must be re-proven (backend swap behind the same endpoint must not fail open)" + ); + } + + #[test] + fn ssec_passthrough_evidence_requires_customer_algorithm_echo() { + let with_evidence = HeadObjectOutput::builder().sse_customer_algorithm("AES256").build(); + assert!(ssec_passthrough_evidence_present(&with_evidence)); + + let empty_algorithm = HeadObjectOutput::builder().sse_customer_algorithm("").build(); + assert!( + !ssec_passthrough_evidence_present(&empty_algorithm), + "an empty echo is not evidence of preserved SSE-C material" + ); + + let without_evidence = HeadObjectOutput::builder().e_tag("\"abc\"").content_length(8).build(); + assert!( + !ssec_passthrough_evidence_present(&without_evidence), + "a plain HEAD response must classify the target as having dropped the material" + ); + } + #[test] fn replication_put_options_adds_ssec_checksum_metadata() { let metadata = HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]); diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 9eb893792..8279f4012 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -17,7 +17,8 @@ use super::storage_api::bucket::metadata_sys; use super::storage_api::bucket::replication::{self, BucketReplicationResyncStatus, BucketStats, ReplicationStatusType}; use super::storage_api::bucket::target::{BucketTarget, BucketTargetType, BucketTargets}; use super::storage_api::bucket::target_sys::{ - BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, TargetClient, append_version_id_query, + BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, SsecPassthroughCapability, TargetClient, + append_version_id_query, }; use super::storage_api::bucket::versioning_sys::BucketVersioningSys; use super::storage_api::bucket::{AdminReplicationConfigExt as _, AdminVersioningConfigExt as _}; @@ -67,6 +68,9 @@ use rustfs_policy::policy::action::{Action, S3Action}; use rustfs_s3_types::EventName; use rustfs_signer::pre_sign_v4; use rustfs_utils::egress::{OutboundDnsResolver, OutboundPolicy}; +use rustfs_utils::http::object_encryption_keys::{ + REPLICATION_SSEC_ALGORITHM_HEADER, REPLICATION_SSEC_KEY_MD5_HEADER, REPLICATION_SSEC_ORIGINAL_SIZE_HEADER, +}; use rustfs_utils::http::{ SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK, SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, get_source_scheme, insert_header, @@ -210,6 +214,13 @@ const REPLICATION_CHECK_ERROR_MAX_BYTES: usize = 512; /// RustFS extension code (no madmin analogue): the target does not adopt the /// source version id, breaking the version-identity replication contract. const REPLICATION_CHECK_CODE_VERSION_MISMATCH: &str = "BucketRemoteTargetVersionMismatch"; +/// RustFS extension code (no madmin analogue): the target drops the +/// `X-Rustfs-Replication-*` SSE-C passthrough headers, so an SSE-C replica +/// would lose its decryption material (N2 fail-closed). +const REPLICATION_CHECK_CODE_SSEC_PASSTHROUGH: &str = "BucketRemoteSsecPassthroughUnsupported"; +/// Syntactically valid stand-in SSE-C key MD5 for the passthrough probe (the +/// probe object is never decrypted; it only has to round-trip the metadata). +const REPLICATION_CHECK_SSEC_PROBE_KEY_MD5: &str = "AAAAAAAAAAAAAAAAAAAAAA=="; #[derive(Debug, Clone, serde::Serialize)] struct ReplicationCheckResponse { @@ -251,6 +262,8 @@ struct ReplicationCheckPhases { put: ReplicationCheckPhaseStatus, #[serde(rename = "VersionFidelity")] version_fidelity: ReplicationCheckPhaseStatus, + #[serde(rename = "SsecPassthrough")] + ssec_passthrough: ReplicationCheckPhaseStatus, #[serde(rename = "DeleteMarker")] delete_marker: ReplicationCheckPhaseStatus, #[serde(rename = "VersionDelete")] @@ -1853,7 +1866,7 @@ fn build_replication_check_response(mut targets: Vec { + BucketTargetSys::get() + .record_ssec_passthrough_capability(&target.arn, SsecPassthroughCapability::Supported) + .await; + } + ("FAILED", Some(REPLICATION_CHECK_CODE_SSEC_PASSTHROUGH)) => { + BucketTargetSys::get() + .record_ssec_passthrough_capability(&target.arn, SsecPassthroughCapability::Unsupported) + .await; + } + _ => {} + } + result } @@ -2088,6 +2120,15 @@ struct ReplicationProbePutOutcome { response_version_id: Option, } +/// Outcome of the SSE-C passthrough probe: whether the HEAD-back of the probe +/// replica echoed SSE-C evidence (the customer-algorithm header a RustFS +/// target restores from the passthrough transport headers), plus the version +/// the target assigned so cleanup can address it. +struct ReplicationSsecProbeOutcome { + evidence_present: bool, + version_id: Option, +} + struct ReplicationProbeMultipartError { primary: S3ClientError, cleanup_error: Option, @@ -2110,9 +2151,14 @@ trait ReplicationProbeOperations { /// there: a target can adopt PutObject version ids and still mint its own /// for CreateMultipartUpload. async fn multipart_put(&mut self) -> Result; + /// PUT a probe version carrying the SSE-C passthrough transport headers, + /// HEAD it back through the replication-check channel, and report whether + /// the SSE-C evidence survived. Cleanup of the created version is the + /// caller's job (the outcome carries its version id). + async fn ssec_passthrough_probe(&mut self) -> Result; async fn create_delete_marker(&mut self, version_id: Option<&str>) -> Result, S3ClientError>; async fn delete_version(&mut self, version_id: Option<&str>) -> Result<(), S3ClientError>; - async fn cleanup(&mut self, known_version_ids: [Option<&str>; 3]) -> Result<(), String>; + async fn cleanup(&mut self, known_version_ids: [Option<&str>; 4]) -> Result<(), String>; } struct RemoteReplicationProbeOperations<'a> { @@ -2132,6 +2178,10 @@ impl ReplicationProbeOperations for RemoteReplicationProbeOperations<'_> { multipart_put_replication_probe_object(self.client, self.bucket, self.key, self.time).await } + async fn ssec_passthrough_probe(&mut self) -> Result { + ssec_passthrough_probe_object(self.client, self.bucket, self.key, self.time).await + } + async fn create_delete_marker(&mut self, version_id: Option<&str>) -> Result, S3ClientError> { delete_replication_probe_object( self.client, @@ -2155,7 +2205,7 @@ impl ReplicationProbeOperations for RemoteReplicationProbeOperations<'_> { .map(|_| ()) } - async fn cleanup(&mut self, known_version_ids: [Option<&str>; 3]) -> Result<(), String> { + async fn cleanup(&mut self, known_version_ids: [Option<&str>; 4]) -> Result<(), String> { cleanup_replication_probe(self.client, self.bucket, self.key, known_version_ids).await } } @@ -2176,6 +2226,7 @@ fn version_fidelity_error(api: &str, outcome: &ReplicationProbePutOutcome) -> Op async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, operations: &mut impl ReplicationProbeOperations) { let mut probe_version_id = None; let mut multipart_probe_version_id = None; + let mut ssec_probe_version_id = None; let mut delete_marker_version_id = None; let mut cleanup_required = true; let mut multipart_cleanup_error = None; @@ -2231,6 +2282,38 @@ async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, op } } + // N2: probe SSE-C passthrough with the same transport headers live + // replication sends. A target that drops them (MinIO, generic S3) stores + // the probe as a plain object and echoes no SSE-C evidence on the + // HEAD-back; SSE-C replicas there would silently lose their decryption + // material, so the target must be flagged with a machine-readable code. + // Deliberately unlike VersionFidelity, a failed SsecPassthrough phase + // does NOT fail the target overall: version-identity drift breaks the + // replication contract for every object, while dropped SSE-C passthrough + // headers only limit a capability — a plaintext-only deployment against a + // MinIO target is perfectly healthy and must not turn red. The phase's + // own FAILED + machine-readable Code remains for madmin consumers (and + // the verdict still reaches the runtime capability cache). + if result.phases.put.status == "OK" && result.phases.version_fidelity.status == "OK" { + match operations.ssec_passthrough_probe().await { + Ok(outcome) => { + ssec_probe_version_id = outcome.version_id; + if outcome.evidence_present { + result.phases.ssec_passthrough = ReplicationCheckPhaseStatus::passed(); + } else { + let error = "target drops SSE-C passthrough replication headers; \ + SSE-C replicas would lose their decryption material on this target"; + result.phases.ssec_passthrough = + ReplicationCheckPhaseStatus::failed_with_code(error, REPLICATION_CHECK_CODE_SSEC_PASSTHROUGH); + } + } + Err(err) => { + let error = format_replication_check_client_error(&err, ReplicationCheckFailureContext::ReplicateObject); + result.phases.ssec_passthrough = ReplicationCheckPhaseStatus::failed(&error); + } + } + } + if result.phases.put.status == "OK" && result.phases.version_fidelity.status == "OK" { match operations.create_delete_marker(probe_version_id.as_deref()).await { Ok(version_id) => { @@ -2259,6 +2342,7 @@ async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, op .cleanup([ probe_version_id.as_deref(), multipart_probe_version_id.as_deref(), + ssec_probe_version_id.as_deref(), delete_marker_version_id.as_deref(), ]) .await @@ -2553,6 +2637,72 @@ async fn put_replication_probe_object( }) } +/// PUT a fresh probe version carrying the SSE-C passthrough transport headers +/// (the wire shape live SSE-C replication uses), then HEAD it back through the +/// worker channel (replication-check exemption + proxy suppression). A RustFS +/// target restores the transport headers into stored SSE-C metadata and its +/// HEAD echoes `x-amz-server-side-encryption-customer-algorithm`; a target +/// that dropped the headers echoes nothing. The probe body is never SSE-C +/// encrypted — only the metadata round-trip matters — and the version is +/// deleted by the shared probe cleanup. +async fn ssec_passthrough_probe_object( + target_client: &TargetClient, + target_bucket: &str, + probe_key: &str, + now: OffsetDateTime, +) -> Result { + let options = build_replication_probe_put_options(now); + let sent_version_id = options.internal.source_version_id.clone(); + let mut headers = build_replication_probe_headers(&options); + // These are full wire names (not x-rustfs/x-minio suffixes), so they must + // be inserted verbatim — `insert_header` would mangle them. + for (name, value) in [ + (REPLICATION_SSEC_ALGORITHM_HEADER, "AES256"), + (REPLICATION_SSEC_KEY_MD5_HEADER, REPLICATION_CHECK_SSEC_PROBE_KEY_MD5), + (REPLICATION_SSEC_ORIGINAL_SIZE_HEADER, "8"), + ] { + let name = name + .parse::() + .map_err(|err| S3ClientError::new(format!("invalid ssec probe header name: {err}")))?; + let value = + HeaderValue::from_str(value).map_err(|err| S3ClientError::new(format!("invalid ssec probe header value: {err}")))?; + headers.insert(name, value); + } + + let query_version_id = sent_version_id.clone(); + let response = target_client + .client + .put_object() + .bucket(target_bucket) + .key(probe_key) + .content_length(8) + .body(AwsByteStream::from_static(b"aaaaaaaa")) + .customize() + .map_request(move |mut req| { + for (key, value) in headers.clone() { + req.headers_mut().insert(key.expect("operation should succeed"), value); + } + let uri = append_version_id_query(req.uri(), &query_version_id); + req.set_uri(uri).map_err(std::io::Error::other)?; + Result::<_, std::io::Error>::Ok(req) + }) + .send() + .await + .map_err(S3ClientError::from)?; + let version_id = response.version_id().map(ToOwned::to_owned); + + let head_version = version_id.clone().or_else(|| Some(sent_version_id.clone())); + let head = target_client + .head_object(target_bucket, probe_key, head_version) + .await + .map_err(S3ClientError::from)?; + + Ok(ReplicationSsecProbeOutcome { + evidence_present: head.sse_customer_algorithm().is_some_and(|algorithm| !algorithm.is_empty()), + version_id, + }) +} + async fn delete_replication_probe_object( target_client: &TargetClient, target_bucket: &str, @@ -3723,6 +3873,12 @@ mod tests { /// Same, for the multipart leg: a target may mirror PutObject ids and /// still mint its own at CreateMultipartUpload. minted_multipart_version_id: Option<&'static str>, + /// Transport failure of the SSE-C passthrough probe itself. + ssec_probe_error: Option<&'static str>, + /// Models a MinIO-like target that drops the SSE-C passthrough + /// headers: the probe HEAD-back echoes no SSE-C evidence. The default + /// (false) models a RustFS target that preserves them. + ssec_evidence_missing: bool, delete_marker_error: Option<&'static str>, version_delete_error: Option<&'static str>, cleanup_error: Option<&'static str>, @@ -3760,6 +3916,17 @@ mod tests { }) } + async fn ssec_passthrough_probe(&mut self) -> Result { + self.calls.push("ssec-probe"); + match self.ssec_probe_error { + Some(code) => Err(scripted_probe_error(code)), + None => Ok(ReplicationSsecProbeOutcome { + evidence_present: !self.ssec_evidence_missing, + version_id: Some("ssec-version".to_string()), + }), + } + } + async fn create_delete_marker(&mut self, _version_id: Option<&str>) -> Result, S3ClientError> { self.calls.push("delete-marker"); match self.delete_marker_error { @@ -3776,7 +3943,7 @@ mod tests { } } - async fn cleanup(&mut self, known_version_ids: [Option<&str>; 3]) -> Result<(), String> { + async fn cleanup(&mut self, known_version_ids: [Option<&str>; 4]) -> Result<(), String> { self.calls.push("cleanup"); self.cleanup_ids = known_version_ids .into_iter() @@ -3811,8 +3978,9 @@ mod tests { assert_eq!(result.phases.version_fidelity.code, Some(REPLICATION_CHECK_CODE_VERSION_MISMATCH)); assert_eq!(result.phases.delete_marker.status, "SKIPPED"); assert_eq!(result.phases.version_delete.status, "SKIPPED"); + assert_eq!(result.phases.ssec_passthrough.status, "SKIPPED"); assert_eq!(result.phases.cleanup.status, "OK"); - assert_eq!(operations.cleanup_ids, [Some("target-minted-version".to_string()), None, None]); + assert_eq!(operations.cleanup_ids, [Some("target-minted-version".to_string()), None, None, None]); } #[tokio::test] @@ -3825,6 +3993,70 @@ mod tests { assert_eq!(result.status, "OK"); assert_eq!(result.phases.version_fidelity.status, "OK"); assert_eq!(result.phases.version_fidelity.code, None); + assert_eq!(result.phases.ssec_passthrough.status, "OK"); + assert_eq!(result.phases.ssec_passthrough.code, None); + } + + /// N2: a target that drops the SSE-C passthrough transport headers must + /// fail the SsecPassthrough phase with the machine-readable code while the + /// target overall stays OK — deliberately unlike VersionFidelity: this is + /// a capability limit, not a broken replication contract, and a + /// plaintext-only deployment against such a target must not turn red. The + /// other mutation phases keep running and the probe version is cleaned up. + #[tokio::test] + async fn replication_probe_flags_ssec_passthrough_dropping_target_without_failing_target() { + let mut result = replication_check_target("arn:a", "OK", None); + let mut operations = ScriptedReplicationProbe { + ssec_evidence_missing: true, + ..Default::default() + }; + + execute_replication_probe(&mut result, &mut operations).await; + + assert_eq!( + operations.calls, + [ + "put", + "multipart-put", + "ssec-probe", + "delete-marker", + "version-delete", + "cleanup" + ] + ); + assert_eq!(result.status, "OK", "a capability-only failure must not fail the target overall"); + assert_eq!(result.error, None); + assert_eq!(result.phases.ssec_passthrough.status, "FAILED"); + assert_eq!(result.phases.ssec_passthrough.code, Some(REPLICATION_CHECK_CODE_SSEC_PASSTHROUGH)); + assert_eq!( + operations.cleanup_ids, + [ + Some("object-version".to_string()), + Some("multipart-version".to_string()), + Some("ssec-version".to_string()), + Some("marker-version".to_string()) + ] + ); + } + + /// A transport failure of the SSE-C probe is not evidence of a dropping + /// target: the phase fails without the capability code (the runtime cache + /// stays Unknown and the worker keeps auditing), and the target overall + /// stays OK. + #[tokio::test] + async fn replication_probe_ssec_transport_failure_carries_no_capability_code() { + let mut result = replication_check_target("arn:a", "OK", None); + let mut operations = ScriptedReplicationProbe { + ssec_probe_error: Some("InternalError"), + ..Default::default() + }; + + execute_replication_probe(&mut result, &mut operations).await; + + assert_eq!(result.status, "OK"); + assert_eq!(result.phases.ssec_passthrough.status, "FAILED"); + assert_eq!(result.phases.ssec_passthrough.code, None); + assert_eq!(operations.cleanup_ids[2], None, "a failed ssec probe leaves no version to clean"); } #[tokio::test] @@ -3855,12 +4087,23 @@ mod tests { execute_replication_probe(&mut result, &mut operations).await; - assert_eq!(operations.calls, ["put", "multipart-put", "delete-marker", "version-delete", "cleanup"]); + assert_eq!( + operations.calls, + [ + "put", + "multipart-put", + "ssec-probe", + "delete-marker", + "version-delete", + "cleanup" + ] + ); assert_eq!( operations.cleanup_ids, [ Some("object-version".to_string()), Some("multipart-version".to_string()), + Some("ssec-version".to_string()), None ] ); @@ -3880,12 +4123,23 @@ mod tests { execute_replication_probe(&mut result, &mut operations).await; - assert_eq!(operations.calls, ["put", "multipart-put", "delete-marker", "version-delete", "cleanup"]); + assert_eq!( + operations.calls, + [ + "put", + "multipart-put", + "ssec-probe", + "delete-marker", + "version-delete", + "cleanup" + ] + ); assert_eq!( operations.cleanup_ids, [ Some("object-version".to_string()), Some("multipart-version".to_string()), + Some("ssec-version".to_string()), Some("marker-version".to_string()) ] ); diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index ec89f5e4e..8b6fad0a1 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -196,6 +196,7 @@ pub(crate) mod bucket_target_sys { pub(crate) type PutObjectOptions = super::ecstore_bucket::bucket_target_sys::PutObjectOptions; pub(crate) type RemoveObjectOptions = super::ecstore_bucket::bucket_target_sys::RemoveObjectOptions; pub(crate) type S3ClientError = super::ecstore_bucket::bucket_target_sys::S3ClientError; + pub(crate) type SsecPassthroughCapability = super::ecstore_bucket::bucket_target_sys::SsecPassthroughCapability; pub(crate) type TargetClient = super::ecstore_bucket::bucket_target_sys::TargetClient; }