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 <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>

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<ECStore>, bucket: &str, object: &str) -
     buf
 }
 
+async fn live_object_version_count(ecstore: &Arc<ECStore>, 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"]

rustfs/src/app/multipart_usecase.rs

@@ -348,7 +348,7 @@ impl DefaultMultipartUsecase {
         );
         let previous_current_size = match store.get_object_info(&bucket, &key, &current_opts).await {
             Ok(existing_obj_info) => {
-                validate_existing_object_lock_for_write(&existing_obj_info)?;
+                validate_existing_object_lock_for_write(&existing_obj_info, &current_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, &current_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());

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, &current_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, &current_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<String, String>) -> 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();