fix(object-lock): unblock authorized replication writes on locked versions and tolerate cleared lock metadata (#6413)

crates/ecstore/src/set_disk/mod.rs

@@ -47,6 +47,7 @@ use crate::bucket::metadata_sys;
 use crate::bucket::metadata_sys::ObjectLockConfigState;
 use crate::bucket::object_lock::objectlock_sys::{
     check_object_lock_for_deletion_with_config, check_object_lock_for_deletion_with_state, check_retention_for_modification,
+    replication_write_may_pass_worm_gate,
 };
 use crate::bucket::replication::{
     ReplicateDecision, ReplicationObjectBridge, ReplicationState, ReplicationStatusType, VersionPurgeStatusType,

crates/ecstore/src/set_disk/ops/object.rs

@@ -2671,7 +2671,17 @@ impl SetDisks {
                         let object_lock_config = opts.object_lock_config_snapshot.as_deref().ok_or_else(|| {
                             Error::other("explicit-version PUT is missing its Object Lock configuration snapshot")
                         })?;
-                        if check_object_lock_for_deletion_with_state(object_lock_config.state(), &existing, false)?.is_some() {
+                        // The WORM gate protects the locked version from local
+                        // overwrites; an authorized replication write passes it
+                        // only when the LWW merge below will judge every
+                        // locking category (see
+                        // `replication_write_may_pass_worm_gate`, which judges
+                        // the same authoritative lock state as the gate,
+                        // bucket default retention included). Gate first so
+                        // malformed lock metadata still fails closed.
+                        if check_object_lock_for_deletion_with_state(object_lock_config.state(), &existing, false)?.is_some()
+                            && !replication_write_may_pass_worm_gate(object_lock_config.state(), &existing, opts)?
+                        {
                             return Err(StorageError::PrefixAccessDenied(bucket.to_string(), object.to_string()));
                         }
                         // Receiver-side LWW (rustfs/backlog#1953): reuse this
@@ -8589,6 +8599,259 @@ mod replication_lww_tests {
         );
         assert_eq!(get_str(&info.user_defined, SUFFIX_TAGGING_TIMESTAMP).as_deref(), Some(T_LOCAL));
     }
+
+    /// Destination version under an active legal hold at `hold_timestamp`,
+    /// plus an active COMPLIANCE retention (no retention timestamp).
+    async fn seed_locked_version(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, version_id: &str, hold_timestamp: &str) {
+        let mut local = HashMap::new();
+        local.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), "ON".to_string());
+        insert_str(&mut local, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, hold_timestamp.to_string());
+        local.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), "COMPLIANCE".to_string());
+        local.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), "2099-01-01T00:00:00Z".to_string());
+        put_version(set_disks, bucket, object, version_id, &versioned_opts(version_id, local)).await;
+    }
+
+    /// Inbound legal-hold release from a source that also carries the (same)
+    /// COMPLIANCE retention; the sender stamps a source timestamp for every
+    /// category the source version has.
+    fn inbound_legal_hold_release_opts(version_id: &str, timestamp: &str) -> ObjectOptions {
+        let mut inbound = HashMap::new();
+        inbound.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), "OFF".to_string());
+        insert_str(&mut inbound, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, timestamp.to_string());
+        inbound.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), "COMPLIANCE".to_string());
+        inbound.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), "2099-01-01T00:00:00Z".to_string());
+        insert_str(&mut inbound, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, T_OLD.to_string());
+        ObjectOptions {
+            replication_request: true,
+            replication_legalhold_timestamp: Some(parse_ts(timestamp)),
+            replication_retention_timestamp: Some(parse_ts(T_OLD)),
+            ..versioned_opts(version_id, inbound)
+        }
+    }
+
+    /// The source's lock state governs the replica: a legal-hold release (or a
+    /// retention change) can only reach this site through the authorized
+    /// replication write, so the commit-time WORM gate must not reject it
+    /// because the destination version is currently locked.
+    #[tokio::test]
+    async fn inbound_newer_legal_hold_release_updates_locked_version() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "lww-locked-release-newer";
+        let object = "object";
+        let version_id = Uuid::new_v4().to_string();
+        make_bucket(&disk_stores, bucket).await;
+        seed_locked_version(&set_disks, bucket, object, &version_id, T_OLD).await;
+
+        put_version(
+            &set_disks,
+            bucket,
+            object,
+            &version_id,
+            &inbound_legal_hold_release_opts(&version_id, T_NEW),
+        )
+        .await;
+
+        let info = version_info(&set_disks, bucket, object, &version_id).await;
+        assert_eq!(
+            info.user_defined.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).map(String::as_str),
+            Some("OFF"),
+            "a newer source-side legal hold release must be applied to the locked replica"
+        );
+        assert_eq!(get_str(&info.user_defined, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP).as_deref(), Some(T_NEW));
+        assert_eq!(
+            info.user_defined.get(AMZ_OBJECT_LOCK_MODE_LOWER).map(String::as_str),
+            Some("COMPLIANCE"),
+            "the untouched retention category must survive the write"
+        );
+    }
+
+    /// Skipping the WORM gate for replication writes must not weaken LWW: a
+    /// stale inbound release still loses to a hold applied more recently here.
+    #[tokio::test]
+    async fn inbound_stale_legal_hold_release_keeps_newer_local_hold() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "lww-locked-release-stale";
+        let object = "object";
+        let version_id = Uuid::new_v4().to_string();
+        make_bucket(&disk_stores, bucket).await;
+        seed_locked_version(&set_disks, bucket, object, &version_id, T_LOCAL).await;
+
+        put_version(
+            &set_disks,
+            bucket,
+            object,
+            &version_id,
+            &inbound_legal_hold_release_opts(&version_id, T_OLD),
+        )
+        .await;
+
+        let info = version_info(&set_disks, bucket, object, &version_id).await;
+        assert_eq!(
+            info.user_defined.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).map(String::as_str),
+            Some("ON"),
+            "a stale inbound release must not lift a hold applied more recently on this site"
+        );
+        assert_eq!(
+            get_str(&info.user_defined, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP).as_deref(),
+            Some(T_LOCAL)
+        );
+    }
+
+    /// A replication write that carries no source decision for a locking
+    /// category (here: tags changed at a source that never held the object)
+    /// must not lift the destination's hold by replacing the metadata
+    /// unjudged; it stays WORM-rejected like a local overwrite.
+    #[tokio::test]
+    async fn inbound_without_legal_hold_timestamp_stays_rejected_on_held_version() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "lww-locked-unjudged-category";
+        let object = "object";
+        let version_id = Uuid::new_v4().to_string();
+        make_bucket(&disk_stores, bucket).await;
+        seed_locked_version(&set_disks, bucket, object, &version_id, T_OLD).await;
+
+        let mut inbound = HashMap::new();
+        inbound.insert(AMZ_OBJECT_TAGGING.to_string(), "k=v".to_string());
+        insert_str(&mut inbound, SUFFIX_TAGGING_TIMESTAMP, T_NEW.to_string());
+        inbound.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), "COMPLIANCE".to_string());
+        inbound.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), "2099-01-01T00:00:00Z".to_string());
+        let opts = ObjectOptions {
+            replication_request: true,
+            replication_tagging_timestamp: Some(parse_ts(T_NEW)),
+            replication_retention_timestamp: Some(parse_ts(T_NEW)),
+            replication_legalhold_timestamp: None,
+            ..versioned_opts(&version_id, inbound)
+        };
+        let mut reader = PutObjReader::from_vec(b"lww-body".to_vec());
+        let err = set_disks
+            .put_object(bucket, object, &mut reader, &opts)
+            .await
+            .expect_err("a replication write without the legal-hold source timestamp must stay rejected");
+        assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)), "unexpected error: {err}");
+
+        let info = version_info(&set_disks, bucket, object, &version_id).await;
+        assert_eq!(info.user_defined.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).map(String::as_str), Some("ON"));
+    }
+
+    fn default_retention_snapshot(mode: &'static str) -> Arc<ObjectLockConfigSnapshot> {
+        Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::Configured {
+            config: s3s::dto::ObjectLockConfiguration {
+                object_lock_enabled: Some(s3s::dto::ObjectLockEnabled::from_static(s3s::dto::ObjectLockEnabled::ENABLED)),
+                rule: Some(s3s::dto::ObjectLockRule {
+                    default_retention: Some(s3s::dto::DefaultRetention {
+                        mode: Some(s3s::dto::ObjectLockRetentionMode::from_static(mode)),
+                        days: Some(1),
+                        years: None,
+                    }),
+                }),
+            },
+            updated_at: OffsetDateTime::now_utc(),
+        }))
+    }
+
+    /// The bucket default retention locks a version that carries no explicit
+    /// retention keys. A tagging-only authorized replication write carries no
+    /// source retention decision, so it must stay WORM-rejected exactly like
+    /// it does for an explicitly retained version; with the retention source
+    /// timestamp the write passes and LWW judges it.
+    #[tokio::test]
+    async fn inbound_without_retention_timestamp_stays_rejected_under_bucket_default_retention() {
+        for mode in [
+            s3s::dto::ObjectLockRetentionMode::COMPLIANCE,
+            s3s::dto::ObjectLockRetentionMode::GOVERNANCE,
+        ] {
+            let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+            let bucket = "lww-locked-default-retention";
+            let object = "object";
+            let version_id = Uuid::new_v4().to_string();
+            make_bucket(&disk_stores, bucket).await;
+            seed_local_tagged_version(&set_disks, bucket, object, &version_id).await;
+            let seeded = version_info(&set_disks, bucket, object, &version_id).await;
+            assert!(
+                !seeded.user_defined.contains_key(AMZ_OBJECT_LOCK_MODE_LOWER),
+                "the seeded version must be protected by the bucket default only"
+            );
+
+            let tagging_only = ObjectOptions {
+                object_lock_config_snapshot: Some(default_retention_snapshot(mode)),
+                ..inbound_tagging_opts(&version_id, "site=remote", T_NEW)
+            };
+            let mut reader = PutObjReader::from_vec(b"lww-body".to_vec());
+            let err = set_disks
+                .put_object(bucket, object, &mut reader, &tagging_only)
+                .await
+                .expect_err("{mode}: a tagging-only replication write must not pass the bucket default retention lock");
+            assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)), "{mode}: unexpected error: {err}");
+            let info = version_info(&set_disks, bucket, object, &version_id).await;
+            assert_eq!(
+                info.user_tags.as_str(),
+                "site=local",
+                "{mode}: the default-protected version must be untouched"
+            );
+
+            let with_retention_decision = ObjectOptions {
+                replication_retention_timestamp: Some(parse_ts(T_NEW)),
+                ..tagging_only
+            };
+            put_version(&set_disks, bucket, object, &version_id, &with_retention_decision).await;
+            let info = version_info(&set_disks, bucket, object, &version_id).await;
+            assert_eq!(
+                info.user_tags.as_str(),
+                "site=remote",
+                "{mode}: with the retention source timestamp the newer inbound tags win"
+            );
+        }
+    }
+
+    /// The gate runs before the replication bypass, so malformed persisted
+    /// lock metadata still fails closed for an authorized replication write.
+    #[tokio::test]
+    async fn replication_write_on_malformed_lock_metadata_still_fails_closed() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "lww-locked-malformed";
+        let object = "object";
+        let version_id = Uuid::new_v4().to_string();
+        make_bucket(&disk_stores, bucket).await;
+        let mut local = HashMap::new();
+        local.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), "MAYBE".to_string());
+        put_version(&set_disks, bucket, object, &version_id, &versioned_opts(&version_id, local)).await;
+
+        let mut reader = PutObjReader::from_vec(b"lww-body".to_vec());
+        let err = set_disks
+            .put_object(bucket, object, &mut reader, &inbound_legal_hold_release_opts(&version_id, T_NEW))
+            .await
+            .expect_err("malformed persisted lock metadata must fail the replication write closed");
+        assert!(!matches!(err, StorageError::PrefixAccessDenied(_, _)), "unexpected error: {err}");
+
+        let info = version_info(&set_disks, bucket, object, &version_id).await;
+        assert_eq!(info.user_defined.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).map(String::as_str), Some("MAYBE"));
+    }
+
+    /// The bypass is scoped to authorized replication writes: the same
+    /// explicit-version PUT without `replication_request` stays WORM-rejected.
+    #[tokio::test]
+    async fn non_replication_overwrite_of_locked_version_is_still_rejected() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "lww-locked-plain-put";
+        let object = "object";
+        let version_id = Uuid::new_v4().to_string();
+        make_bucket(&disk_stores, bucket).await;
+        seed_locked_version(&set_disks, bucket, object, &version_id, T_OLD).await;
+
+        let opts = ObjectOptions {
+            replication_request: false,
+            ..inbound_legal_hold_release_opts(&version_id, T_NEW)
+        };
+        let mut reader = PutObjReader::from_vec(b"lww-body".to_vec());
+        let err = set_disks
+            .put_object(bucket, object, &mut reader, &opts)
+            .await
+            .expect_err("a non-replication overwrite of a locked version must stay rejected");
+        assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)), "unexpected error: {err}");
+
+        let info = version_info(&set_disks, bucket, object, &version_id).await;
+        assert_eq!(info.user_defined.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).map(String::as_str), Some("ON"));
+    }
 }
 
 #[cfg(test)]
