From 3d9caff3a40754888d668484d3e42fd8691e8ac3 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Wed, 3 Jun 2026 10:50:53 +0800 Subject: [PATCH] fix(object-lock): allow locked objects to receive new versions (#3179) * fix(object-lock): allow locked objects to receive new versions * fix(object-lock): validate copy destination writes --------- Co-authored-by: Henry Guo Co-authored-by: houseme --- .../src/app/lifecycle_transition_api_test.rs | 106 +++++++++++++++++- rustfs/src/app/multipart_usecase.rs | 4 +- rustfs/src/app/object_usecase.rs | 95 +++++++++++++++- 3 files changed, 197 insertions(+), 8 deletions(-) diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index 830804f2b..c52294ffa 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -30,8 +30,8 @@ use rustfs_ecstore::{ global::GLOBAL_TierConfigMgr, store::ECStore, store_api::{ - BucketOperations, BucketOptions, MakeBucketOptions, MultipartOperations, ObjectIO, ObjectOperations, ObjectOptions, - PutObjReader, + BucketOperations, BucketOptions, ListOperations, MakeBucketOptions, MultipartOperations, ObjectIO, ObjectOperations, + ObjectOptions, PutObjReader, }, tier::{ tier_config::{TierConfig, TierType}, @@ -498,6 +498,20 @@ async fn read_object_bytes(ecstore: &Arc, bucket: &str, object: &str) - buf } +async fn live_object_version_count(ecstore: &Arc, bucket: &str, object: &str) -> usize { + let versions = ecstore + .clone() + .list_object_versions(bucket, object, None, None, None, 1000) + .await + .expect("Failed to list object versions"); + + versions + .objects + .iter() + .filter(|info| info.name == object && !info.delete_marker) + .count() +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[serial] #[ignore = "requires isolated global object layer state"] @@ -600,6 +614,94 @@ async fn copy_object_if_none_match_existing_destination_returns_precondition_fai ); } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[serial] +#[ignore = "requires isolated global object layer state"] +async fn copy_object_allows_new_version_for_locked_destination_but_blocks_explicit_overwrite() { + let (_disk_paths, ecstore) = setup_test_env().await; + let fs = FS::new(); + let usecase = DefaultObjectUsecase::without_context(); + + let bucket = format!("test-copy-object-lock-{}", &Uuid::new_v4().simple().to_string()[..8]); + let src_object = "test/source.txt"; + let dst_object = "test/destination.txt"; + let locked_payload = b"locked destination payload"; + let source_payload = b"copy source payload"; + + create_test_bucket(&ecstore, bucket.as_str()).await; + set_bucket_object_lock_enabled(bucket.as_str()) + .await + .expect("Failed to enable object lock for bucket"); + let _ = upload_test_object(&ecstore, bucket.as_str(), src_object, source_payload).await; + + let retain_until = Timestamp::from(time::OffsetDateTime::now_utc().saturating_add(time::Duration::days(1))); + let locked_input = PutObjectInput::builder() + .bucket(bucket.clone()) + .key(dst_object.to_string()) + .body(Some(streaming_blob_from_bytes(locked_payload))) + .content_length(Some(locked_payload.len() as i64)) + .object_lock_mode(Some(ObjectLockMode::from_static(ObjectLockMode::COMPLIANCE))) + .object_lock_retain_until_date(Some(retain_until)) + .build() + .unwrap(); + + Box::pin(usecase.execute_put_object(&fs, build_request(locked_input, Method::PUT))) + .await + .expect("Failed to upload locked destination object"); + + let locked_info = ecstore + .get_object_info(bucket.as_str(), dst_object, &ObjectOptions::default()) + .await + .expect("Failed to fetch locked destination object info"); + let locked_version_id = locked_info + .version_id + .expect("locked destination should have a version ID") + .to_string(); + + let copy_input = CopyObjectInput::builder() + .copy_source(CopySource::Bucket { + bucket: bucket.clone().into(), + key: src_object.to_string().into(), + version_id: None, + }) + .bucket(bucket.clone()) + .key(dst_object.to_string()) + .build() + .unwrap(); + + let copy_output = Box::pin(usecase.execute_copy_object(build_request(copy_input, Method::PUT))) + .await + .expect("CopyObject should create a new version over a locked current version"); + let copied_version_id = copy_output + .output + .version_id + .expect("versioned CopyObject should return the created version ID"); + + assert_ne!(copied_version_id, locked_version_id); + assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), dst_object).await, source_payload); + assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), dst_object).await, 2); + + let explicit_overwrite_input = CopyObjectInput::builder() + .copy_source(CopySource::Bucket { + bucket: bucket.clone().into(), + key: src_object.to_string().into(), + version_id: None, + }) + .bucket(bucket.clone()) + .key(dst_object.to_string()) + .version_id(Some(locked_version_id)) + .build() + .unwrap(); + + let err = Box::pin(usecase.execute_copy_object(build_request(explicit_overwrite_input, Method::PUT))) + .await + .expect_err("explicit CopyObject overwrite of a locked version should be blocked"); + + assert_eq!(err.code(), &s3s::S3ErrorCode::AccessDenied); + assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), dst_object).await, source_payload); + assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), dst_object).await, 2); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial] #[ignore = "requires isolated global object layer state"] diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index d11eebaac..68c41a6b8 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -348,7 +348,7 @@ impl DefaultMultipartUsecase { ); let previous_current_size = match store.get_object_info(&bucket, &key, ¤t_opts).await { Ok(existing_obj_info) => { - validate_existing_object_lock_for_write(&existing_obj_info)?; + validate_existing_object_lock_for_write(&existing_obj_info, ¤t_opts)?; Some(existing_obj_info.size.max(0) as u64) } Err(err) => { @@ -614,7 +614,7 @@ 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)?, + Ok(existing_obj_info) => validate_existing_object_lock_for_write(&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 64d6859fb..c1f1dcc4e 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -1036,7 +1036,15 @@ pub(crate) async fn build_put_like_object_lock_metadata( Ok(Some(eval_metadata)) } -pub(crate) fn validate_existing_object_lock_for_write(existing_obj_info: &ObjectInfo) -> S3Result<()> { +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<()> { + if put_like_write_creates_new_version(opts) { + return Ok(()); + } + let legal_hold = get_object_legalhold_meta(&existing_obj_info.user_defined); if legal_hold .status @@ -1866,7 +1874,7 @@ impl DefaultObjectUsecase { ); let previous_current_size = match store.get_object_info(&bucket, &key, ¤t_opts).await { Ok(existing_obj_info) => { - validate_existing_object_lock_for_write(&existing_obj_info)?; + validate_existing_object_lock_for_write(&existing_obj_info, &opts)?; Some(existing_obj_info.size.max(0) as u64) } Err(err) => { @@ -2720,7 +2728,7 @@ impl DefaultObjectUsecase { ..Default::default() }; - let mut dst_opts = copy_dst_opts(&bucket, &key, version_id, &req.headers, HashMap::new()) + let mut dst_opts = copy_dst_opts(&bucket, &key, dest_version_id.clone(), &req.headers, HashMap::new()) .await .map_err(ApiError::from)?; @@ -2750,7 +2758,7 @@ impl DefaultObjectUsecase { } let previous_current_size = match store.get_object_info(&bucket, &key, ¤t_opts).await { Ok(existing_obj_info) => { - validate_existing_object_lock_for_write(&existing_obj_info)?; + validate_existing_object_lock_for_write(&existing_obj_info, &dst_opts)?; Some(existing_obj_info.size.max(0) as u64) } Err(err) => { @@ -4558,6 +4566,85 @@ mod tests { assert_eq!(err.message(), Some(ERR_OBJECT_LOCK_RETENTION_HEADERS_MUST_BE_PAIRED)); } + fn object_info_with_lock_metadata(metadata: HashMap) -> ObjectInfo { + ObjectInfo { + user_defined: Arc::new(metadata), + ..Default::default() + } + } + + fn compliance_retained_object_info() -> ObjectInfo { + let mut metadata = HashMap::new(); + metadata.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), ObjectLockRetentionMode::COMPLIANCE.to_string()); + metadata.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), "2030-01-01T00:00:00Z".to_string()); + object_info_with_lock_metadata(metadata) + } + + fn legal_hold_object_info() -> ObjectInfo { + let mut metadata = HashMap::new(); + metadata.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), ObjectLockLegalHoldStatus::ON.to_string()); + object_info_with_lock_metadata(metadata) + } + + #[test] + fn validate_existing_object_lock_allows_versioned_new_version_with_compliance_retention() { + let opts = ObjectOptions { + versioned: true, + version_id: None, + ..Default::default() + }; + + validate_existing_object_lock_for_write(&compliance_retained_object_info(), &opts) + .expect("versioned put should create a new version"); + } + + #[test] + fn validate_existing_object_lock_allows_versioned_new_version_with_legal_hold() { + let opts = ObjectOptions { + versioned: true, + version_id: None, + ..Default::default() + }; + + validate_existing_object_lock_for_write(&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"); + + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[test] + fn validate_existing_object_lock_blocks_suspended_version_compliance_overwrite() { + let opts = ObjectOptions { + versioned: true, + version_suspended: true, + version_id: None, + ..Default::default() + }; + let err = validate_existing_object_lock_for_write(&compliance_retained_object_info(), &opts) + .expect_err("suspended versioning overwrite should still be blocked"); + + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[test] + fn validate_existing_object_lock_blocks_explicit_version_compliance_overwrite() { + let opts = ObjectOptions { + versioned: true, + version_id: Some(Uuid::new_v4().to_string()), + ..Default::default() + }; + let err = validate_existing_object_lock_for_write(&compliance_retained_object_info(), &opts) + .expect_err("explicit version overwrite should still be blocked"); + + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + #[test] fn is_put_object_extract_requested_accepts_meta_header() { let mut headers = HeaderMap::new();