From 3ce01dcc737a13a138cc9f2262e8a7ef85c8ae9b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Sun, 23 Aug 2026 22:35:23 +0800 Subject: [PATCH] fix(object-lock): unblock authorized replication writes on locked versions and tolerate cleared lock metadata (#6413) --- crates/ecstore/src/api/mod.rs | 2 +- .../src/bucket/object_lock/objectlock_sys.rs | 384 +++++++++++++++--- crates/ecstore/src/set_disk/mod.rs | 1 + crates/ecstore/src/set_disk/ops/object.rs | 331 ++++++++++++++- rustfs/src/app/multipart_usecase.rs | 7 +- rustfs/src/app/object_usecase.rs | 140 ++++++- rustfs/src/app/storage_api.rs | 10 + 7 files changed, 803 insertions(+), 72 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 72f87633a..ed5f0fba2 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -161,7 +161,7 @@ pub mod bucket { pub mod objectlock_sys { pub use crate::bucket::object_lock::objectlock_sys::{ BucketObjectLockSys, ObjectLockBlockReason, add_years, check_object_lock_for_deletion, - check_retention_for_modification, is_retention_active, + check_retention_for_modification, is_retention_active, replication_write_may_pass_worm_gate, }; } } diff --git a/crates/ecstore/src/bucket/object_lock/objectlock_sys.rs b/crates/ecstore/src/bucket/object_lock/objectlock_sys.rs index c8ca80288..4eb0f79fb 100644 --- a/crates/ecstore/src/bucket/object_lock/objectlock_sys.rs +++ b/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,50 @@ 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) -> 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. +/// +/// The locking categories come from the same authoritative evaluation as the +/// commit-time WORM gate (`check_object_lock_for_deletion_with_state`): the +/// bucket default retention locks a version that carries no explicit +/// retention keys, so it is judged here too rather than read off the keys. +/// Malformed persisted lock metadata or a non-authoritative bucket +/// configuration is an error, never a pass. +pub fn replication_write_may_pass_worm_gate( + state: &ObjectLockConfigState, + obj_info: &ObjectInfo, + opts: &ObjectOptions, +) -> Result { + if !opts.replication_request { + return Ok(false); + } + if obj_info.delete_marker { + // Delete markers are never locked (same as the WORM gate). + return Ok(true); + } + let config = object_lock_config_from_state(state)?; + if legal_hold_locks(obj_info)? && opts.replication_legalhold_timestamp.is_none() { + return Ok(false); + } + let retention_locked = active_retention(config, obj_info)?.is_some(); + Ok(!(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. /// @@ -239,69 +277,101 @@ pub(crate) fn check_object_lock_for_deletion_with_config( return Ok(None); } - if let Some(status) = obj_info.user_defined.get(X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str()) { - if status.eq_ignore_ascii_case(ObjectLockLegalHoldStatus::ON) { - return Ok(Some(ObjectLockBlockReason::LegalHold)); - } - if !status.eq_ignore_ascii_case(ObjectLockLegalHoldStatus::OFF) { - return Err(Error::other("persisted object legal-hold metadata is invalid")); - } + if legal_hold_locks(obj_info)? { + return Ok(Some(ObjectLockBlockReason::LegalHold)); } - let mode = obj_info.user_defined.get(X_AMZ_OBJECT_LOCK_MODE.as_str()); - let retain_until = obj_info.user_defined.get(X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str()); - let explicit_ret = match (mode, retain_until) { - (None, None) => None, + if let Some((mode_str, retain_until)) = active_retention(config, obj_info)? + && let Some(reason) = check_retention_blocks_deletion(mode_str, Some(retain_until), bypass_governance) + { + return Ok(Some(reason)); + } + + Ok(None) +} + +/// A cleared retention / legal hold is persisted as empty strings (the MinIO +/// on-disk shape, `parse_object_lock_retention`); read it as "no lock" rather +/// than as corrupt metadata. +fn persisted_lock_value<'a>(obj_info: &'a ObjectInfo, key: &str) -> Option<&'a String> { + obj_info.user_defined.get(key).filter(|value| !value.is_empty()) +} + +/// Whether the version's persisted legal hold is ON. Any other non-empty +/// value than ON/OFF is malformed metadata and fails closed. +fn legal_hold_locks(obj_info: &ObjectInfo) -> Result { + let Some(status) = persisted_lock_value(obj_info, X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str()) else { + return Ok(false); + }; + if status.eq_ignore_ascii_case(ObjectLockLegalHoldStatus::ON) { + return Ok(true); + } + if !status.eq_ignore_ascii_case(ObjectLockLegalHoldStatus::OFF) { + return Err(Error::other("persisted object legal-hold metadata is invalid")); + } + Ok(false) +} + +/// The retention that currently locks the version, if any: the explicit +/// persisted retention when the keys are present, otherwise the bucket +/// default retention computed from the version's modification time. Returns +/// `(mode, retain_until)` only while the retention is still active. +fn active_retention<'a>( + config: Option<&'a ObjectLockConfiguration>, + obj_info: &ObjectInfo, +) -> Result> { + let mode = persisted_lock_value(obj_info, X_AMZ_OBJECT_LOCK_MODE.as_str()); + let retain_until = persisted_lock_value(obj_info, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str()); + match (mode, retain_until) { + (None, None) => {} (Some(mode), Some(retain_until)) => { let mode = objectlock::parse_ret_mode(mode).ok_or_else(|| Error::other("persisted object retention mode is invalid"))?; let retain_until = OffsetDateTime::parse(retain_until, &time::format_description::well_known::Iso8601::DEFAULT) .map(Date::from) .map_err(|_| Error::other("persisted object retention date is invalid"))?; - Some((mode, retain_until)) + let mode_str = match mode.as_str() { + ObjectLockRetentionMode::COMPLIANCE => ObjectLockRetentionMode::COMPLIANCE, + ObjectLockRetentionMode::GOVERNANCE => ObjectLockRetentionMode::GOVERNANCE, + _ => return Err(Error::other("persisted object retention mode is invalid")), + }; + return Ok(is_retention_active(mode_str, Some(&retain_until)).then(|| (mode_str, OffsetDateTime::from(retain_until)))); } _ => return Err(Error::other("persisted object retention metadata is incomplete")), + } + + let Some(default_retention) = config.and_then(|config| config.rule.as_ref()?.default_retention.as_ref()) else { + return Ok(None); }; - - if let Some((mode, retain_until)) = &explicit_ret { - let mode_str = mode.as_str(); - if is_retention_active(mode_str, Some(retain_until)) - && let Some(reason) = - check_retention_blocks_deletion(mode_str, Some(OffsetDateTime::from(retain_until.clone())), bypass_governance) - { - return Ok(Some(reason)); - } + let Some(mode) = &default_retention.mode else { + return Ok(None); + }; + let mode_str = mode.as_str(); + if mode_str != ObjectLockRetentionMode::COMPLIANCE && mode_str != ObjectLockRetentionMode::GOVERNANCE { + return Ok(None); } + // Calculate retention expiration date from object modification time + let mod_time = obj_info + .mod_time + .ok_or_else(|| Error::other("persisted object modification time is missing"))?; + let now = objectlock::utc_now_ntp(); + let retain_until = if let Some(days) = default_retention.days { + mod_time.saturating_add(time::Duration::days(i64::from(days))) + } else { + let years = default_retention + .years + .ok_or_else(|| Error::other("persisted bucket Object Lock retention period is invalid"))?; + add_years(mod_time, years) + }; + Ok((retain_until.unix_timestamp() > now.unix_timestamp()).then_some((mode_str, retain_until))) +} - if explicit_ret.is_none() - && let Some(default_retention) = config.and_then(|config| config.rule.as_ref()?.default_retention.as_ref()) - && let Some(mode) = &default_retention.mode - { - let mode_str = mode.as_str(); - if mode_str == ObjectLockRetentionMode::COMPLIANCE || mode_str == ObjectLockRetentionMode::GOVERNANCE { - // Calculate retention expiration date from object modification time - let mod_time = obj_info - .mod_time - .ok_or_else(|| Error::other("persisted object modification time is missing"))?; - let now = objectlock::utc_now_ntp(); - let retain_until = if let Some(days) = default_retention.days { - mod_time.saturating_add(time::Duration::days(i64::from(days))) - } else { - let years = default_retention - .years - .ok_or_else(|| Error::other("persisted bucket Object Lock retention period is invalid"))?; - add_years(mod_time, years) - }; - - if retain_until.unix_timestamp() > now.unix_timestamp() - && let Some(reason) = check_retention_blocks_deletion(mode_str, Some(retain_until), bypass_governance) - { - return Ok(Some(reason)); - } - } +fn object_lock_config_from_state(state: &ObjectLockConfigState) -> Result> { + match state { + ObjectLockConfigState::Configured { config, .. } => Ok(Some(config)), + ObjectLockConfigState::ConfirmedAbsent => Ok(None), + ObjectLockConfigState::Fabricated => Err(Error::other("bucket Object Lock metadata is not authoritative")), } - - Ok(None) } pub(crate) fn check_object_lock_for_deletion_with_state( @@ -309,13 +379,7 @@ pub(crate) fn check_object_lock_for_deletion_with_state( obj_info: &ObjectInfo, bypass_governance: bool, ) -> Result> { - match state { - ObjectLockConfigState::Configured { config, .. } => { - check_object_lock_for_deletion_with_config(Some(config), obj_info, bypass_governance) - } - ObjectLockConfigState::ConfirmedAbsent => check_object_lock_for_deletion_with_config(None, obj_info, bypass_governance), - ObjectLockConfigState::Fabricated => Err(Error::other("bucket Object Lock metadata is not authoritative")), - } + check_object_lock_for_deletion_with_config(object_lock_config_from_state(state)?, obj_info, bypass_governance) } /// Compatibility wrapper for callers that predate fallible metadata lookup. @@ -486,6 +550,210 @@ mod tests { } } + fn replication_opts(hold_ts: bool, retention_ts: bool) -> ObjectOptions { + 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() + } + } + + fn lock_metadata(entries: &[&[(&str, &str)]]) -> std::collections::HashMap { + entries + .iter() + .flat_map(|entries| entries.iter()) + .map(|(key, value)| (key.to_string(), value.to_string())) + .collect() + } + + fn lock_object_info(user_defined: std::collections::HashMap) -> ObjectInfo { + ObjectInfo { + user_defined: Arc::new(user_defined), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + } + } + + /// 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 absent = ObjectLockConfigState::ConfirmedAbsent; + let passes = |state: &ObjectLockConfigState, entries: &[&[(&str, &str)]], opts: &ObjectOptions| { + replication_write_may_pass_worm_gate(state, &lock_object_info(lock_metadata(entries)), opts) + .expect("well-formed lock metadata must be judged") + }; + + assert!(passes(&absent, &[&hold, &retention], &replication_opts(true, true))); + assert!(!passes(&absent, &[&hold, &retention], &replication_opts(true, false))); + assert!(!passes(&absent, &[&hold, &retention], &replication_opts(false, true))); + + assert!(passes(&absent, &[&hold], &replication_opts(true, false))); + assert!(!passes(&absent, &[&hold], &replication_opts(false, true))); + assert!(passes(&absent, &[&retention], &replication_opts(false, true))); + assert!(!passes(&absent, &[&retention], &replication_opts(true, false))); + + // Expired retention and a released hold no longer lock anything. + assert!(passes( + &absent, + &[&expired, &[(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, "OFF")]], + &replication_opts(false, false) + )); + + // Never for a non-replication write, whatever it carries. + let local = ObjectOptions { + replication_request: false, + ..replication_opts(true, true) + }; + assert!(!passes(&absent, &[&hold], &local)); + } + + /// The bucket default retention locks a version that carries no explicit + /// retention keys (`check_object_lock_for_deletion_with_config` judges it + /// from the modification time), so the replication bypass must demand the + /// retention source timestamp for it too — a tagging-only replication + /// write must not overwrite the default-protected version unjudged. + #[test] + fn replication_write_under_bucket_default_retention_requires_retention_timestamp() { + use rustfs_utils::http::headers::{AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER}; + + for mode in [ObjectLockRetentionMode::COMPLIANCE, ObjectLockRetentionMode::GOVERNANCE] { + let state = ObjectLockConfigState::Configured { + config: default_retention_config(mode), + updated_at: OffsetDateTime::now_utc(), + }; + let no_keys = lock_object_info(std::collections::HashMap::new()); + assert!( + check_object_lock_for_deletion_with_state(&state, &no_keys, false) + .expect("default retention must be judged") + .is_some(), + "{mode}: the gate must report the default retention lock" + ); + + let tagging_only = ObjectOptions { + replication_request: true, + replication_tagging_timestamp: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + assert!( + !replication_write_may_pass_worm_gate(&state, &no_keys, &tagging_only).expect("judged"), + "{mode}: a tagging-only replication write must not pass the default retention lock" + ); + assert!( + replication_write_may_pass_worm_gate(&state, &no_keys, &replication_opts(false, true)).expect("judged"), + "{mode}: the retention source timestamp lets LWW judge the default retention" + ); + + // Default retention plus a legal hold: both categories need a timestamp. + let held = lock_object_info(lock_metadata(&[&[(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, "ON")]])); + assert!(!replication_write_may_pass_worm_gate(&state, &held, &replication_opts(false, true)).expect("judged")); + assert!(!replication_write_may_pass_worm_gate(&state, &held, &replication_opts(true, false)).expect("judged")); + assert!(replication_write_may_pass_worm_gate(&state, &held, &replication_opts(true, true)).expect("judged")); + + // A version whose default retention has already expired (old + // mod_time) is not locked by the default any more. + let expired_default = ObjectInfo { + mod_time: Some(make_datetime(2000, 1, 1)), + ..lock_object_info(std::collections::HashMap::new()) + }; + assert!(replication_write_may_pass_worm_gate(&state, &expired_default, &tagging_only).expect("judged")); + + // A delete marker is never locked, so there is nothing to judge. + let delete_marker = ObjectInfo { + delete_marker: true, + ..lock_object_info(std::collections::HashMap::new()) + }; + assert!(replication_write_may_pass_worm_gate(&state, &delete_marker, &tagging_only).expect("judged")); + + // Cleared (empty) explicit keys fall back to the bucket default. + let cleared = lock_object_info(lock_metadata(&[&[(AMZ_OBJECT_LOCK_MODE_LOWER, "")]])); + assert!(!replication_write_may_pass_worm_gate(&state, &cleared, &tagging_only).expect("judged")); + } + } + + /// The replication bypass never judges from a non-authoritative bucket + /// state or malformed persisted lock metadata; both are errors, not a pass. + #[test] + fn replication_write_worm_gate_fails_closed_on_unverifiable_lock_state() { + use rustfs_utils::http::headers::AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER; + + let opts = replication_opts(true, true); + let err = replication_write_may_pass_worm_gate( + &ObjectLockConfigState::Fabricated, + &lock_object_info(std::collections::HashMap::new()), + &opts, + ) + .expect_err("fabricated bucket lock metadata must not be judged"); + assert!(err.to_string().contains("not authoritative")); + + let malformed = lock_object_info(lock_metadata(&[&[(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, "MAYBE")]])); + let err = replication_write_may_pass_worm_gate(&ObjectLockConfigState::ConfirmedAbsent, &malformed, &opts) + .expect_err("malformed legal hold must not be judged"); + assert!(err.to_string().contains("legal-hold")); + + let state = ObjectLockConfigState::Configured { + config: default_retention_config(ObjectLockRetentionMode::COMPLIANCE), + updated_at: OffsetDateTime::now_utc(), + }; + let no_mod_time = ObjectInfo::default(); + let err = replication_write_may_pass_worm_gate(&state, &no_mod_time, &opts) + .expect_err("default retention without a modification time must not be judged"); + assert!(err.to_string().contains("modification time")); + } + + /// 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 + /// must not wedge later explicit-version PUTs or deletes + /// (rustfs/backlog#1953). + #[test] + fn deletion_treats_cleared_empty_lock_metadata_as_unlocked() { + use rustfs_utils::http::headers::{ + AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, + }; + + let cases: [(&str, &[&str]); 3] = [ + ( + "cleared retention", + &[AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER], + ), + ("cleared legal hold", &[AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER]), + ( + "all cleared", + &[ + AMZ_OBJECT_LOCK_MODE_LOWER, + AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, + AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, + ], + ), + ]; + + for (case, keys) in cases { + let user_defined = keys.iter().map(|key| (key.to_string(), String::new())).collect(); + let obj_info = ObjectInfo { + user_defined: Arc::new(user_defined), + ..Default::default() + }; + + let result = check_object_lock_for_deletion_with_config(None, &obj_info, false); + assert!(matches!(result, Ok(None)), "{case}: empty lock keys must read as unlocked: {result:?}"); + } + } + #[test] fn deletion_rejects_invalid_persisted_legal_hold_metadata() { let mut user_defined = std::collections::HashMap::new(); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 0f35fa260..9ccd8287a 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/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, diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 0730277d4..3759f126c 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/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, 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 { + 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; diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index a20b13188..5549c01b8 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -524,9 +524,10 @@ impl DefaultMultipartUsecase { .await .map_err(ApiError::from)?, ); + let object_lock_config_state = load_bucket_object_lock_config_state(&bucket).await?; let previous_current_sizes = match store.get_object_info(&bucket, &key, ¤t_opts).await { Ok(existing_obj_info) => { - validate_existing_object_lock_for_write(&existing_obj_info, ¤t_opts)?; + validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, ¤t_opts)?; let physical_size = existing_obj_info.size.max(0) as u64; let logical_size = quota_object_size(&existing_obj_info); Some((physical_size, logical_size)) @@ -897,7 +898,9 @@ impl DefaultMultipartUsecase { .await .map_err(ApiError::from)?; match store.get_object_info(&bucket, &key, ¤t_opts).await { - Ok(existing_obj_info) => validate_existing_object_lock_for_write(&existing_obj_info, &opts)?, + Ok(existing_obj_info) => { + validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &opts)? + } Err(err) => { if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) { return Err(ApiError::from(err).into()); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index d1978f2db..94da99818 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -39,7 +39,7 @@ use super::storage_api::object_usecase::bucket::{ metadata_sys, object_lock::{ objectlock::{get_object_legalhold_meta, get_object_retention_meta}, - objectlock_sys::{check_object_lock_for_deletion, is_retention_active}, + objectlock_sys::{check_object_lock_for_deletion, is_retention_active, replication_write_may_pass_worm_gate}, }, predict_lifecycle_expiration, quota::{QuotaCheckResult, QuotaError, QuotaOperation}, @@ -3934,10 +3934,33 @@ fn put_like_write_creates_new_version(opts: &ObjectOptions) -> bool { opts.version_id.is_none() && opts.versioned && !opts.version_suspended } -pub(crate) fn validate_existing_object_lock_for_write(existing_obj_info: &ObjectInfo, opts: &ObjectOptions) -> S3Result<()> { +pub(crate) fn validate_existing_object_lock_for_write( + object_lock_config_state: &metadata_sys::ObjectLockConfigState, + existing_obj_info: &ObjectInfo, + opts: &ObjectOptions, +) -> S3Result<()> { if put_like_write_creates_new_version(opts) { return Ok(()); } + // An authorized replication write may replace the locked version only + // when the set layer's commit-lock LWW will judge every locking category, + // judged against the bucket's authoritative lock state (default retention + // included) exactly like the set-layer gate, which re-checks the same + // rule under the lock. A non-authoritative state or malformed lock + // metadata fails closed here. + if opts.replication_request { + let may_pass = replication_write_may_pass_worm_gate(object_lock_config_state, existing_obj_info, opts).map_err(|_| { + S3Error::with_message(S3ErrorCode::AccessDenied, "Object Lock state could not be verified.".to_string()) + })?; + return if may_pass { + Ok(()) + } else { + Err(S3Error::with_message( + S3ErrorCode::AccessDenied, + "Object is locked and the replication write carries no source lock decision for it.".to_string(), + )) + }; + } let legal_hold = get_object_legalhold_meta(&existing_obj_info.user_defined); if legal_hold @@ -6104,7 +6127,7 @@ impl DefaultObjectUsecase { }; Some(match previous_current_info { Ok(existing_obj_info) => { - validate_existing_object_lock_for_write(&existing_obj_info, &opts)?; + validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &opts)?; Some(if quota_enabled { quota_object_size(&existing_obj_info).map_err(ApiError::from)? } else { @@ -7790,7 +7813,7 @@ impl DefaultObjectUsecase { } let previous_current_sizes = match store.get_object_info(&bucket, &key, ¤t_opts).await { Ok(existing_obj_info) => { - validate_existing_object_lock_for_write(&existing_obj_info, &dst_opts)?; + validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &dst_opts)?; if let Some(expected) = expected_current_version_id.as_deref() && existing_obj_info.version_id.unwrap_or_default().to_string() != expected { @@ -10990,9 +11013,28 @@ mod tests { assert_eq!(err.message(), Some(ERR_OBJECT_LOCK_RETENTION_HEADERS_MUST_BE_PAIRED)); } + const NO_BUCKET_LOCK: metadata_sys::ObjectLockConfigState = metadata_sys::ObjectLockConfigState::ConfirmedAbsent; + + fn bucket_default_retention_state(mode: &'static str) -> metadata_sys::ObjectLockConfigState { + metadata_sys::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(ObjectLockRetentionMode::from_static(mode)), + days: Some(1), + years: None, + }), + }), + }, + updated_at: OffsetDateTime::now_utc(), + } + } + fn object_info_with_lock_metadata(metadata: HashMap) -> ObjectInfo { ObjectInfo { user_defined: Arc::new(metadata), + mod_time: Some(OffsetDateTime::now_utc()), ..Default::default() } } @@ -11031,7 +11073,7 @@ mod tests { ..Default::default() }; - validate_existing_object_lock_for_write(&compliance_retained_object_info(), &opts) + validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &compliance_retained_object_info(), &opts) .expect("versioned put should create a new version"); } @@ -11043,14 +11085,18 @@ mod tests { ..Default::default() }; - validate_existing_object_lock_for_write(&legal_hold_object_info(), &opts) + validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &legal_hold_object_info(), &opts) .expect("versioned put should create a new version"); } #[test] fn validate_existing_object_lock_blocks_unversioned_compliance_overwrite() { - let err = validate_existing_object_lock_for_write(&compliance_retained_object_info(), &ObjectOptions::default()) - .expect_err("unversioned overwrite should still be blocked"); + let err = validate_existing_object_lock_for_write( + &NO_BUCKET_LOCK, + &compliance_retained_object_info(), + &ObjectOptions::default(), + ) + .expect_err("unversioned overwrite should still be blocked"); assert_eq!(err.code(), &S3ErrorCode::AccessDenied); } @@ -11063,7 +11109,7 @@ mod tests { version_id: None, ..Default::default() }; - let err = validate_existing_object_lock_for_write(&compliance_retained_object_info(), &opts) + let err = validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &compliance_retained_object_info(), &opts) .expect_err("suspended versioning overwrite should still be blocked"); assert_eq!(err.code(), &S3ErrorCode::AccessDenied); @@ -11076,12 +11122,86 @@ mod tests { version_id: Some(Uuid::new_v4().to_string()), ..Default::default() }; - let err = validate_existing_object_lock_for_write(&compliance_retained_object_info(), &opts) + let err = validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &compliance_retained_object_info(), &opts) .expect_err("explicit version overwrite should still be blocked"); assert_eq!(err.code(), &S3ErrorCode::AccessDenied); } + /// The source's lock state governs the replica (rustfs/backlog#1953): + /// an authorized replication write carrying the locking category's source + /// timestamp may overwrite a locked version; the set layer's LWW then + /// decides per category. + #[test] + fn validate_existing_object_lock_allows_authorized_replication_overwrite() { + let opts = ObjectOptions { + versioned: true, + version_id: Some(Uuid::new_v4().to_string()), + replication_request: true, + replication_retention_timestamp: Some(OffsetDateTime::UNIX_EPOCH), + replication_legalhold_timestamp: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + + validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &compliance_retained_object_info(), &opts) + .expect("replication write must bypass the destination COMPLIANCE lock"); + validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &legal_hold_object_info(), &opts) + .expect("replication write must bypass the destination legal hold"); + } + + /// Without the locking category's source timestamp the LWW merge cannot + /// judge it, so the write stays rejected instead of lifting the lock. + #[test] + fn validate_existing_object_lock_rejects_replication_overwrite_without_lock_timestamp() { + let opts = ObjectOptions { + versioned: true, + version_id: Some(Uuid::new_v4().to_string()), + replication_request: true, + replication_tagging_timestamp: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + + let err = validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &compliance_retained_object_info(), &opts) + .expect_err("COMPLIANCE lock must hold without a retention source timestamp"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + let err = validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &legal_hold_object_info(), &opts) + .expect_err("legal hold must hold without a legal-hold source timestamp"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + /// The bucket default retention locks a version without explicit + /// retention keys; the pre-check judges the same authoritative state as + /// the set-layer gate, so a tagging-only replication write is rejected + /// and one carrying the retention source timestamp passes to LWW. + #[test] + fn validate_existing_object_lock_judges_bucket_default_retention_for_replication_overwrite() { + let default_protected = object_info_with_lock_metadata(HashMap::new()); + let tagging_only = ObjectOptions { + versioned: true, + version_id: Some(Uuid::new_v4().to_string()), + replication_request: true, + replication_tagging_timestamp: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + let with_retention_decision = ObjectOptions { + replication_retention_timestamp: Some(OffsetDateTime::UNIX_EPOCH), + ..tagging_only.clone() + }; + + for mode in [ObjectLockRetentionMode::COMPLIANCE, ObjectLockRetentionMode::GOVERNANCE] { + let state = bucket_default_retention_state(mode); + let err = validate_existing_object_lock_for_write(&state, &default_protected, &tagging_only) + .expect_err("bucket default retention must hold without a retention source timestamp"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied, "{mode}"); + validate_existing_object_lock_for_write(&state, &default_protected, &with_retention_decision) + .expect("the retention source timestamp hands the default retention to LWW"); + } + + // Without a bucket default the same version is simply unlocked. + validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &default_protected, &tagging_only) + .expect("no bucket default, no lock"); + } + #[test] fn is_put_object_extract_requested_accepts_meta_header() { let mut headers = HeaderMap::new(); diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 6bb5e9728..a08877d79 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -592,6 +592,16 @@ pub(crate) mod bucket { retain_until_date, ) } + + pub(crate) fn replication_write_may_pass_worm_gate( + state: &crate::storage::storage_api::ecstore_bucket::metadata_sys::ObjectLockConfigState, + obj_info: &crate::storage::storage_api::ObjectInfo, + opts: &crate::storage::storage_api::StorageObjectOptions, + ) -> Result { + crate::storage::storage_api::ecstore_bucket::object_lock::objectlock_sys::replication_write_may_pass_worm_gate( + state, obj_info, opts, + ) + } } }