fix(s3): allow CopyObject to restore a historical object version (#4250)

crates/e2e_test/src/copy_object_version_restore_test.rs

@@ -0,0 +1,163 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Regression test for Issue #4238: CopyObject must allow restoring a historical
+//! object version onto the current key (same bucket/key + sourceVersionId) using the
+//! default COPY metadata directive, preserving data, Content-Type and user metadata.
+
+#[cfg(test)]
+mod tests {
+    use crate::common::{RustFSTestEnvironment, init_logging};
+    use aws_sdk_s3::primitives::ByteStream;
+    use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
+    use serial_test::serial;
+    use tracing::info;
+
+    #[tokio::test]
+    #[serial]
+    async fn test_self_copy_of_historical_version_restores_data_and_metadata() {
+        init_logging();
+        info!("Issue #4238: self-copy of a historical version must be allowed and preserve metadata");
+
+        let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
+        env.start_rustfs_server(vec![]).await.expect("Failed to start RustFS");
+
+        let client = env.create_s3_client();
+        let bucket = "copy-object-version-restore-test";
+        let key = "docs/report.txt";
+
+        client
+            .create_bucket()
+            .bucket(bucket)
+            .send()
+            .await
+            .expect("Failed to create bucket");
+
+        client
+            .put_bucket_versioning()
+            .bucket(bucket)
+            .versioning_configuration(
+                VersioningConfiguration::builder()
+                    .status(BucketVersioningStatus::Enabled)
+                    .build(),
+            )
+            .send()
+            .await
+            .expect("Failed to enable versioning");
+
+        // Version 1: the content we want to restore later.
+        let v1_content = b"original report contents -- version one";
+        let put_v1 = client
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .content_type("text/plain; charset=utf-8")
+            .metadata("origin", "v1")
+            .body(ByteStream::from_static(v1_content))
+            .send()
+            .await
+            .expect("PUT v1 failed");
+        let v1_id = put_v1.version_id().expect("v1 must have a version id").to_string();
+
+        // Version 2: becomes the current version.
+        let v2_content = b"updated report contents -- version two (current)";
+        let put_v2 = client
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .content_type("application/octet-stream")
+            .metadata("origin", "v2")
+            .body(ByteStream::from_static(v2_content))
+            .send()
+            .await
+            .expect("PUT v2 failed");
+        let v2_id = put_v2.version_id().expect("v2 must have a version id").to_string();
+
+        assert_ne!(v1_id, v2_id, "the two puts must produce distinct versions");
+
+        // Restore v1 by copying it onto the current key with the DEFAULT (COPY) directive.
+        // This is the exact operation issue #4238 says is wrongly rejected as a self-copy.
+        let copy_out = client
+            .copy_object()
+            .bucket(bucket)
+            .key(key)
+            .copy_source(format!("{bucket}/{key}?versionId={v1_id}"))
+            .send()
+            .await
+            .expect("CopyObject restoring a historical version must succeed (issue #4238)");
+
+        let restored_id = copy_out
+            .copy_object_result()
+            .and_then(|_| copy_out.version_id())
+            .expect("restore copy must create a new version id")
+            .to_string();
+        assert_ne!(restored_id, v1_id, "restore must create a NEW version, not mutate v1");
+        assert_ne!(restored_id, v2_id, "restore must create a NEW version, not mutate v2");
+
+        // The current object must now serve v1's data and metadata (COPY directive preserves both).
+        let head = client
+            .head_object()
+            .bucket(bucket)
+            .key(key)
+            .send()
+            .await
+            .expect("HEAD failed after restore");
+        assert_eq!(head.content_length(), Some(v1_content.len() as i64), "current size must equal v1");
+        assert_eq!(
+            head.content_type(),
+            Some("text/plain; charset=utf-8"),
+            "Content-Type must be preserved from v1 (issue #4238 reports it is lost)"
+        );
+        assert_eq!(
+            head.metadata().and_then(|m| m.get("origin")),
+            Some(&"v1".to_string()),
+            "user metadata must be preserved from v1"
+        );
+
+        let get = client
+            .get_object()
+            .bucket(bucket)
+            .key(key)
+            .send()
+            .await
+            .expect("GET failed after restore");
+        let body = get.body.collect().await.expect("collect body").into_bytes();
+        assert_eq!(body.as_ref(), v1_content, "current object data must equal v1 after restore");
+
+        // The original versions must remain intact and independently readable.
+        let get_v1 = client
+            .get_object()
+            .bucket(bucket)
+            .key(key)
+            .version_id(&v1_id)
+            .send()
+            .await
+            .expect("GET v1 failed");
+        let v1_body = get_v1.body.collect().await.expect("collect v1 body").into_bytes();
+        assert_eq!(v1_body.as_ref(), v1_content, "v1 must remain intact after restore");
+
+        let get_v2 = client
+            .get_object()
+            .bucket(bucket)
+            .key(key)
+            .version_id(&v2_id)
+            .send()
+            .await
+            .expect("GET v2 failed");
+        let v2_body = get_v2.body.collect().await.expect("collect v2 body").into_bytes();
+        assert_eq!(v2_body.as_ref(), v2_content, "v2 must remain intact after restore");
+
+        env.stop_server();
+    }
+}

crates/e2e_test/src/kms/copy_object_version_restore_sse_test.rs

@@ -0,0 +1,183 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Regression test for Issue #4238 (encryption case): restoring a historical version of an
+//! SSE-S3 encrypted object onto the current key must produce a readable object.
+//!
+//! The version-restore path writes a NEW version. A metadata-only (zero-copy) version copy
+//! would make the new version point at the source ciphertext while carrying freshly derived
+//! encryption metadata, yielding an undecryptable object. The store layer therefore performs a
+//! full read/write copy through `put_object` when the handler supplies a reader, keeping the
+//! stored bytes consistent with the new version's key metadata.
+
+use super::common::{LocalKMSTestEnvironment, create_key_with_specific_id};
+use crate::common::init_logging;
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::types::{BucketVersioningStatus, ServerSideEncryption, VersioningConfiguration};
+use serial_test::serial;
+use tracing::info;
+
+#[tokio::test]
+#[serial]
+async fn test_self_copy_of_historical_sse_s3_version_is_readable() {
+    init_logging();
+    info!("Issue #4238 (SSE): restoring an encrypted historical version must stay decryptable");
+
+    let mut kms_env = LocalKMSTestEnvironment::new().await.expect("failed to create local KMS env");
+    // Start the server with the local KMS backend. The dev-defaults flag is required for the
+    // in-process local KMS master key used by e2e tests.
+    let default_key_id = "rustfs-e2e-test-default-key";
+    let keys_dir = kms_env.kms_keys_dir.clone();
+    create_key_with_specific_id(&keys_dir, default_key_id)
+        .await
+        .expect("failed to create local KMS key");
+    kms_env
+        .base_env
+        .start_rustfs_server_with_env(
+            vec![
+                "--kms-enable",
+                "--kms-backend",
+                "local",
+                "--kms-key-dir",
+                &keys_dir,
+                "--kms-default-key-id",
+                default_key_id,
+            ],
+            &[("RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS", "true")],
+        )
+        .await
+        .expect("failed to start RustFS with local KMS");
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+
+    let client = kms_env.base_env.create_s3_client();
+    let bucket = "copy-object-version-restore-sse-test";
+    let key = "secrets/report.txt";
+
+    client
+        .create_bucket()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to create bucket");
+    client
+        .put_bucket_versioning()
+        .bucket(bucket)
+        .versioning_configuration(
+            VersioningConfiguration::builder()
+                .status(BucketVersioningStatus::Enabled)
+                .build(),
+        )
+        .send()
+        .await
+        .expect("failed to enable versioning");
+
+    // Version 1: SSE-S3 encrypted content we want to restore later.
+    let v1_content = b"encrypted original report -- version one";
+    let put_v1 = client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .content_type("text/plain; charset=utf-8")
+        .metadata("origin", "v1")
+        .server_side_encryption(ServerSideEncryption::Aes256)
+        .body(ByteStream::from_static(v1_content))
+        .send()
+        .await
+        .expect("PUT v1 failed");
+    assert_eq!(put_v1.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
+    let v1_id = put_v1.version_id().expect("v1 must have a version id").to_string();
+
+    // Version 2: becomes the current version.
+    let v2_content = b"encrypted updated report -- version two (current)";
+    let put_v2 = client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .content_type("application/octet-stream")
+        .metadata("origin", "v2")
+        .server_side_encryption(ServerSideEncryption::Aes256)
+        .body(ByteStream::from_static(v2_content))
+        .send()
+        .await
+        .expect("PUT v2 failed");
+    let v2_id = put_v2.version_id().expect("v2 must have a version id").to_string();
+    assert_ne!(v1_id, v2_id, "the two puts must produce distinct versions");
+
+    // Restore v1 by copying it onto the current key with the DEFAULT (COPY) directive, keeping
+    // the destination encrypted. Under the old zero-copy version-restore path this new version
+    // would be undecryptable.
+    let copy_out = client
+        .copy_object()
+        .bucket(bucket)
+        .key(key)
+        .copy_source(format!("{bucket}/{key}?versionId={v1_id}"))
+        .server_side_encryption(ServerSideEncryption::Aes256)
+        .send()
+        .await
+        .expect("CopyObject restoring an encrypted historical version must succeed (issue #4238)");
+    let restored_id = copy_out
+        .version_id()
+        .expect("restore copy must create a new version id")
+        .to_string();
+    assert_ne!(restored_id, v1_id, "restore must create a NEW version, not mutate v1");
+    assert_ne!(restored_id, v2_id, "restore must create a NEW version, not mutate v2");
+
+    // The current object must now decrypt to v1's plaintext (proves stored ciphertext matches
+    // the new version's key metadata) and preserve Content-Type and user metadata.
+    let get = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await
+        .expect("GET after restore failed");
+    assert_eq!(get.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
+    assert_eq!(
+        get.content_type(),
+        Some("text/plain; charset=utf-8"),
+        "Content-Type must be preserved from v1"
+    );
+    assert_eq!(
+        get.metadata().and_then(|m| m.get("origin")),
+        Some(&"v1".to_string()),
+        "user metadata must be preserved from v1"
+    );
+    let body = get.body.collect().await.expect("collect body").into_bytes();
+    assert_eq!(body.as_ref(), v1_content, "restored current object must decrypt to v1 content");
+
+    // Original versions must remain intact and independently decryptable.
+    let get_v1 = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .version_id(&v1_id)
+        .send()
+        .await
+        .expect("GET v1 failed");
+    let v1_body = get_v1.body.collect().await.expect("collect v1 body").into_bytes();
+    assert_eq!(v1_body.as_ref(), v1_content, "v1 must remain intact after restore");
+
+    let get_v2 = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .version_id(&v2_id)
+        .send()
+        .await
+        .expect("GET v2 failed");
+    let v2_body = get_v2.body.collect().await.expect("collect v2 body").into_bytes();
+    assert_eq!(v2_body.as_ref(), v2_content, "v2 must remain intact after restore");
+
+    kms_env.base_env.stop_server();
+}

crates/e2e_test/src/kms/mod.rs

@@ -47,3 +47,6 @@ mod bucket_default_encryption_test;
 
 #[cfg(test)]
 mod encryption_metadata_test;
+
+#[cfg(test)]
+mod copy_object_version_restore_sse_test;

crates/e2e_test/src/lib.rs

@@ -141,6 +141,9 @@ mod heal_erasure_disk_rebuild_test;
 #[cfg(test)]
 mod copy_object_metadata_test;
 
+#[cfg(test)]
+mod copy_object_version_restore_test;
+
 // S3 dummy-compat bucket API tests
 #[cfg(test)]
 mod bucket_logging_test;

crates/ecstore/src/store/object.rs

@@ -778,6 +778,25 @@ impl ECStore {
             }
 
             if dst_opts.versioned && src_opts.version_id != dst_opts.version_id {
+                // Restoring a specific historical version onto the current key creates a NEW
+                // version. When the caller supplies a reader (S3 CopyObject), write the fetched
+                // bytes through put_object so any re-encryption/compression applied to the reader
+                // stays consistent with the new version's metadata. Sharing the source data_dir via
+                // a metadata-only version copy would corrupt SSE/compressed objects (issue #4238).
+                if let Some(reader) = src_info.put_object_reader.as_mut() {
+                    let put_opts = ObjectOptions {
+                        user_defined: (*src_info.user_defined).clone(),
+                        versioned: dst_opts.versioned,
+                        version_id: dst_opts.version_id.clone(),
+                        no_lock: dst_opts.no_lock,
+                        mod_time: dst_opts.mod_time,
+                        http_preconditions: dst_opts.http_preconditions.clone(),
+                        ..Default::default()
+                    };
+                    return self.pools[pool_idx]
+                        .put_object(dst_bucket, &dst_object, reader, &put_opts)
+                        .await;
+                }
                 src_info.version_only = true;
                 return self.pools[pool_idx]
                     .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, &dst_opts)

rustfs/src/app/object_usecase.rs

@@ -4234,6 +4234,22 @@ impl DefaultObjectUsecase {
             } => (bucket.to_string(), key.to_string(), version_id.map(|v| v.to_string())),
         };
 
+        // Normalize the copy-source version id like GET/HEAD do: trim, treat "null" as the
+        // nil UUID, and reject malformed ids up front (issue #4238).
+        let version_id = match version_id {
+            Some(v) => {
+                let trimmed = v.trim();
+                if trimmed.eq_ignore_ascii_case("null") {
+                    Some(Uuid::nil().to_string())
+                } else if Uuid::parse_str(trimmed).is_ok() {
+                    Some(trimmed.to_string())
+                } else {
+                    return Err(s3_error!(InvalidArgument, "Invalid version id specified in copy source"));
+                }
+            }
+            None => None,
+        };
+
         if let Some(ref sc) = storage_class
             && !is_valid_storage_class(sc.as_str())
         {
@@ -4246,10 +4262,12 @@ impl DefaultObjectUsecase {
         validate_table_catalog_object_mutation(&bucket, &key).await?;
 
         // AWS S3 allows self-copy when metadata directive is REPLACE (used to update metadata in-place),
-        // or when an explicit storage class change is requested.
+        // when an explicit storage class change is requested, or when restoring a specific historical
-        // Reject only when neither condition applies.
+        // version onto the current key (source carries a versionId). Reject only a true no-op self-copy
+        // where none of these apply (issue #4238).
         if metadata_directive.as_ref().map(|d| d.as_str()) != Some(MetadataDirective::REPLACE)
             && storage_class.is_none()
+            && version_id.is_none()
             && src_bucket == bucket
             && src_key == key
         {
@@ -7994,6 +8012,71 @@ mod tests {
         assert_ne!(err.code(), &S3ErrorCode::NotImplemented);
     }
 
+    #[tokio::test]
+    async fn execute_copy_object_allows_self_copy_of_historical_version() {
+        // Restoring a specific historical version onto the current key (same bucket/key with a
+        // source versionId, default COPY directive) must pass the self-copy guard (issue #4238).
+        let input = CopyObjectInput::builder()
+            .copy_source(CopySource::Bucket {
+                bucket: "test-bucket".into(),
+                key: "test-key".into(),
+                version_id: Some("11111111-1111-1111-1111-111111111111".into()),
+            })
+            .bucket("test-bucket".to_string())
+            .key("test-key".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let usecase = DefaultObjectUsecase::without_context();
+
+        let err = Box::pin(usecase.execute_copy_object(req)).await.unwrap_err();
+        // Must not be rejected by the self-copy guard; it fails later at store init instead.
+        assert_ne!(err.code(), &S3ErrorCode::InvalidRequest);
+    }
+
+    #[tokio::test]
+    async fn execute_copy_object_allows_self_copy_of_null_version() {
+        // A "null" source version id is a restore of the null version, not a no-op self-copy.
+        let input = CopyObjectInput::builder()
+            .copy_source(CopySource::Bucket {
+                bucket: "test-bucket".into(),
+                key: "test-key".into(),
+                version_id: Some("null".into()),
+            })
+            .bucket("test-bucket".to_string())
+            .key("test-key".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let usecase = DefaultObjectUsecase::without_context();
+
+        let err = Box::pin(usecase.execute_copy_object(req)).await.unwrap_err();
+        assert_ne!(err.code(), &S3ErrorCode::InvalidRequest);
+    }
+
+    #[tokio::test]
+    async fn execute_copy_object_rejects_malformed_copy_source_version_id() {
+        // A malformed (non-null, non-UUID) source version id is rejected up front.
+        let input = CopyObjectInput::builder()
+            .copy_source(CopySource::Bucket {
+                bucket: "src-bucket".into(),
+                key: "src-key".into(),
+                version_id: Some("not-a-uuid".into()),
+            })
+            .bucket("dst-bucket".to_string())
+            .key("dst-key".to_string())
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let usecase = DefaultObjectUsecase::without_context();
+
+        let err = Box::pin(usecase.execute_copy_object(req)).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+    }
+
     #[tokio::test]
     async fn execute_delete_object_rejects_invalid_object_key() {
         let input = DeleteObjectInput::builder()