fix(object-lock): require source timestamps before replication passes WORM gate

Adversarial review of the replication WORM bypass found a silent unlock:
an authorized replication write that carries no source timestamp for a
category (the source never held the object, only its tags changed) is
not judged by receiver-side LWW, so the metadata replace would drop the
destination's legal hold or retention unjudged. Before the bypass that
write was merely rejected.

Gate the bypass on `replication_write_may_pass_worm_gate`: the write
passes only when it carries the source timestamp of every category that
currently locks the version, so LWW decides each one; otherwise it stays
WORM-rejected. The set layer evaluates the lock gate first so malformed
persisted lock metadata still fails closed, and the app-layer pre-check
applies the same rule.

Refs rustfs/backlog#1953

crates/ecstore/src/bucket/object_lock/objectlock_sys.rs

@@ -15,7 +15,7 @@
 use crate::bucket::metadata_sys::{ObjectLockConfigState, get_object_lock_config, get_object_lock_config_state};
 use crate::bucket::object_lock::objectlock;
 use crate::error::{Error, Result, StorageError};
-use crate::object_api::ObjectInfo;
+use crate::object_api::{ObjectInfo, ObjectOptions};
 use s3s::dto::{Date, DefaultRetention, ObjectLockConfiguration, ObjectLockLegalHoldStatus, ObjectLockRetentionMode};
 use s3s::header::{X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE};
 use std::sync::Arc;
@@ -136,12 +136,41 @@ pub fn add_years(dt: OffsetDateTime, years: i32) -> OffsetDateTime {
 
 /// Check if an object has legal hold enabled.
 /// Returns true if legal hold is ON.
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> bool {
     let lhold = objectlock::get_object_legalhold_meta(user_defined);
     matches!(lhold.status, Some(ref st) if st.as_str() == ObjectLockLegalHoldStatus::ON)
 }
 
+/// Whether an authorized replication write (`ObjectOptions::replication_request`)
+/// may overwrite a locked destination version.
+///
+/// The source's lock state governs a replica (MinIO `checkPutObjectLockAllowed`
+/// skips the existing-version check for replicas), and a source-side hold
+/// release or retention change reaches this site only through this write. The
+/// overwrite is allowed only when the write carries the source timestamp of
+/// every category that currently locks the version, so receiver-side LWW
+/// (`merge_replication_metadata_lww`) judges each of them: a category locked
+/// more recently here is kept, otherwise the source's newer state wins. A write
+/// without that timestamp carries no source decision for the category — the
+/// metadata replace would lift the lock unjudged — so it stays WORM-rejected.
+pub fn replication_write_may_pass_worm_gate(
+    user_defined: &std::collections::HashMap<String, String>,
+    opts: &ObjectOptions,
+) -> bool {
+    if !opts.replication_request {
+        return false;
+    }
+    if has_legal_hold(user_defined) && opts.replication_legalhold_timestamp.is_none() {
+        return false;
+    }
+    let ret = objectlock::get_object_retention_meta(user_defined);
+    let retention_locked = ret
+        .mode
+        .as_ref()
+        .is_some_and(|mode| is_retention_active(mode.as_str(), ret.retain_until_date.as_ref()));
+    !(retention_locked && opts.replication_retention_timestamp.is_none())
+}
+
 /// Check if an object is locked based on its metadata.
 /// This is a common function used by both lifecycle evaluation and deletion checks.
 ///
@@ -491,6 +520,59 @@ mod tests {
         }
     }
 
+    /// A replication write passes the WORM gate only when it carries the
+    /// source timestamp of every category that currently locks the version.
+    #[test]
+    fn replication_write_passes_worm_gate_only_with_every_locking_category_timestamp() {
+        use rustfs_utils::http::headers::{
+            AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
+        };
+
+        let hold = [(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, "ON")];
+        let retention = [
+            (AMZ_OBJECT_LOCK_MODE_LOWER, "GOVERNANCE"),
+            (AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, "2099-01-01T00:00:00Z"),
+        ];
+        let expired = [
+            (AMZ_OBJECT_LOCK_MODE_LOWER, "COMPLIANCE"),
+            (AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, "2000-01-01T00:00:00Z"),
+        ];
+        let metadata = |entries: &[&[(&str, &str)]]| -> std::collections::HashMap<String, String> {
+            entries
+                .iter()
+                .flat_map(|entries| entries.iter())
+                .map(|(key, value)| (key.to_string(), value.to_string()))
+                .collect()
+        };
+        let opts = |hold_ts: bool, retention_ts: bool| ObjectOptions {
+            replication_request: true,
+            replication_legalhold_timestamp: hold_ts.then_some(OffsetDateTime::UNIX_EPOCH),
+            replication_retention_timestamp: retention_ts.then_some(OffsetDateTime::UNIX_EPOCH),
+            ..Default::default()
+        };
+
+        let locked_by_both = metadata(&[&hold, &retention]);
+        assert!(replication_write_may_pass_worm_gate(&locked_by_both, &opts(true, true)));
+        assert!(!replication_write_may_pass_worm_gate(&locked_by_both, &opts(true, false)));
+        assert!(!replication_write_may_pass_worm_gate(&locked_by_both, &opts(false, true)));
+
+        assert!(replication_write_may_pass_worm_gate(&metadata(&[&hold]), &opts(true, false)));
+        assert!(!replication_write_may_pass_worm_gate(&metadata(&[&hold]), &opts(false, true)));
+        assert!(replication_write_may_pass_worm_gate(&metadata(&[&retention]), &opts(false, true)));
+        assert!(!replication_write_may_pass_worm_gate(&metadata(&[&retention]), &opts(true, false)));
+
+        // Expired retention and a released hold no longer lock anything.
+        let unlocked = metadata(&[&expired, &[(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, "OFF")]]);
+        assert!(replication_write_may_pass_worm_gate(&unlocked, &opts(false, false)));
+
+        // Never for a non-replication write, whatever it carries.
+        let local = ObjectOptions {
+            replication_request: false,
+            ..opts(true, true)
+        };
+        assert!(!replication_write_may_pass_worm_gate(&metadata(&[&hold]), &local));
+    }
+
     /// A local PutObjectRetention / PutObjectLegalHold "clear" persists the
     /// lock keys as empty strings (the MinIO on-disk shape, see
     /// `parse_object_lock_retention`); that is "no lock", not corruption, and