From dcf3e4b9e8c618459911cca0a2bdae5503fd6106 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Sun, 16 Aug 2026 05:56:04 +0800 Subject: [PATCH] fix(replication): transport and persist LWW timestamps for tag, retention, and legal hold (#6129) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(replication): pin missing LWW timestamp header transport Red-light tests for the replication timestamp three-header contract: - put_object_headers_carry_replication_timestamp_headers pins that PutObjectOptions::header() must emit the x-{rustfs,minio}-source-replication-{tagging,retention,legalhold}-timestamp headers when the internal timestamps are set (currently missing). - test_put_opts_from_headers_gates_replication_timestamp_persistence_on_authorization and test_complete_multipart_opts_persist_replication_timestamps_when_authorized pin that an authorized replication PUT / multipart complete must persist the inbound timestamps into the internal metadata keys while unauthorized requests must not (currently never persisted). - fake_s3_target journals the three timestamp headers per request (ReplicationTimestampHeaders on RequestRecord) so sender-side e2e assertions can observe what a real target receives; self-test included. * fix(replication): transport and persist LWW timestamps for tag, retention, and legal hold Active-active conflict resolution for concurrent tag/retention/legal-hold edits needs the source's per-category modification times on both sides of the wire; the three AdvancedPutOptions timestamp fields were dead and the headers were neither sent nor parsed. - Emit x-{rustfs,minio}-source-replication-{tagging,retention,legalhold}- timestamp from PutObjectOptions::header(); names and RFC3339 values interoperate with MinIO (minio-go constants.go, object-api-options.go), pinned by a header_compat wire-name test. - Default the three AdvancedPutOptions timestamps to UNIX_EPOCH and skip epoch values in header(), so "never modified" is not sent as a modification made now. - Parse the headers only on authorized replication PUTs and multipart completes, expose them as Option on ObjectOptions, and persist them into the dual-prefix internal metadata keys so the outbound pass (replication_target_boundary) reads the source's timestamps instead of the mod_time fallback. - Record the local tagging timestamp in the PutObjectTagging and DeleteObjectTagging eval metadata, mirroring the object-lock handlers; without it the sender only ever had the mod_time fallback to offer. Receiver-side LWW comparison (keep newer stored category metadata over a stale inbound copy) is left as a TODO at the parse site. * fix(replication): load the stored tagging timestamp independently of remaining tags Review: DeleteObjectTagging persists the tagging-timestamp internal key but leaves the object tagless, and the outbound mapper only loaded the key inside the user_tags-nonempty branch — the deletion's LWW timestamp stayed at the epoch and the header was omitted, so the deletion could never win conflict resolution on the replica. The stored key is now loaded unconditionally; the mod_time fallback still applies only while tags exist (MinIO parity), and a tagless object without the key keeps the epoch default (no header). Deletion-path regression test added. * fix(storage): reserve replication transport names at metadata ingest Second review round: a client PUT of x-amz-meta-x-rustfs-source-replication-tagging-timestamp materialized the bare transport key as stored user metadata. The outbound replication header builder forwards user metadata verbatim on a server-authorized request, so the receiver would persist the attacker-chosen value as trusted internal LWW state — and for a tagless object nothing later overwrites it. The ingest namespacing guard now reserves the whole x-rustfs-source- / x-minio-source- families (the new timestamps and their siblings: source-mtime/-etag/-version-id/-replication-request), folding forged keys back under x-amz-meta-. Forged-ingress regression covers both prefixes and a sibling. * fix(replication): harden timestamp replay * fix(app): route retention helper through facade --------- Co-authored-by: overtrue --- crates/e2e_test/src/fake_s3_target/mod.rs | 92 +++++++- .../ecstore/src/bucket/bucket_target_sys.rs | 74 +++++- .../replication_target_boundary.rs | 166 ++++++++++++- crates/ecstore/src/object_api/types.rs | 6 + crates/utils/src/http/header_compat.rs | 29 +++ rustfs/src/app/multipart_usecase.rs | 8 +- rustfs/src/app/object_usecase.rs | 21 +- rustfs/src/app/storage_api.rs | 5 +- rustfs/src/storage/ecfs.rs | 12 +- rustfs/src/storage/options.rs | 223 +++++++++++++++++- rustfs/src/storage/storage_api.rs | 5 +- 11 files changed, 606 insertions(+), 35 deletions(-) diff --git a/crates/e2e_test/src/fake_s3_target/mod.rs b/crates/e2e_test/src/fake_s3_target/mod.rs index d094f8cdb..c8ddcecf3 100644 --- a/crates/e2e_test/src/fake_s3_target/mod.rs +++ b/crates/e2e_test/src/fake_s3_target/mod.rs @@ -76,6 +76,18 @@ const SOURCE_MTIME_HEADERS: [&str; 2] = ["x-rustfs-source-mtime", "x-minio-sourc const SOURCE_REPLICATION_REQUEST_HEADERS: [&str; 2] = ["x-rustfs-source-replication-request", "x-minio-source-replication-request"]; const SOURCE_ETAG_HEADERS: [&str; 2] = ["x-rustfs-source-etag", "x-minio-source-etag"]; +const SOURCE_TAGGING_TIMESTAMP_HEADERS: [&str; 2] = [ + "x-rustfs-source-replication-tagging-timestamp", + "x-minio-source-replication-tagging-timestamp", +]; +const SOURCE_RETENTION_TIMESTAMP_HEADERS: [&str; 2] = [ + "x-rustfs-source-replication-retention-timestamp", + "x-minio-source-replication-retention-timestamp", +]; +const SOURCE_LEGALHOLD_TIMESTAMP_HEADERS: [&str; 2] = [ + "x-rustfs-source-replication-legalhold-timestamp", + "x-minio-source-replication-legalhold-timestamp", +]; 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"]; @@ -118,6 +130,25 @@ pub enum FaultAction { WrongEtag, } +/// Replication LWW timestamp headers observed on a request, journaled so +/// sender-side tests can assert what a real target would receive. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ReplicationTimestampHeaders { + pub tagging: Option, + pub retention: Option, + pub legalhold: Option, +} + +impl ReplicationTimestampHeaders { + fn from_headers(headers: &HeaderMap) -> Self { + Self { + tagging: header_value(headers, &SOURCE_TAGGING_TIMESTAMP_HEADERS).map(bounded_journal_value), + retention: header_value(headers, &SOURCE_RETENTION_TIMESTAMP_HEADERS).map(bounded_journal_value), + legalhold: header_value(headers, &SOURCE_LEGALHOLD_TIMESTAMP_HEADERS).map(bounded_journal_value), + } + } +} + /// Credential-free request metadata retained for deterministic assertions. #[derive(Debug, Clone, PartialEq, Eq)] pub struct RequestRecord { @@ -131,6 +162,7 @@ pub struct RequestRecord { pub part_number: Option, pub content_length: Option, pub consumed_bytes: Option, + pub replication_timestamps: ReplicationTimestampHeaders, pub fault: Option, } @@ -536,7 +568,15 @@ impl S3Access for FaultAccess { .get(CONTENT_LENGTH) .and_then(|value| value.to_str().ok()) .and_then(|value| value.parse().ok()); - let fault = record_request(&self.control, operation, context.method().clone(), parsed, content_length); + let replication_timestamps = ReplicationTimestampHeaders::from_headers(context.headers()); + let fault = record_request( + &self.control, + operation, + context.method().clone(), + parsed, + content_length, + replication_timestamps, + ); if let Some(RequestFault { action: FaultAction::Status(status), .. @@ -589,6 +629,7 @@ fn record_request( method: Method, parsed: ParsedRequest, content_length: Option, + replication_timestamps: ReplicationTimestampHeaders, ) -> Option { let mut state = lock(control); let action = parsed @@ -613,6 +654,7 @@ fn record_request( part_number: parsed.part_number, content_length, consumed_bytes: None, + replication_timestamps, fault: action.clone(), }); action.map(|action| RequestFault { sequence, action }) @@ -1699,6 +1741,52 @@ mod tests { .await?) } + #[tokio::test] + async fn journals_replication_timestamp_headers() -> Result<(), BoxError> { + let target = FakeS3Target::start().await?; + target.create_bucket("target-bucket"); + let client = client(&target); + + client + .put_object() + .bucket("target-bucket") + .key("plain") + .body(ByteStream::from_static(b"plain")) + .send() + .await?; + client + .put_object() + .bucket("target-bucket") + .key("stamped") + .body(ByteStream::from_static(b"stamped")) + .customize() + .map_request(move |mut request| { + let headers = request.headers_mut(); + headers.insert("x-rustfs-source-replication-tagging-timestamp", "2026-01-02T03:04:05Z"); + headers.insert("x-minio-source-replication-retention-timestamp", "2026-01-02T03:04:06Z"); + headers.insert("x-rustfs-source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z"); + Ok::<_, std::convert::Infallible>(request) + }) + .send() + .await?; + + let requests = target.requests(); + let plain = requests + .iter() + .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("plain")) + .expect("plain PUT must be journaled"); + assert_eq!(plain.replication_timestamps, ReplicationTimestampHeaders::default()); + + let stamped = requests + .iter() + .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("stamped")) + .expect("stamped PUT must be journaled"); + assert_eq!(stamped.replication_timestamps.tagging.as_deref(), Some("2026-01-02T03:04:05Z")); + assert_eq!(stamped.replication_timestamps.retention.as_deref(), Some("2026-01-02T03:04:06Z")); + assert_eq!(stamped.replication_timestamps.legalhold.as_deref(), Some("2026-01-02T03:04:07Z")); + Ok(()) + } + macro_rules! assert_sdk_error { ($error:expr, $status:expr, $code:expr) => {{ let error = &$error; @@ -2985,6 +3073,7 @@ mod tests { part_number: None, }, Some(0), + ReplicationTimestampHeaders::default(), ); } let records = lock(&control).requests.clone(); @@ -3006,6 +3095,7 @@ mod tests { part_number: None, }, None, + ReplicationTimestampHeaders::default(), ); { let bounded_records = lock(&bounded_control); diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 1e1e18dc0..c6bcc5742 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -58,7 +58,9 @@ use rustfs_utils::http::{ }; use rustfs_utils::http::{ SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK, - SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header, + SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST, + SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID, + insert_header, }; use rustls_pki_types::pem::PemObject; use serde::{Deserialize, Serialize}; @@ -1476,9 +1478,12 @@ impl Default for AdvancedPutOptions { replication_status: ReplicationStatusType::Pending, source_mtime: OffsetDateTime::now_utc(), replication_request: false, - retention_timestamp: OffsetDateTime::now_utc(), - tagging_timestamp: OffsetDateTime::now_utc(), - legalhold_timestamp: OffsetDateTime::now_utc(), + // UNIX_EPOCH means "never modified": header() must not emit a + // timestamp header for it, otherwise a receiver would treat an + // unset category as a modification made right now. + retention_timestamp: OffsetDateTime::UNIX_EPOCH, + tagging_timestamp: OffsetDateTime::UNIX_EPOCH, + legalhold_timestamp: OffsetDateTime::UNIX_EPOCH, replication_validity_check: false, } } @@ -1675,6 +1680,16 @@ impl PutObjectOptions { ); } + for (suffix, timestamp) in [ + (SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, self.internal.tagging_timestamp), + (SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, self.internal.retention_timestamp), + (SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, self.internal.legalhold_timestamp), + ] { + if timestamp.unix_timestamp() != 0 { + insert_header(&mut header, suffix, timestamp.format(&Rfc3339).unwrap_or_default()); + } + } + if self.internal.replication_request { insert_header(&mut header, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); } @@ -2842,6 +2857,57 @@ mod tests { ); } + #[test] + fn put_object_headers_carry_replication_timestamp_headers() { + // MinIO receivers resolve concurrent tag/retention/legal-hold edits by + // last-writer-wins on these headers (object-api-options.go parses them + // as RFC3339); a replica without them loses every conflict resolution. + let mut opts = PutObjectOptions::default(); + opts.internal.replication_request = true; + let tagging = OffsetDateTime::from_unix_timestamp(1_700_000_001).expect("valid timestamp"); + let retention = OffsetDateTime::from_unix_timestamp(1_700_000_002).expect("valid timestamp"); + let legalhold = OffsetDateTime::from_unix_timestamp(1_700_000_003).expect("valid timestamp"); + opts.internal.tagging_timestamp = tagging; + opts.internal.retention_timestamp = retention; + opts.internal.legalhold_timestamp = legalhold; + + let header = opts.header(); + for (suffix, expected) in [ + ("source-replication-tagging-timestamp", tagging), + ("source-replication-retention-timestamp", retention), + ("source-replication-legalhold-timestamp", legalhold), + ] { + assert_eq!( + rustfs_utils::http::get_header(&header, suffix).as_deref(), + Some(expected.format(&Rfc3339).expect("RFC3339 timestamp").as_str()), + "replication put requests must carry the {suffix} header" + ); + } + } + + #[test] + fn put_object_headers_omit_unset_replication_timestamps() { + // UNIX_EPOCH means "never modified on the source"; sending it would + // make the receiver treat an unset category as a fresh modification. + let mut opts = PutObjectOptions::default(); + opts.internal.replication_request = true; + opts.internal.tagging_timestamp = OffsetDateTime::UNIX_EPOCH; + opts.internal.retention_timestamp = OffsetDateTime::UNIX_EPOCH; + opts.internal.legalhold_timestamp = OffsetDateTime::UNIX_EPOCH; + + let header = opts.header(); + for suffix in [ + "source-replication-tagging-timestamp", + "source-replication-retention-timestamp", + "source-replication-legalhold-timestamp", + ] { + assert!( + rustfs_utils::http::get_header(&header, suffix).is_none(), + "unset {suffix} must not be sent to replication targets" + ); + } + } + #[tokio::test] async fn get_remote_target_client_internal_rejects_loopback_endpoint() { let sys = BucketTargetSys::default(); diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index 156c94cd7..e82ab7e52 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -27,8 +27,10 @@ use rustfs_utils::http::{ AMZ_OBJECT_TAGGING, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_SERVER_SIDE_ENCRYPTION_KMS_CONTEXT, 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, is_object_encryption_marker, is_replication_stripped_encryption_key, ssec_replication_transport_header, + SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, + SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP, + get_str, insert_header_map, is_internal_key, is_object_encryption_marker, is_replication_stripped_encryption_key, + ssec_replication_transport_header, }; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; @@ -119,6 +121,27 @@ fn classify_replication_source_encryption(metadata: &HashMap) -> } } +fn is_legacy_source_replication_timestamp_key(key: &str) -> bool { + fn has_prefix_and_suffix(key: &str, prefix: &str, suffix: &str) -> bool { + let key = key.as_bytes(); + key.len() == prefix.len() + suffix.len() + && key[..prefix.len()].eq_ignore_ascii_case(prefix.as_bytes()) + && key[prefix.len()..].eq_ignore_ascii_case(suffix.as_bytes()) + } + + [ + SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, + SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, + SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, + ] + .iter() + .any(|suffix| { + ["x-rustfs-", "x-minio-"] + .iter() + .any(|prefix| has_prefix_and_suffix(key, prefix, suffix)) + }) +} + pub(crate) fn replication_object_is_ssec_encrypted(user_defined: &HashMap) -> bool { rustfs_replication::is_ssec_encrypted(user_defined) } @@ -176,6 +199,11 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo) continue; } + if is_legacy_source_replication_timestamp_key(key) { + meta.insert(format!("x-amz-meta-{key}"), value.to_string()); + continue; + } + if is_internal_key(key) || is_standard_header(key) { continue; } @@ -259,15 +287,23 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo) if !tags.is_empty() { put_options.user_tags = tags; - put_options.internal.tagging_timestamp = - if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP) { - OffsetDateTime::parse(×tamp, &Rfc3339) - .map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))? - } else { - object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) - }; } } + // Load the stored tagging timestamp independently of whether any tags + // remain: DeleteObjectTagging leaves the object tagless but stamps this + // key, and the deletion's LWW timestamp must still reach the replica. + // With no stored key, fall back to mod_time only while tags exist + // (MinIO parity); a tagless object without the key was never tagged and + // keeps the epoch default (no header). + put_options.internal.tagging_timestamp = if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP) + { + OffsetDateTime::parse(×tamp, &Rfc3339) + .map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))? + } else if !put_options.user_tags.is_empty() { + object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) + } else { + OffsetDateTime::UNIX_EPOCH + }; let metadata = &*object_info.user_defined; @@ -283,13 +319,15 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo) put_options.cache_control = cache_control.to_string(); } - if let Some(mode) = metadata.lookup(AMZ_OBJECT_LOCK_MODE) { + if let Some(mode) = metadata.lookup(AMZ_OBJECT_LOCK_MODE).filter(|mode| !mode.is_empty()) { put_options.mode = Some(ObjectLockRetentionMode::from(mode.to_uppercase().as_str())); } if let Some(retain_until_date) = metadata.lookup(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE) { - put_options.retain_until_date = OffsetDateTime::parse(retain_until_date, &Rfc3339) - .map_err(|err| Error::other(format!("Failed to parse retain until date: {err}")))?; + if !retain_until_date.is_empty() { + put_options.retain_until_date = OffsetDateTime::parse(retain_until_date, &Rfc3339) + .map_err(|err| Error::other(format!("Failed to parse retain until date: {err}")))?; + } put_options.internal.retention_timestamp = if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP) { OffsetDateTime::parse(×tamp, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH) @@ -694,6 +732,110 @@ mod tests { assert!(options.internal.replication_request); } + /// DeleteObjectTagging leaves the object tagless but stamps the + /// tagging-timestamp internal key; the deletion's LWW timestamp must + /// still be loaded (and therefore sent) so the replica can order the + /// deletion against concurrent tag edits. + #[test] + fn replication_put_options_carry_tagging_timestamp_after_tag_deletion() { + let mut metadata = std::collections::HashMap::new(); + rustfs_utils::http::insert_str(&mut metadata, SUFFIX_TAGGING_TIMESTAMP, "2026-01-02T03:04:05Z".to_string()); + + let object_info = ObjectInfo { + user_defined: Arc::new(metadata), + user_tags: Arc::new(String::new()), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + version_id: Some(Uuid::nil()), + ..Default::default() + }; + + let (options, _) = replication_put_object_options("", &object_info).expect("build put options"); + + assert!(options.user_tags.is_empty()); + assert_eq!( + options.internal.tagging_timestamp, + OffsetDateTime::parse("2026-01-02T03:04:05Z", &Rfc3339).expect("valid timestamp"), + "the stored tagging timestamp must load independently of remaining tags" + ); + + // A tagless object without the stored key was never tagged: the epoch + // default keeps the header unsent. + let untagged = ObjectInfo { + user_tags: Arc::new(String::new()), + mod_time: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")), + version_id: Some(Uuid::nil()), + ..Default::default() + }; + let (options, _) = replication_put_object_options("", &untagged).expect("build put options"); + assert_eq!(options.internal.tagging_timestamp, OffsetDateTime::UNIX_EPOCH); + } + + #[test] + fn replication_put_options_do_not_promote_legacy_user_timestamp_metadata() { + let legacy_keys = [ + "x-rustfs-source-replication-tagging-timestamp", + "x-rustfs-source-replication-retention-timestamp", + "x-rustfs-source-replication-legalhold-timestamp", + "x-minio-source-replication-tagging-timestamp", + "x-minio-source-replication-retention-timestamp", + "x-minio-source-replication-legalhold-timestamp", + ]; + let object_info = ObjectInfo { + user_defined: Arc::new( + legacy_keys + .iter() + .map(|key| (key.to_string(), "2099-01-02T03:04:05Z".to_string())) + .collect(), + ), + ..Default::default() + }; + + let (options, _) = replication_put_object_options("", &object_info).expect("build put options"); + + for legacy_key in legacy_keys { + assert!(!options.user_metadata.contains_key(legacy_key)); + assert_eq!( + options + .user_metadata + .get(&format!("x-amz-meta-{legacy_key}")) + .map(String::as_str), + Some("2099-01-02T03:04:05Z") + ); + } + assert_eq!(options.internal.tagging_timestamp, OffsetDateTime::UNIX_EPOCH); + assert_eq!(options.internal.retention_timestamp, OffsetDateTime::UNIX_EPOCH); + assert_eq!(options.internal.legalhold_timestamp, OffsetDateTime::UNIX_EPOCH); + } + + #[test] + fn replication_put_options_carry_retention_timestamp_after_clear() { + let mut metadata = HashMap::from([ + (AMZ_OBJECT_LOCK_MODE.to_string(), String::new()), + (AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.to_string(), String::new()), + ]); + rustfs_utils::http::insert_str(&mut metadata, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, "2026-01-02T03:04:05Z".to_string()); + let object_info = ObjectInfo { + user_defined: Arc::new(metadata), + ..Default::default() + }; + + let (options, _) = replication_put_object_options("", &object_info).expect("retention clear must replicate"); + + assert!(options.mode.is_none()); + assert_eq!(options.retain_until_date, OffsetDateTime::UNIX_EPOCH); + assert_eq!( + options.internal.retention_timestamp, + OffsetDateTime::parse("2026-01-02T03:04:05Z", &Rfc3339).expect("valid timestamp") + ); + let headers = options.header(); + assert!(!headers.contains_key(AMZ_OBJECT_LOCK_MODE)); + assert!(!headers.contains_key(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE)); + assert_eq!( + rustfs_utils::http::get_header(&headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP).as_deref(), + Some("2026-01-02T03:04:05Z") + ); + } + #[test] fn replication_put_options_strip_encryption_metadata_from_plaintext_objects() { use rustfs_utils::http::object_encryption_keys::{INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER, SSEC_ORIGINAL_SIZE_HEADER}; diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 291b9039a..624cb7593 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -277,6 +277,12 @@ pub struct ObjectOptions { /// fence avoids recursively acquiring the read lock behind a queued writer. pub bucket_lifecycle_lock_fence: Option, pub replication_request: bool, + /// Source-cluster LWW timestamps carried by an authorized replication + /// request; None when the source never modified the category. Only the + /// replication-authorized options builders may set these. + pub replication_tagging_timestamp: Option, + pub replication_retention_timestamp: Option, + pub replication_legalhold_timestamp: Option, /// Authorized SSE-C replication passthrough: the body is already /// ciphertext, so the write path must not encrypt or compress it and /// stores the restored encryption metadata verbatim. Only the diff --git a/crates/utils/src/http/header_compat.rs b/crates/utils/src/http/header_compat.rs index 38486ca07..c36632d8e 100644 --- a/crates/utils/src/http/header_compat.rs +++ b/crates/utils/src/http/header_compat.rs @@ -50,6 +50,14 @@ pub const SUFFIX_SOURCE_DELETEMARKER: &str = "source-deletemarker"; pub const SUFFIX_SOURCE_PROXY_REQUEST: &str = "source-proxy-request"; pub const SUFFIX_SOURCE_REPLICATION_REQUEST: &str = "source-replication-request"; pub const SUFFIX_SOURCE_REPLICATION_CHECK: &str = "source-replication-check"; +// LWW timestamps for replicated tag/retention/legal-hold modifications. MinIO +// declares these with mixed case (internal/http/headers.go: +// X-Minio-Source-Replication-Tagging-Timestamp / -Retention-Timestamp / +// -LegalHold-Timestamp); HTTP header names compare case-insensitively, so the +// lowercase suffix forms interoperate. Values are RFC3339 on the wire. +pub const SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP: &str = "source-replication-tagging-timestamp"; +pub const SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP: &str = "source-replication-retention-timestamp"; +pub const SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP: &str = "source-replication-legalhold-timestamp"; pub const SUFFIX_REPLICATION_SSEC_CRC: &str = "replication-ssec-crc"; /// Returns true if the key is object-encryption metadata understood by RustFS or MinIO. @@ -196,6 +204,27 @@ mod tests { assert_eq!(get_object_encryption_original_size(&metadata).expect("valid size"), Some(42)); } + #[test] + fn replication_timestamp_headers_match_minio_wire_names() { + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, "2026-01-02T03:04:05Z"); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, "2026-01-02T03:04:06Z"); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, "2026-01-02T03:04:07Z"); + + // The exact names MinIO's object-api-options.go reads (its Get() + // canonicalizes case, so a case-insensitive match is wire-equivalent). + for name in [ + "X-Minio-Source-Replication-Tagging-Timestamp", + "X-Minio-Source-Replication-Retention-Timestamp", + "X-Minio-Source-Replication-LegalHold-Timestamp", + "x-rustfs-source-replication-tagging-timestamp", + "x-rustfs-source-replication-retention-timestamp", + "x-rustfs-source-replication-legalhold-timestamp", + ] { + assert!(headers.contains_key(name), "replication timestamp header {name} must be written"); + } + } + #[test] fn test_get_header() { let mut headers = HeaderMap::new(); diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 1b871a0df..a20b13188 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -44,8 +44,8 @@ use super::storage_api::multipart_usecase::io::{HashReader, WriteEncryption, Wri use super::storage_api::multipart_usecase::object_utils::to_s3s_etag; use super::storage_api::multipart_usecase::options::{ copy_src_opts, extract_metadata_from_mime, get_complete_multipart_upload_opts_with_replication_authorization, - get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, parse_copy_source_range, - put_opts_with_replication_authorization, validate_archive_content_encoding, + get_content_sha256_with_query, get_opts, has_replication_retention_update, namespace_reserved_user_metadata, + parse_copy_source_range, put_opts_with_replication_authorization, validate_archive_content_encoding, }; use super::storage_api::multipart_usecase::request_context::spawn_traced_join; use super::storage_api::multipart_usecase::s3_api::multipart::{ @@ -777,7 +777,9 @@ impl DefaultMultipartUsecase { let mut metadata = create_multipart_upload_metadata(input_metadata, &req.headers, tagging, storage_class.as_ref()); - let has_explicit_object_lock_retention = object_lock_mode.is_some() || object_lock_retain_until_date.is_some(); + let has_explicit_object_lock_retention = object_lock_mode.is_some() + || object_lock_retain_until_date.is_some() + || has_replication_retention_update(&req.headers, replication_authorized); let object_lock_config_state = load_bucket_object_lock_config_state(&bucket).await?; if let Some(object_lock_metadata) = build_put_like_object_lock_metadata( &bucket, diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 3bf35bc1b..913d6eb52 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -86,8 +86,8 @@ use super::storage_api::object_usecase::object_utils::to_s3s_etag; use super::storage_api::object_usecase::options::{ copy_dst_opts_with_replication_authorization, copy_src_opts, del_opts_with_versioning, extract_metadata, extract_metadata_from_mime_with_object_name, filter_object_metadata, get_content_sha256_with_query, get_opts, - namespace_reserved_user_metadata, normalize_content_encoding_for_storage, preserve_unclassified_user_metadata, - put_opts_with_replication_authorization, validate_archive_content_encoding, + has_replication_retention_update, namespace_reserved_user_metadata, normalize_content_encoding_for_storage, + preserve_unclassified_user_metadata, put_opts_with_replication_authorization, validate_archive_content_encoding, }; use super::storage_api::object_usecase::request_context::{self, spawn_traced, spawn_traced_join}; use super::storage_api::object_usecase::s3_api::multipart::parse_list_parts_params; @@ -5799,7 +5799,9 @@ impl DefaultObjectUsecase { )?; let mut metadata = metadata.unwrap_or_default(); - let has_explicit_object_lock_retention = object_lock_mode.is_some() || object_lock_retain_until_date.is_some(); + let has_explicit_object_lock_retention = object_lock_mode.is_some() + || object_lock_retain_until_date.is_some() + || has_replication_retention_update(&req.headers, inbound_replication_put); let object_lock_config_stage_start = put_stage_metrics_enabled.then(Instant::now); let object_lock_config_state = load_bucket_object_lock_config_state(&bucket).await?; rustfs_io_metrics::record_put_object_stage_duration_from("app_object_lock_config_lookup", object_lock_config_stage_start); @@ -10081,6 +10083,19 @@ mod tests { assert_eq!(metadata.get(AMZ_OBJECT_LOCK_MODE_LOWER).map(String::as_str), Some("COMPLIANCE")); assert!(metadata.contains_key(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER)); assert_eq!(metadata.get("x-amz-meta-x-amz-object-lock-mode").map(String::as_str), Some("GOVERNANCE")); + + let mut replication_headers = HeaderMap::new(); + insert_header(&mut replication_headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); + insert_header( + &mut replication_headers, + rustfs_utils::http::SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, + "2026-01-01T00:00:00Z", + ); + let mut replica_metadata = HashMap::new(); + let explicit_clear = has_replication_retention_update(&replication_headers, true); + apply_bucket_default_lock_retention("bucket", &state, &mut replica_metadata, explicit_clear).unwrap(); + assert!(!replica_metadata.contains_key(AMZ_OBJECT_LOCK_MODE_LOWER)); + assert!(!replica_metadata.contains_key(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER)); } fn pax_record(key: &str, value: &[u8]) -> Vec { diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index b97b16bcb..839b8a38f 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -996,8 +996,9 @@ pub(crate) mod options { copy_dst_opts_with_replication_authorization, copy_src_opts, del_opts_with_versioning, extract_metadata, extract_metadata_from_mime, extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts_with_replication_authorization, get_content_sha256_with_query, get_opts, - namespace_reserved_user_metadata, normalize_content_encoding_for_storage, parse_copy_source_range, - preserve_unclassified_user_metadata, put_opts_with_replication_authorization, validate_archive_content_encoding, + has_replication_retention_update, namespace_reserved_user_metadata, normalize_content_encoding_for_storage, + parse_copy_source_range, preserve_unclassified_user_metadata, put_opts_with_replication_authorization, + validate_archive_content_encoding, }; } diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index ce4be49d5..9369d7224 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -45,7 +45,7 @@ use rustfs_targets::EventName; use rustfs_utils::http::headers::{ AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, }; -use rustfs_utils::http::{SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, insert_str}; +use rustfs_utils::http::{SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP, insert_str}; use s3s::{S3, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, dto::*, s3_error}; use std::collections::HashMap; use std::fmt::Debug; @@ -461,6 +461,11 @@ impl S3 for FS { let mut eval_metadata = HashMap::new(); insert_str(&mut eval_metadata, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string()); insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default()); + insert_str( + &mut eval_metadata, + SUFFIX_TAGGING_TIMESTAMP, + OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_default(), + ); opts.eval_metadata = Some(eval_metadata); } @@ -1645,6 +1650,11 @@ impl S3 for FS { let mut eval_metadata = HashMap::new(); insert_str(&mut eval_metadata, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string()); insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default()); + insert_str( + &mut eval_metadata, + SUFFIX_TAGGING_TIMESTAMP, + OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_default(), + ); opts.eval_metadata = Some(eval_metadata); } diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index 609b2ea62..ac920b5c3 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -17,11 +17,13 @@ use crate::storage::storage_api::options_consumer::contract::{object::HTTPPrecon use http::header::{IF_MATCH, IF_NONE_MATCH}; use http::{HeaderMap, HeaderValue}; use rustfs_utils::http::{ - AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, - SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_REQUEST, - SUFFIX_SOURCE_VERSION_ID, get_header, + AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, + SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, + SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, + SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, + SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID, SUFFIX_TAGGING_TIMESTAMP, get_header, header_compat::{MINIO_ENCRYPTION_PREFIX, RUSTFS_ENCRYPTION_PREFIX}, - insert_header_map, + insert_header_map, insert_str, metadata_compat::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX}, }; use rustfs_utils::http::{ @@ -422,6 +424,9 @@ pub fn get_complete_multipart_upload_opts_with_replication_authorization( preserve_etag, ..Default::default() }; + if replication_request { + apply_replication_timestamps_from_headers(headers, &mut opts); + } apply_replica_status_from_headers(headers, &mut opts, replication_request_authorized); fill_conditional_writes_opts_from_header(headers, &mut opts)?; @@ -458,6 +463,12 @@ pub fn put_opts_from_headers(headers: &HeaderMap, metadata: HashMap put_opts_from_headers_with_replication_authorization(headers, metadata, false) } +pub(crate) fn has_replication_retention_update(headers: &HeaderMap, replication_request_authorized: bool) -> bool { + replication_request_authorized + && get_header(headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true") + && replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP).is_some() +} + pub fn put_opts_from_headers_with_replication_authorization( headers: &HeaderMap, metadata: HashMap, @@ -479,6 +490,7 @@ pub fn put_opts_from_headers_with_replication_authorization( if let Some(crc) = get_header(headers, SUFFIX_REPLICATION_SSEC_CRC) { insert_header_map(&mut opts.user_defined, SUFFIX_REPLICATION_SSEC_CRC, crc.into_owned()); } + apply_replication_timestamps_from_headers(headers, &mut opts); } Ok(opts) } @@ -504,6 +516,47 @@ fn replication_source_mtime(headers: &HeaderMap) -> Option, suffix: &str) -> Option { + let value = get_header(headers, suffix)?; + let value = value.trim(); + match time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339) { + Ok(timestamp) => Some(timestamp), + Err(err) => { + tracing::warn!("Invalid {} value '{}' (replication request=true): {}", suffix, value, err); + None + } + } +} + +/// Callers must gate on an authorized replication request: these headers are +/// trusted source-cluster state, not client input. +fn apply_replication_timestamps_from_headers(headers: &HeaderMap, opts: &mut ObjectOptions) { + opts.replication_tagging_timestamp = replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP); + opts.replication_retention_timestamp = replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP); + opts.replication_legalhold_timestamp = replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP); + + // Persist into the internal metadata keys so a later outbound replication + // pass (replication_target_boundary) reads the source's modification + // times instead of falling back to mod_time. + // TODO(P1-6): receiver-side LWW is still missing — when the stored + // per-category timestamp is newer than the inbound one, the existing + // tags/retention/legal-hold should win instead of being overwritten. + for (timestamp, suffix) in [ + (opts.replication_tagging_timestamp, SUFFIX_TAGGING_TIMESTAMP), + (opts.replication_retention_timestamp, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP), + (opts.replication_legalhold_timestamp, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP), + ] { + if let Some(timestamp) = timestamp + && let Ok(value) = timestamp.format(&time::format_description::well_known::Rfc3339) + { + insert_str(&mut opts.user_defined, suffix, value); + } + } +} + fn apply_replica_status_from_headers(headers: &HeaderMap, opts: &mut ObjectOptions, authorized: bool) { if !authorized { return; @@ -663,6 +716,13 @@ fn is_reserved_user_metadata_key(key: &str) -> bool { || starts_with_ignore_ascii_case(key, MINIO_INTERNAL_PREFIX) || starts_with_ignore_ascii_case(key, RUSTFS_ENCRYPTION_PREFIX) || starts_with_ignore_ascii_case(key, MINIO_ENCRYPTION_PREFIX) + // Replication transport names (source-replication timestamps, + // source-mtime/-etag/-version-id, ...). A bare stored key with one of + // these names is forwarded verbatim by the outbound replication + // header builder on a server-authorized request, so the receiver + // would persist attacker-chosen values as trusted internal LWW state. + || starts_with_ignore_ascii_case(key, "x-rustfs-source-") + || starts_with_ignore_ascii_case(key, "x-minio-source-") } fn stored_user_metadata_key(key: &str) -> String { @@ -1082,15 +1142,16 @@ mod tests { del_opts_with_versioning, detect_content_type_from_object_name, extract_metadata, extract_metadata_from_mime, extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts, get_complete_multipart_upload_opts_with_replication_authorization, get_default_opts, get_opts, - namespace_reserved_user_metadata, parse_copy_source_range, put_opts, put_opts_from_headers, - put_opts_from_headers_with_replication_authorization, put_opts_with_replication_authorization, + has_replication_retention_update, namespace_reserved_user_metadata, parse_copy_source_range, put_opts, + put_opts_from_headers, put_opts_from_headers_with_replication_authorization, put_opts_with_replication_authorization, validate_archive_content_encoding, }; use http::{HeaderMap, HeaderValue}; use rustfs_utils::http::{ AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, - SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header, + SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, + SUFFIX_SOURCE_VERSION_ID, insert_header, }; use s3s::S3ErrorCode; use s3s::dto::{BucketVersioningStatus, ExcludedPrefix, VersioningConfiguration}; @@ -1520,6 +1581,24 @@ mod tests { assert!(opts.preserve_etag.is_none()); } + #[test] + fn test_replication_retention_update_requires_authorization() { + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, "2026-01-01T00:00:00Z"); + + assert!(!has_replication_retention_update(&headers, false)); + assert!(has_replication_retention_update(&headers, true)); + + let mut missing_request = HeaderMap::new(); + insert_header( + &mut missing_request, + SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, + "2026-01-01T00:00:00Z", + ); + assert!(!has_replication_retention_update(&missing_request, true)); + } + #[test] fn test_put_opts_from_headers_gates_ssec_passthrough_on_authorization() { use rustfs_utils::http::object_encryption_keys::{ @@ -1599,6 +1678,136 @@ mod tests { assert!(opts_invalid.mod_time.is_none()); } + /// A client PUT must not materialize the replication transport names as + /// bare stored user-metadata keys: the outbound replication header + /// builder forwards user metadata verbatim on a server-authorized + /// request, so a bare `x-rustfs-source-replication-*-timestamp` key would + /// deliver an attacker-chosen value into the replica's trusted internal + /// LWW state (for a tagless object nothing later overwrites it). + #[test] + fn test_replication_transport_names_cannot_be_forged_via_user_metadata() { + let mut headers = HeaderMap::new(); + for name in [ + "x-amz-meta-x-rustfs-source-replication-tagging-timestamp", + "x-amz-meta-x-minio-source-replication-legalhold-timestamp", + "x-rustfs-meta-x-rustfs-source-replication-retention-timestamp", + "x-amz-meta-x-rustfs-source-mtime", + ] { + headers.insert( + http::header::HeaderName::from_static(name), + HeaderValue::from_static("2026-01-02T03:04:05Z"), + ); + } + + let metadata = extract_metadata(&headers); + + for forged in [ + "x-rustfs-source-replication-tagging-timestamp", + "x-minio-source-replication-legalhold-timestamp", + "x-rustfs-source-replication-retention-timestamp", + "x-rustfs-source-mtime", + ] { + assert!( + !metadata.contains_key(forged), + "{forged} must not be storable as a bare user-metadata key" + ); + } + // The values survive, namespaced back under the user-metadata prefix. + assert_eq!( + metadata + .get("x-amz-meta-x-rustfs-source-replication-tagging-timestamp") + .map(String::as_str), + Some("2026-01-02T03:04:05Z") + ); + } + + #[test] + fn test_put_opts_from_headers_gates_replication_timestamp_persistence_on_authorization() { + // Sender-side LWW state (replication_target_boundary.rs) is read back + // from these internal metadata keys, so an authorized replication PUT + // must persist the inbound timestamp headers; an unauthorized client + // must not be able to forge them. + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); + insert_header(&mut headers, "source-replication-tagging-timestamp", "2026-01-02T03:04:05Z"); + insert_header(&mut headers, "source-replication-retention-timestamp", "2026-01-02T03:04:06Z"); + insert_header(&mut headers, "source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z"); + + let untrusted = put_opts_from_headers(&headers, HashMap::new()).expect("ordinary PUT options should be created"); + for suffix in [ + "tagging-timestamp", + "objectlock-retention-timestamp", + "objectlock-legalhold-timestamp", + ] { + assert!( + rustfs_utils::http::get_str(&untrusted.user_defined, suffix).is_none(), + "unauthorized clients must not persist the {suffix} internal key" + ); + } + assert!(untrusted.replication_tagging_timestamp.is_none()); + assert!(untrusted.replication_retention_timestamp.is_none()); + assert!(untrusted.replication_legalhold_timestamp.is_none()); + + let trusted = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true) + .expect("authorized replication request should parse"); + let parse = |value: &str| { + time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339).expect("valid RFC3339") + }; + assert_eq!(trusted.replication_tagging_timestamp, Some(parse("2026-01-02T03:04:05Z"))); + assert_eq!(trusted.replication_retention_timestamp, Some(parse("2026-01-02T03:04:06Z"))); + assert_eq!(trusted.replication_legalhold_timestamp, Some(parse("2026-01-02T03:04:07Z"))); + for (suffix, expected) in [ + ("tagging-timestamp", "2026-01-02T03:04:05Z"), + ("objectlock-retention-timestamp", "2026-01-02T03:04:06Z"), + ("objectlock-legalhold-timestamp", "2026-01-02T03:04:07Z"), + ] { + assert_eq!( + rustfs_utils::http::get_str(&trusted.user_defined, suffix).as_deref(), + Some(expected), + "authorized replication must persist the {suffix} internal key" + ); + } + } + + #[test] + fn test_complete_multipart_opts_persist_replication_timestamps_when_authorized() { + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); + insert_header(&mut headers, "replication-actual-object-size", "1"); + insert_header(&mut headers, "source-replication-tagging-timestamp", "2026-01-02T03:04:05Z"); + insert_header(&mut headers, "source-replication-retention-timestamp", "2026-01-02T03:04:06Z"); + insert_header(&mut headers, "source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z"); + + let untrusted = get_complete_multipart_upload_opts(&headers).expect("ordinary multipart options should be created"); + for suffix in [ + "tagging-timestamp", + "objectlock-retention-timestamp", + "objectlock-legalhold-timestamp", + ] { + assert!( + rustfs_utils::http::get_str(&untrusted.user_defined, suffix).is_none(), + "unauthorized multipart completes must not persist the {suffix} internal key" + ); + } + + let trusted = get_complete_multipart_upload_opts_with_replication_authorization(&headers, true) + .expect("authorized multipart complete should parse"); + for (suffix, expected) in [ + ("tagging-timestamp", "2026-01-02T03:04:05Z"), + ("objectlock-retention-timestamp", "2026-01-02T03:04:06Z"), + ("objectlock-legalhold-timestamp", "2026-01-02T03:04:07Z"), + ] { + assert_eq!( + rustfs_utils::http::get_str(&trusted.user_defined, suffix).as_deref(), + Some(expected), + "authorized multipart completes must persist the {suffix} internal key" + ); + } + assert!(trusted.replication_tagging_timestamp.is_some()); + assert!(trusted.replication_retention_timestamp.is_some()); + assert!(trusted.replication_legalhold_timestamp.is_some()); + } + #[test] fn test_put_opts_from_headers_with_replica_status() { let mut headers = HeaderMap::new(); diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index f4f084a31..5d217f18e 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -185,8 +185,9 @@ pub(crate) mod options_consumer { copy_dst_opts_with_replication_authorization, copy_src_opts, del_opts_with_versioning, extract_metadata, extract_metadata_from_mime, extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts_with_replication_authorization, get_content_sha256_with_query, get_opts, - namespace_reserved_user_metadata, normalize_content_encoding_for_storage, parse_copy_source_range, - preserve_unclassified_user_metadata, put_opts_with_replication_authorization, validate_archive_content_encoding, + has_replication_retention_update, namespace_reserved_user_metadata, normalize_content_encoding_for_storage, + parse_copy_source_range, preserve_unclassified_user_metadata, put_opts_with_replication_authorization, + validate_archive_content_encoding, }; pub(crate) mod contract {