diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index dcee31d61..0503f4ce1 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -89,6 +89,7 @@ const EVENT_REPLICATION_FORCE_DELETE_SKIPPED: &str = "replication_force_delete_s const EVENT_RESYNC_TASK_FAILED: &str = "replication_resync_task_failed"; const EVENT_RESYNC_TARGET_OPERATION_FAILED: &str = "replication_resync_target_operation_failed"; const EVENT_RESYNC_RUNTIME_CHANNEL_FAILED: &str = "replication_resync_runtime_channel_failed"; +const ERR_REPLICATION_METADATA_COPY_UNSUPPORTED: &str = "metadata-only replication is not implemented"; pub(crate) const RESYNC_META_FORMAT: u16 = rustfs_replication::resync::RESYNC_META_FORMAT; pub(crate) const RESYNC_META_VERSION: u16 = rustfs_replication::resync::RESYNC_META_VERSION; @@ -2718,7 +2719,28 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { rinfo.replication_action = replication_action; if replication_action != ReplicationAction::All { - // TODO: copy object + rinfo.replication_status = ReplicationStatusType::Failed; + rinfo.error = Some(ERR_REPLICATION_METADATA_COPY_UNSUPPORTED.to_string()); + warn!( + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + object = %object, + operation = "copy_object_metadata", + error = ERR_REPLICATION_METADATA_COPY_UNSUPPORTED, + "Replication target operation failed" + ); + send_local_event(EventArgs { + event_name: EventName::ObjectReplicationNotTracked.to_string(), + bucket_name: bucket.clone(), + object: object_info, + user_agent: "Internal: [Replication]".to_string(), + ..Default::default() + }); + rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); + return rinfo; } else { let (put_opts, is_multipart) = match replication_put_object_options(&tgt_client.storage_class, &object_info) { Ok((put_opts, is_mp)) => (put_opts, is_mp), diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index 8ab618699..99462f3d7 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -20,10 +20,10 @@ use aws_sdk_s3::types::{ObjectLockLegalHoldStatus, ObjectLockRetentionMode}; use http::HeaderMap; use rustfs_utils::http::{ AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_MODE, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, - AMZ_OBJECT_TAGGING, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_STORAGE_CLASS, AMZ_TAG_COUNT, CACHE_CONTROL, CONTENT_DISPOSITION, - CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, HeaderExt as _, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, - SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, - SUFFIX_TAGGING_TIMESTAMP, get_str, insert_header_map, is_internal_key, + AMZ_OBJECT_TAGGING, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID, AMZ_STORAGE_CLASS, AMZ_TAG_COUNT, + CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, HeaderExt as _, + SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, + SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_TAGGING_TIMESTAMP, get_str, insert_header_map, is_internal_key, }; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; @@ -72,6 +72,8 @@ static VALID_SSE_REPLICATION_HEADERS: &[(&str, &str)] = &[ ("X-Rustfs-Internal-Actual-Object-Size", "X-Rustfs-Replication-Actual-Object-Size"), ]; +const ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED: &str = "managed SSE replication requires target encryption support"; + pub(crate) struct ReplicationTargetStore; impl ReplicationTargetStore { @@ -219,6 +221,20 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo) }; } + let has_sse_s3 = object_info + .user_defined + .get(AMZ_SERVER_SIDE_ENCRYPTION) + .is_some_and(|value| value.eq_ignore_ascii_case("AES256")); + let has_sse_kms = object_info + .user_defined + .get(AMZ_SERVER_SIDE_ENCRYPTION) + .is_some_and(|value| value.eq_ignore_ascii_case("aws:kms")) + || object_info.user_defined.contains_key(AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID); + + if has_sse_s3 || has_sse_kms { + return Err(Error::other(ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED)); + } + Ok((put_options, is_multipart)) } @@ -394,4 +410,37 @@ mod tests { assert!(get_header_map(&options.user_metadata, SUFFIX_REPLICATION_SSEC_CRC).is_some()); } + + #[test] + fn replication_put_options_rejects_sse_s3_until_target_encryption_is_supported() { + let object_info = ObjectInfo { + user_defined: Arc::new(HashMap::from([(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "AES256".to_string())])), + ..Default::default() + }; + + let err = match replication_put_object_options("", &object_info) { + Ok(_) => panic!("SSE-S3 replication should fail closed until target encryption headers are supported"), + Err(err) => err, + }; + + assert!(err.to_string().contains(ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED)); + } + + #[test] + fn replication_put_options_rejects_sse_kms_until_target_encryption_is_supported() { + let object_info = ObjectInfo { + user_defined: Arc::new(HashMap::from([ + (AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "aws:kms".to_string()), + (AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID.to_string(), "key-1".to_string()), + ])), + ..Default::default() + }; + + let err = match replication_put_object_options("", &object_info) { + Ok(_) => panic!("SSE-KMS replication should fail closed until target encryption headers are supported"), + Err(err) => err, + }; + + assert!(err.to_string().contains(ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED)); + } }