@@ -14578,6 +14841,72 @@ mod put_object_tmp_cleanup_tests {
         assert_eq!(body, original_body);
     }
 
+    /// A local PutObjectRetention / PutObjectLegalHold clear persists empty
+    /// lock keys (`parse_object_lock_retention`). The commit-time WORM gate
+    /// must read that as unlocked: an explicit-version PUT (the inbound
+    /// replication transport) and a version delete both have to succeed
+    /// (rustfs/backlog#1953).
+    #[tokio::test]
+    async fn explicit_version_overwrite_and_delete_succeed_after_local_lock_clear() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "put-explicit-version-cleared-lock";
+        let object = "object";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let mut initial_reader = PutObjReader::from_vec(b"original".to_vec());
+        let initial = set_disks
+            .put_object(
+                bucket,
+                object,
+                &mut initial_reader,
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("initial version should be written");
+        let version_id = initial
+            .version_id
+            .expect("versioned PUT should return a version ID")
+            .to_string();
+        let version_opts = ObjectOptions {
+            versioned: true,
+            version_id: Some(version_id.clone()),
+            delete_replication_config_snapshot: Some(Arc::new(DeleteReplicationConfigSnapshot::default())),
+            object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::ConfirmedAbsent))),
+            ..Default::default()
+        };
+        set_disks
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    eval_metadata: Some(HashMap::from([
+                        (X_AMZ_OBJECT_LOCK_MODE.as_str().to_string(), String::new()),
+                        (X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str().to_string(), String::new()),
+                        (X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str().to_string(), String::new()),
+                    ])),
+                    ..version_opts.clone()
+                },
+            )
+            .await
+            .expect("cleared lock metadata should be written");
+
+        let mut replacement = PutObjReader::from_vec(b"replacement".to_vec());
+        set_disks
+            .put_object(bucket, object, &mut replacement, &version_opts)
+            .await
+            .expect("explicit-version PUT must not be wedged by cleared lock metadata");
+
+        set_disks
+            .delete_object(bucket, object, version_opts)
+            .await
+            .expect("version delete must not be wedged by cleared lock metadata");
+    }
+
     #[tokio::test]
     async fn version_only_copy_checks_the_destination_version_object_lock() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;