diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 2966647c2..e30e3dd0a 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -24,6 +24,7 @@ use crate::runtime::sources as runtime_sources; use aws_credential_types::Credentials as SdkCredentials; use aws_credential_types::provider::{ProvideCredentials, error::CredentialsError, future}; use aws_sdk_s3::config::Region as SdkRegion; +use aws_sdk_s3::config::RequestChecksumCalculation; use aws_sdk_s3::config::SharedHttpClient; use aws_sdk_s3::error::ProvideErrorMetadata; use aws_sdk_s3::error::SdkError; @@ -39,6 +40,7 @@ use aws_sdk_s3::primitives::ByteStream; use aws_sdk_s3::types::Tagging as SdkTagging; use aws_sdk_s3::types::{ ChecksumMode, CompletedMultipartUpload, CompletedPart, ObjectLockLegalHoldStatus, ObjectLockRetentionMode, + ServerSideEncryption, }; use aws_sdk_s3::{Client as S3Client, Config as S3Config, operation::head_object::HeadObjectOutput}; use aws_sdk_s3::{config::SharedCredentialsProvider, types::BucketVersioningStatus}; @@ -1071,7 +1073,8 @@ impl BucketTargetSys { .endpoint_url(endpoint.clone()) .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider { credentials: creds })) .region(SdkRegion::new(target.region.clone())) - .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest()); + .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest()) + .request_checksum_calculation(replication_request_checksum_calculation()); if should_force_path_style(target) { config_builder = config_builder.force_path_style(true); @@ -1367,6 +1370,25 @@ fn loopback_replication_targets_allowed() -> bool { .unwrap_or(false) } +const REPLICATION_STREAMING_CHECKSUMS_ENV: &str = "RUSTFS_REPLICATION_STREAMING_CHECKSUMS"; + +/// Streaming trailer checksums make the SDK frame request bodies as +/// `aws-chunked`; a target that does not decode that framing stores the frames +/// verbatim, silently corrupting every replica while the transfer itself +/// succeeds (#6853). Plain signed payloads are the compatible default; the env +/// knob restores trailer checksums for fleets whose targets are all known to +/// decode them. +fn replication_request_checksum_calculation() -> RequestChecksumCalculation { + if std::env::var(REPLICATION_STREAMING_CHECKSUMS_ENV) + .map(|v| v.eq_ignore_ascii_case("true") || v == "1") + .unwrap_or(false) + { + RequestChecksumCalculation::WhenSupported + } else { + RequestChecksumCalculation::WhenRequired + } +} + fn validate_replication_target_endpoint(url: &Url) -> Result<(), OutboundUrlError> { validate_replication_target_endpoint_inner(url, loopback_replication_targets_allowed()) } @@ -1746,6 +1768,17 @@ impl Default for AdvancedPutOptions { } } +/// The subset of the target's PutObject response replication audits. +#[derive(Debug, Clone)] +pub struct RemotePutObjectResponse { + /// Version id the target assigned (`x-amz-version-id`). + pub version_id: Option, + /// ETag of what the target stored; `None` when the target withheld it or + /// when its encryption mode (SSE-KMS / SSE-C) makes it incomparable to + /// the source ETag. `None` is therefore "not decidable", never evidence. + pub etag: Option, +} + #[derive(Clone)] pub struct PutObjectOptions { pub user_metadata: HashMap, @@ -2291,7 +2324,9 @@ impl TargetClient { /// On success returns the version id the target assigned (from /// `x-amz-version-id`), letting callers audit the version-identity - /// contract — a target that adopts the source version echoes it back. + /// contract — a target that adopts the source version echoes it back — + /// together with the ETag of what the target actually stored, so callers + /// can detect a target that persisted transformed bytes (#6853). pub async fn put_object( &self, bucket: &str, @@ -2299,7 +2334,7 @@ impl TargetClient { size: i64, body: ByteStream, opts: &PutObjectOptions, - ) -> Result, S3ClientError> { + ) -> Result { let mut headers = opts.header(); let builder = self.client.put_object(); @@ -2334,7 +2369,25 @@ impl TargetClient { .send() .await { - Ok(output) => Ok(output.version_id().map(ToOwned::to_owned)), + Ok(output) => { + // Under SSE-KMS/DSSE or SSE-C the target's ETag is not the MD5 + // of the stored plaintext, so it cannot be compared against the + // source ETag; withhold it rather than let a caller conclude + // corruption from an opaque value. + let etag_comparable = output.sse_customer_algorithm().is_none() + && !matches!( + output.server_side_encryption(), + Some(ServerSideEncryption::AwsKms) | Some(ServerSideEncryption::AwsKmsDsse) + ); + Ok(RemotePutObjectResponse { + version_id: output.version_id().map(ToOwned::to_owned), + etag: if etag_comparable { + output.e_tag().map(ToOwned::to_owned) + } else { + None + }, + }) + } Err(e) => match e { SdkError::ServiceError(service_err) => { let err = service_err.into_err(); @@ -2673,6 +2726,145 @@ mod tests { } } + type RecordedHeaders = Arc>>>; + + /// Records full request headers and answers with canned response headers, + /// for asserting wire framing and response parsing. + #[derive(Clone, Debug)] + struct RecordingHeaderConnector { + request_headers: RecordedHeaders, + response_headers: Vec<(String, String)>, + } + + impl SmithyHttpConnector for RecordingHeaderConnector { + fn call(&self, request: HttpRequest) -> HttpConnectorFuture { + self.request_headers + .lock() + .expect("recorded header lock should not be poisoned") + .push( + request + .headers() + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + ); + let mut response = HttpResponse::new( + aws_smithy_runtime_api::http::StatusCode::try_from(200_u16).expect("200 should be a valid response status"), + SdkBody::empty(), + ); + for (name, value) in &self.response_headers { + response.headers_mut().insert(name.clone(), value.clone()); + } + HttpConnectorFuture::ready(Ok(response)) + } + } + + fn header_recording_target_client(response_headers: Vec<(String, String)>) -> (TargetClient, RecordedHeaders) { + let request_headers: RecordedHeaders = Arc::new(std::sync::Mutex::new(Vec::new())); + let connector = SharedHttpConnector::new(RecordingHeaderConnector { + request_headers: Arc::clone(&request_headers), + response_headers, + }); + let http_client = http_client_fn(move |_settings, _components| connector.clone()); + let client = s3_client_for_test(443, Some(http_client)); + ( + TargetClient { + endpoint: "https://localhost:443".to_string(), + credentials: None, + bucket: "target-bucket".to_string(), + storage_class: String::new(), + disable_proxy: false, + arn: "arn:rustfs:replication:us-east-1:target:bucket".to_string(), + reset_id: String::new(), + secure: true, + health_check_duration: Duration::from_secs(5), + replicate_sync: false, + client: Arc::new(client), + }, + request_headers, + ) + } + + fn streaming_test_body(payload: &'static [u8]) -> ByteStream { + let stream = tokio_util::io::ReaderStream::new(std::io::Cursor::new(payload)); + let body = http_body_util::StreamBody::new(futures::StreamExt::map(stream, |r| r.map(http_body::Frame::data))); + ByteStream::new(SdkBody::from_body_1_x(body)) + } + + #[test] + fn replication_checksums_default_to_plain_payloads() { + assert!(matches!( + replication_request_checksum_calculation(), + RequestChecksumCalculation::WhenRequired + )); + } + + #[tokio::test] + async fn replication_put_object_sends_plain_signed_payloads_by_default() { + let (client, recorded) = header_recording_target_client(Vec::new()); + client + .put_object("target-bucket", "object", 4, streaming_test_body(b"data"), &PutObjectOptions::default()) + .await + .expect("recorded put_object should succeed"); + + let recorded = recorded.lock().expect("recorded header lock should not be poisoned"); + let headers = &recorded[0]; + let header = |name: &str| { + headers + .iter() + .find(|(k, _)| k.eq_ignore_ascii_case(name)) + .map(|(_, v)| v.as_str()) + }; + // The #6853 regression shape: trailer checksums force aws-chunked + // framing, which a non-decoding target stores verbatim as the object. + assert_eq!(header("x-amz-trailer"), None, "streaming uploads must not carry a trailer checksum"); + assert!( + header("content-encoding").is_none_or(|v| !v.contains("aws-chunked")), + "streaming uploads must not be aws-chunked framed" + ); + assert_eq!(header("x-amz-decoded-content-length"), None); + assert_eq!(header("content-length"), Some("4")); + } + + #[tokio::test] + async fn put_object_returns_the_etag_the_target_stored() { + let (client, _) = + header_recording_target_client(vec![("etag".to_string(), "\"9a0364b9e99bb480dd25e1f0284c8555\"".to_string())]); + let response = client + .put_object( + "target-bucket", + "object", + 4, + ByteStream::from_static(b"data"), + &PutObjectOptions::default(), + ) + .await + .expect("recorded put_object should succeed"); + assert_eq!(response.etag.as_deref(), Some("\"9a0364b9e99bb480dd25e1f0284c8555\"")); + } + + #[tokio::test] + async fn put_object_withholds_the_etag_under_target_side_kms() { + let (client, _) = header_recording_target_client(vec![ + ("etag".to_string(), "\"9a0364b9e99bb480dd25e1f0284c8555\"".to_string()), + ("x-amz-server-side-encryption".to_string(), "aws:kms".to_string()), + ]); + let response = client + .put_object( + "target-bucket", + "object", + 4, + ByteStream::from_static(b"data"), + &PutObjectOptions::default(), + ) + .await + .expect("recorded put_object should succeed"); + assert!( + response.etag.is_none(), + "a KMS-encrypted replica's etag is not the content MD5 and must be withheld" + ); + } + #[derive(Clone, Debug)] struct RecordingAuthConnector { signed_requests: Arc>>, @@ -2969,7 +3161,10 @@ mod tests { .credentials_provider(SharedCredentialsProvider::new(credentials)) .region(SdkRegion::new("us-east-1")) .force_path_style(true) - .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest()); + .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest()) + // Mirror the production remote-target builder so recorded requests + // exercise the same checksum/framing behavior (#6853). + .request_checksum_calculation(replication_request_checksum_calculation()); if let Some(http_client) = http_client { config = config.http_client(http_client); } diff --git a/crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs b/crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs index c84099a30..9b6136f68 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs @@ -23,5 +23,5 @@ pub(crate) use rustfs_replication::{ delete_replication_object_opts, heal_uses_delete_replication_path, is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size, replication_multipart_part_plan, resync_existing_delete_replication_info, resync_target_for_object, - should_retry_delete_marker_purge, target_delete_version_id, + should_retry_delete_marker_purge, single_part_replica_etag_mismatch, target_delete_version_id, }; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 6076837b6..ec4b888bf 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -33,7 +33,7 @@ use super::replication_object_decision_boundary::{ delete_replication_creates_marker, heal_uses_delete_replication_path, is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size, replication_multipart_part_plan, resync_existing_delete_replication_info, should_retry_delete_marker_purge, - target_delete_version_id, + single_part_replica_etag_mismatch, target_delete_version_id, }; use super::replication_queue_boundary::{DeletedObjectReplicationInfo, ReplicationQueueAdmission}; use super::replication_resync_boundary::ResyncStatusType; @@ -54,7 +54,7 @@ use super::replication_storage_boundary::{ }; use super::replication_target_boundary::{ ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, - ReplicationTargetStore, S3ClientError, SsecPassthroughCapability, SsecPassthroughGate, TargetClient, + RemotePutObjectResponse, ReplicationTargetStore, S3ClientError, SsecPassthroughCapability, SsecPassthroughGate, TargetClient, is_replication_target_offline_error, 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, @@ -195,6 +195,53 @@ const METRIC_VERSION_IDENTITY_DRIFT_TOTAL: &str = "rustfs_replication_version_id /// after a restart is acceptable. static VERSION_IDENTITY_WARNED_ARNS: LazyLock>> = LazyLock::new(|| StdMutex::new(HashSet::new())); +const REPLICA_ETAG_VERIFY_ENV: &str = "RUSTFS_REPLICATION_REPLICA_ETAG_VERIFY"; + +/// Escape hatch for a target whose 32-hex ETags are legitimately not the +/// content MD5 (e.g. a gateway hashing its own ciphertext without announcing +/// SSE in the response) — such a target would otherwise fail every object. +fn replica_etag_verification_enabled() -> bool { + std::env::var(REPLICA_ETAG_VERIFY_ENV) + .map(|v| !(v.eq_ignore_ascii_case("false") || v == "0")) + .unwrap_or(true) +} + +/// A 200 from the target is not proof the replica holds the source bytes: a +/// target that stores a transformed payload (e.g. undecoded `aws-chunked` +/// frames, #6853) returns the ETag of what it actually wrote. Reporting +/// COMPLETED over such a replica is silent corruption, so a decidable +/// mismatch fails the replication instead. An SSE-C ciphertext passthrough +/// transfer is exempt: the wire bytes are ciphertext while the source ETag is +/// the plaintext MD5, and that path has its own HEAD-back audit. +fn verify_single_part_replica( + object_info: &ObjectInfo, + response: &RemotePutObjectResponse, + ciphertext_passthrough: bool, +) -> std::result::Result<(), std::io::Error> { + if ciphertext_passthrough || !replica_etag_verification_enabled() { + return Ok(()); + } + if single_part_replica_etag_mismatch(object_info.etag.as_deref(), response.etag.as_deref()) { + // The differing ETags go into the structured log; the error message + // stays constant so same-cause failures bucket together downstream. + warn!( + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %object_info.bucket, + object = %object_info.name, + source_etag = ?object_info.etag, + replica_etag = ?response.etag, + operation = "verify_replica_etag", + "Replication target operation failed" + ); + return Err(std::io::Error::other(REPLICA_ETAG_MISMATCH_ERROR)); + } + Ok(()) +} + +const REPLICA_ETAG_MISMATCH_ERROR: &str = "replica etag mismatch: the target persisted different bytes than were sent"; + fn audit_target_version_identity(tgt_client: &TargetClient, source_version_id: &str, assigned_version_id: Option<&str>) { if !version_identity_drifted(source_version_id, assigned_version_id) { return; @@ -3275,14 +3322,15 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { let result = tgt_client .put_object(&tgt_client.bucket, &object, transfer_size, byte_stream, &put_opts) .await - .map(|assigned_version_id| { + .map_err(|e| std::io::Error::other(e.to_string())) + .and_then(|response| { audit_target_version_identity( &tgt_client, &put_opts.internal.source_version_id, - assigned_version_id.as_deref(), - ) - }) - .map_err(|e| std::io::Error::other(e.to_string())); + response.version_id.as_deref(), + ); + verify_single_part_replica(&object_info, &response, obj_opts.raw_data_movement_read) + }); result.err() } { rinfo.replication_status = ReplicationStatusType::Failed; @@ -3943,14 +3991,15 @@ async fn replicate_all_payload_to_target( .tgt_client .put_object(&ctx.tgt_client.bucket, ctx.object, ctx.transfer_size, byte_stream, &ctx.put_opts) .await - .map(|assigned_version_id| { + .map_err(|e| std::io::Error::other(e.to_string())) + .and_then(|response| { audit_target_version_identity( ctx.tgt_client, &ctx.put_opts.internal.source_version_id, - assigned_version_id.as_deref(), - ) - }) - .map_err(|e| std::io::Error::other(e.to_string())); + response.version_id.as_deref(), + ); + verify_single_part_replica(ctx.object_info, &response, ctx.obj_opts.raw_data_movement_read) + }); result.err() } } diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index ff94159cd..194f9d85e 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -36,8 +36,8 @@ use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; pub(crate) use crate::bucket::bucket_target_sys::{ - AdvancedPutOptions, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, S3ClientError, - TargetClient, resolve_read_api_version_id, + AdvancedPutOptions, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, RemotePutObjectResponse, RemoveObjectOptions, + S3ClientError, TargetClient, resolve_read_api_version_id, }; #[cfg(test)] pub(crate) use crate::bucket::target::BucketTarget; diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index e9ab2801d..6919f0a60 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -65,7 +65,8 @@ pub use multipart::{ pub use object::{ ReplicationSourceObject, ReplicationTargetObject, SsecPassthroughCapability, SsecPassthroughGate, content_matches_by_etag, is_replication_target_offline_error, replication_action_for_target, replication_etags_match, - ssec_passthrough_evidence_present, ssec_passthrough_gate, target_is_newer_than_source_null_version, version_identity_drifted, + single_part_replica_etag_mismatch, ssec_passthrough_evidence_present, ssec_passthrough_gate, + target_is_newer_than_source_null_version, version_identity_drifted, }; pub use operation::{ MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteSource, ReplicationDeleteStateSource, diff --git a/crates/replication/src/object.rs b/crates/replication/src/object.rs index 7ce7761ca..505907df4 100644 --- a/crates/replication/src/object.rs +++ b/crates/replication/src/object.rs @@ -71,6 +71,32 @@ pub fn replication_etags_match(source: Option<&str>, target: Option<&str>) -> bo source_etag.is_some() && source_etag == target_etag } +fn is_plain_single_part_md5(etag: &str) -> bool { + etag.len() == 32 && etag.bytes().all(|b| b.is_ascii_hexdigit()) +} + +/// Whether the ETag the target returned for a single-part replica proves the +/// stored bytes differ from what the source sent — e.g. a target that does not +/// decode `aws-chunked` framing stores the frames verbatim and returns their +/// ETag. Only a plain single-part MD5 ETag on both sides is decidable; a +/// multipart or opaque (encrypted) ETag, or a withheld replica ETag, returns +/// `false` because no corruption can be concluded from it. +pub fn single_part_replica_etag_mismatch(source_etag: Option<&str>, replica_etag: Option<&str>) -> bool { + let Some(source) = source_etag.map(trim_etag) else { + return false; + }; + if !is_plain_single_part_md5(&source) { + return false; + } + let Some(replica) = replica_etag.map(trim_etag) else { + return false; + }; + if !is_plain_single_part_md5(&replica) { + return false; + } + !source.eq_ignore_ascii_case(&replica) +} + pub fn target_is_newer_than_source_null_version( source: &ReplicationSourceObject<'_>, target: &ReplicationTargetObject<'_>, @@ -276,11 +302,41 @@ pub fn ssec_passthrough_evidence_present(sse_customer_algorithm: Option<&str>) - #[cfg(test)] mod tests { + const SOURCE_MD5: &str = "9a0364b9e99bb480dd25e1f0284c8555"; + const FRAMED_MD5: &str = "0f343b0931126a20f133d67c2b018a3b"; + + #[test] + fn single_part_replica_mismatch_is_only_decided_on_plain_md5_pairs() { + // The #6853 shape: the target stored aws-chunked frames verbatim and + // returned the framed bytes' ETag. + assert!(single_part_replica_etag_mismatch(Some(SOURCE_MD5), Some(FRAMED_MD5))); + assert!(single_part_replica_etag_mismatch( + Some(&format!("\"{SOURCE_MD5}\"")), + Some(&format!("\"{FRAMED_MD5}\"")) + )); + + // A faithful replica, quoted or not, passes; hex case must not matter + // (a target may return the same MD5 uppercased). + assert!(!single_part_replica_etag_mismatch(Some(SOURCE_MD5), Some(SOURCE_MD5))); + assert!(!single_part_replica_etag_mismatch(Some(&format!("\"{SOURCE_MD5}\"")), Some(SOURCE_MD5))); + assert!(!single_part_replica_etag_mismatch( + Some(SOURCE_MD5), + Some(&SOURCE_MD5.to_ascii_uppercase()) + )); + + // Not decidable: multipart source, opaque replica ETag, or either side + // missing must never be reported as corruption. + assert!(!single_part_replica_etag_mismatch(Some(&format!("{SOURCE_MD5}-3")), Some(FRAMED_MD5))); + assert!(!single_part_replica_etag_mismatch(Some(SOURCE_MD5), Some(&format!("{FRAMED_MD5}-3")))); + assert!(!single_part_replica_etag_mismatch(Some(SOURCE_MD5), None)); + assert!(!single_part_replica_etag_mismatch(None, Some(FRAMED_MD5))); + } + use super::{ ReplicationSourceObject, ReplicationTargetObject, SsecPassthroughCapability, SsecPassthroughGate, content_matches_by_etag, is_replication_target_offline_error, replication_action_for_target, replication_etags_match, - ssec_passthrough_evidence_present, ssec_passthrough_gate, target_is_newer_than_source_null_version, - version_identity_drifted, + single_part_replica_etag_mismatch, ssec_passthrough_evidence_present, ssec_passthrough_gate, + target_is_newer_than_source_null_version, version_identity_drifted, }; use crate::filemeta::{ReplicationAction, ReplicationType}; use crate::http::AMZ_OBJECT_LOCK_MODE;