fix(replication): verify replica integrity and default to plain signed payloads (#6895)

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<String>,
+    /// 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<String>,
+}
+
 #[derive(Clone)]
 pub struct PutObjectOptions {
     pub user_metadata: HashMap<String, String>,
@@ -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<Option<String>, S3ClientError> {
+    ) -> Result<RemotePutObjectResponse, S3ClientError> {
         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<std::sync::Mutex<Vec<Vec<(String, String)>>>>;
+
+    /// 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<std::sync::Mutex<Vec<(bool, bool)>>>,
@@ -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);
         }

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,
 };

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<StdMutex<HashSet<String>>> = 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<S: ReplicationObjectIO>(
             .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()
     }
 }

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;