Merge pull request #2417 from versity/sis/fix-copy-from-delete-marker

fix: stop posix copying data from a delete marker

backend/posix/posix.go

@@ -1618,6 +1618,24 @@ func (p *Posix) isObjDeleteMarker(bucket, object string) (bool, error) {
 	return true, nil
 }
 
+// checkCopySourceDeleteMarker rejects a copy whose source resolves to a
+// delete marker: the key has no current version when the marker is the
+// latest, and a marker named by version id holds no data to copy.
+func (p *Posix) checkCopySourceDeleteMarker(bucket, object, versionId string) error {
+	isDel, err := p.isObjDeleteMarker(bucket, object)
+	if err != nil {
+		return err
+	}
+	if !isDel {
+		return nil
+	}
+	if versionId != "" {
+		return s3err.GetAPIError(s3err.ErrCopySourceDeleteMarker)
+	}
+
+	return s3err.GetAPIError(s3err.ErrNoSuchKey)
+}
+
 // Converts the file to object version. Finds all the object versions,
 // delete markers from the versioning directory and returns
 func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
@@ -4034,6 +4052,9 @@ func (p *Posix) UploadPartCopy(ctx context.Context, upi *s3.UploadPartCopyInput)
 	if strings.HasSuffix(srcObject, "/") != fi.IsDir() {
 		return s3response.CopyPartResult{}, s3err.GetAPIError(s3err.ErrNoSuchKey)
 	}
+	if err := p.checkCopySourceDeleteMarker(srcBucket, srcObject, srcVersionId); err != nil {
+		return s3response.CopyPartResult{}, err
+	}
 	// a directory object holds no data
 	srcSize := fi.Size()
 	if fi.IsDir() {
@@ -6237,6 +6258,9 @@ func (p *Posix) CopyObject(ctx context.Context, input s3response.CopyObjectInput
 	if !strings.HasSuffix(srcObject, "/") && fi.IsDir() {
 		return s3response.CopyObjectOutput{}, s3err.GetAPIError(s3err.ErrNoSuchKey)
 	}
+	if err := p.checkCopySourceDeleteMarker(srcBucket, srcObject, srcVersionId); err != nil {
+		return s3response.CopyObjectOutput{}, err
+	}
 	// a directory object holds no data
 	srcSize := fi.Size()
 	var srcBody io.Reader = f
@@ -6286,11 +6310,18 @@ func (p *Posix) CopyObject(ctx context.Context, input s3response.CopyObjectInput
 	var chType types.ChecksumType
 
 	dstObjdPath := joinPathWithTrailer(p.BucketPath(dstBucket), dstObject)
-	if dstObjdPath == objPath {
-		if input.MetadataDirective == types.MetadataDirectiveCopy {
-			return s3response.CopyObjectOutput{}, s3err.GetAPIError(s3err.ErrInvalidCopyDest)
-		}
+	// A copy of an object onto itself is rejected unless it replaces the
+	// object metadata. Naming a source version makes it a regular copy.
+	selfCopy := dstObjdPath == objPath
+	if selfCopy && srcVersionId == "" &&
+		input.MetadataDirective == types.MetadataDirectiveCopy {
+		return s3response.CopyObjectOutput{}, s3err.GetAPIError(s3err.ErrInvalidCopyDest)
+	}
 
+	// In a versioned bucket a self copy creates a new version like any other
+	// write, so only unversioned buckets are rewritten in place.
+	versioned := p.versioningEnabled() && vStatus != ""
+	if selfCopy && !versioned {
 		// Delete the object metadata
 		err = p.meta.DeleteAttribute(dstBucket, dstObject, metadataHdr)
 		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
@@ -6428,6 +6459,14 @@ func (p *Posix) CopyObject(ctx context.Context, input s3response.CopyObjectInput
 			checksums.Algorithm = input.ChecksumAlgorithm
 		}
 
+		// A self copy publishes the new version over the source path, which
+		// on Windows can't be renamed over while the source is still open.
+		// PutObject reads the body before publishing, so the handle is
+		// released as soon as the data has been staged.
+		if selfCopy {
+			srcBody = &closeOnEOFReader{r: srcBody, c: f}
+		}
+
 		putObjectInput := s3response.PutObjectInput{
 			Bucket:                    &dstBucket,
 			Key:                       &dstObject,
@@ -6464,23 +6503,29 @@ func (p *Posix) CopyObject(ctx context.Context, input s3response.CopyObjectInput
 			putObjectInput.Tagging = input.Tagging
 		}
 
-		res, err := p.PutObject(withCtxNoSlot(ctx), putObjectInput)
-		if err != nil {
-			return s3response.CopyObjectOutput{}, err
-		}
-
-		// copy the source object tagging after the destination object
-		// creation, if tagging directive is "COPY"
+		// read the source tagging before the destination is written, as a
+		// self copy replaces the source object's metadata
+		var srcTagging []byte
+		var hasSrcTagging bool
 		if input.TaggingDirective == types.TaggingDirectiveCopy {
 			tagging, err := p.meta.RetrieveAttribute(nil, srcBucket, srcObject, tagHdr)
 			if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
 				return s3response.CopyObjectOutput{}, fmt.Errorf("get source object tagging: %w", err)
 			}
-			if err == nil {
-				err := p.meta.StoreAttribute(nil, dstBucket, dstObject, tagHdr, tagging)
-				if err != nil {
-					return s3response.CopyObjectOutput{}, fmt.Errorf("set destination object tagging: %w", err)
-				}
+			srcTagging, hasSrcTagging = tagging, err == nil
+		}
+
+		res, err := p.PutObject(withCtxNoSlot(ctx), putObjectInput)
+		if err != nil {
+			return s3response.CopyObjectOutput{}, err
+		}
+
+		// the source tagging is stored after the destination object creation,
+		// if tagging directive is "COPY"
+		if hasSrcTagging {
+			err := p.meta.StoreAttribute(nil, dstBucket, dstObject, tagHdr, srcTagging)
+			if err != nil {
+				return s3response.CopyObjectOutput{}, fmt.Errorf("set destination object tagging: %w", err)
 			}
 		}
 

backend/posix/posix_io_helpers.go

@@ -16,6 +16,7 @@ package posix
 
 import (
 	"bufio"
+	"errors"
 	"io"
 	"log"
 	"sync"
@@ -45,6 +46,24 @@ func (b *bufferedReadCloser) Close() error {
 	return b.c.Close()
 }
 
+// closeOnEOFReader closes c once r is drained, for readers whose source has
+// to be released before the caller is done with the reader.
+type closeOnEOFReader struct {
+	r      io.Reader
+	c      io.Closer
+	closed bool
+}
+
+func (e *closeOnEOFReader) Read(p []byte) (int, error) {
+	n, err := e.r.Read(p)
+	if errors.Is(err, io.EOF) && !e.closed {
+		e.closed = true
+		e.c.Close()
+	}
+
+	return n, err
+}
+
 var odirectUnsupportedWarnByOp sync.Map
 
 func warnODirectUnsupportedOnce(op string, err error) {

backend/posix/posix_io_helpers_test.go

@@ -0,0 +1,79 @@
+// Copyright 2026 Versity Software
+// This file is 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.
+
+package posix
+
+import (
+	"bytes"
+	"io"
+	"strings"
+	"testing"
+)
+
+type countingCloser struct {
+	count int
+}
+
+func (c *countingCloser) Close() error {
+	c.count++
+	return nil
+}
+
+func TestCloseOnEOFReader(t *testing.T) {
+	c := &countingCloser{}
+	r := &closeOnEOFReader{r: strings.NewReader("hello"), c: c}
+
+	buf := make([]byte, 2)
+	n, err := r.Read(buf)
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if n != 2 {
+		t.Fatalf("expected 2 bytes, got %v", n)
+	}
+	if c.count != 0 {
+		t.Fatalf("expected the source to stay open before EOF, closed %v times", c.count)
+	}
+
+	data, err := io.ReadAll(r)
+	if err != nil {
+		t.Fatalf("read all: %v", err)
+	}
+	if !bytes.Equal(data, []byte("llo")) {
+		t.Fatalf("expected the remaining data to be llo, got %s", data)
+	}
+	if c.count != 1 {
+		t.Fatalf("expected the source to be closed once at EOF, closed %v times", c.count)
+	}
+
+	// reads past EOF don't close the source again
+	if _, err := r.Read(buf); err != io.EOF {
+		t.Fatalf("expected io.EOF, got %v", err)
+	}
+	if c.count != 1 {
+		t.Fatalf("expected the source to be closed once, closed %v times", c.count)
+	}
+}
+
+func TestCloseOnEOFReaderEmptySource(t *testing.T) {
+	c := &countingCloser{}
+	r := &closeOnEOFReader{r: strings.NewReader(""), c: c}
+
+	if _, err := io.ReadAll(r); err != nil {
+		t.Fatalf("read all: %v", err)
+	}
+	if c.count != 1 {
+		t.Fatalf("expected the source to be closed once at EOF, closed %v times", c.count)
+	}
+}

s3err/s3err.go

@@ -131,6 +131,7 @@ const (
 	ErrMissingDateHeader
 	ErrGetUploadsWithKey
 	ErrVersionsWithKey
+	ErrCopySourceDeleteMarker
 	ErrInvalidRequest
 	ErrAuthNotSetup
 	ErrNotImplemented
@@ -453,6 +454,11 @@ var errorCodeResponse = map[ErrorCode]APIError{
 		Description:    "There is no such thing as the ?versions sub-resource for a key",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
+	ErrCopySourceDeleteMarker: {
+		Code:           "InvalidRequest",
+		Description:    "The source of a copy request may not specifically refer to a delete marker by version id.",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
 	ErrInvalidRequest: {
 		Code:           "InvalidRequest",
 		Description:    "Invalid Request.",

tests/integration/group-tests.go

@@ -1957,6 +1957,9 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_CopyObject_success)
 	ts.Run(Versioning_CopyObject_non_existing_version_id)
 	ts.Run(Versioning_CopyObject_from_an_object_version)
+	ts.Run(Versioning_CopyObject_from_a_delete_marker)
+	ts.Run(Versioning_CopyObject_to_itself)
+	ts.Run(Versioning_CopyObject_to_itself_from_the_current_version)
 	if !ts.conf.windowsTests {
 		ts.Run(Versioning_CopyObject_special_chars)
 	}
@@ -2026,6 +2029,7 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_UploadPartCopy_encoded_versionid_separator_invalid_versionId)
 	ts.Run(Versioning_UploadPartCopy_non_existing_versionId)
 	ts.Run(Versioning_UploadPartCopy_from_an_object_version)
+	ts.Run(Versioning_UploadPartCopy_from_a_delete_marker)
 	// Object lock configuration
 	ts.Run(Versioning_object_lock_not_enabled_on_bucket_creation)
 	ts.Run(Versioning_Enable_object_lock)
@@ -2054,6 +2058,7 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_WORM_delete_marker_locked_object_compliance_retention)
 	ts.Run(Versioning_WORM_PutObject_overwrite_locked_object)
 	ts.Run(Versioning_WORM_CopyObject_overwrite_locked_object)
+	ts.Run(Versioning_WORM_CopyObject_to_itself_locked_object)
 	ts.Run(Versioning_WORM_CompleteMultipartUpload_overwrite_locked_object)
 	if !ts.conf.windowsTests {
 		ts.Run(Versioning_WORM_remove_delete_marker_under_bucket_default_retention)
@@ -3494,6 +3499,9 @@ func GetIntTests() IntTests {
 		"Versioning_CopyObject_success":                                                    Versioning_CopyObject_success,
 		"Versioning_CopyObject_non_existing_version_id":                                    Versioning_CopyObject_non_existing_version_id,
 		"Versioning_CopyObject_from_an_object_version":                                     Versioning_CopyObject_from_an_object_version,
+		"Versioning_CopyObject_from_a_delete_marker":                                       Versioning_CopyObject_from_a_delete_marker,
+		"Versioning_CopyObject_to_itself":                                                  Versioning_CopyObject_to_itself,
+		"Versioning_CopyObject_to_itself_from_the_current_version":                         Versioning_CopyObject_to_itself_from_the_current_version,
 		"Versioning_CopyObject_special_chars":                                              Versioning_CopyObject_special_chars,
 		"Versioning_HeadObject_invalid_versionId":                                          Versioning_HeadObject_invalid_versionId,
 		"Versioning_HeadObject_non_existing_object_version":                                Versioning_HeadObject_non_existing_object_version,
@@ -3550,6 +3558,7 @@ func GetIntTests() IntTests {
 		"Versioning_UploadPartCopy_encoded_versionid_separator_invalid_versionId":          Versioning_UploadPartCopy_encoded_versionid_separator_invalid_versionId,
 		"Versioning_UploadPartCopy_non_existing_versionId":                                 Versioning_UploadPartCopy_non_existing_versionId,
 		"Versioning_UploadPartCopy_from_an_object_version":                                 Versioning_UploadPartCopy_from_an_object_version,
+		"Versioning_UploadPartCopy_from_a_delete_marker":                                   Versioning_UploadPartCopy_from_a_delete_marker,
 		"Versioning_object_lock_not_enabled_on_bucket_creation":                            Versioning_object_lock_not_enabled_on_bucket_creation,
 		"Versioning_Enable_object_lock":                                                    Versioning_Enable_object_lock,
 		"Versioning_status_switch_to_suspended_with_object_lock":                           Versioning_status_switch_to_suspended_with_object_lock,
@@ -3574,6 +3583,7 @@ func GetIntTests() IntTests {
 		"Versioning_WORM_delete_marker_locked_object_compliance_retention":                 Versioning_WORM_delete_marker_locked_object_compliance_retention,
 		"Versioning_WORM_PutObject_overwrite_locked_object":                                Versioning_WORM_PutObject_overwrite_locked_object,
 		"Versioning_WORM_CopyObject_overwrite_locked_object":                               Versioning_WORM_CopyObject_overwrite_locked_object,
+		"Versioning_WORM_CopyObject_to_itself_locked_object":                               Versioning_WORM_CopyObject_to_itself_locked_object,
 		"Versioning_WORM_CompleteMultipartUpload_overwrite_locked_object":                  Versioning_WORM_CompleteMultipartUpload_overwrite_locked_object,
 		"Versioning_WORM_remove_delete_marker_under_bucket_default_retention":              Versioning_WORM_remove_delete_marker_under_bucket_default_retention,
 		"Versioning_WORM_trailing_slash_counterpart":                                       Versioning_WORM_trailing_slash_counterpart,

tests/integration/utils.go

@@ -2375,6 +2375,28 @@ func createObjVersions(client *s3.Client, bucket, object string, count int, opts
 	return versions, nil
 }
 
+// createDeleteMarker deletes object without a version id, making the
+// resulting delete marker the current version, and returns its version id.
+func createDeleteMarker(client *s3.Client, bucket, object string) (string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	out, err := client.DeleteObject(ctx, &s3.DeleteObjectInput{
+		Bucket: &bucket,
+		Key:    &object,
+	})
+	cancel()
+	if err != nil {
+		return "", err
+	}
+	if out.DeleteMarker == nil || !*out.DeleteMarker {
+		return "", fmt.Errorf("expected a delete marker to be created for %v", object)
+	}
+	if getString(out.VersionId) == "" {
+		return "", fmt.Errorf("expected non empty delete marker versionId for %v", object)
+	}
+
+	return *out.VersionId, nil
+}
+
 // objDataLen returns the data length to upload for key: a directory
 // object can't hold data
 func objDataLen(key string, lgth int64) int64 {

tests/integration/versioning.go

@@ -670,6 +670,254 @@ func Versioning_CopyObject_from_an_object_version(s *S3Conf) error {
 	}, withVersioning(types.BucketVersioningStatusEnabled))
 }
 
+// A copy source that resolves to a delete marker is rejected: the key has no
+// current version when the marker is the latest, and naming the marker by
+// version id is an invalid request. Versions the marker hides stay copyable.
+func Versioning_CopyObject_from_a_delete_marker(s *S3Conf) error {
+	testName := "Versioning_CopyObject_from_a_delete_marker"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dstBucket, dstObj := getBucketName(), "dst-obj"
+		if err := setup(s, dstBucket); err != nil {
+			return err
+		}
+
+		err := forEachKey([]string{"my-obj", "my-dir/"}, func(srcObj string) error {
+			srcObjVersions, err := createObjVersions(s3client, bucket, srcObj, 1)
+			if err != nil {
+				return err
+			}
+
+			delMarker, err := createDeleteMarker(s3client, bucket, srcObj)
+			if err != nil {
+				return err
+			}
+
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.CopyObject(ctx, &s3.CopyObjectInput{
+				Bucket:     &dstBucket,
+				Key:        &dstObj,
+				CopySource: getPtr(fmt.Sprintf("%v/%v", bucket, srcObj)),
+			})
+			cancel()
+			if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchKey)); err != nil {
+				return err
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.CopyObject(ctx, &s3.CopyObjectInput{
+				Bucket: &dstBucket,
+				Key:    &dstObj,
+				CopySource: getPtr(fmt.Sprintf("%v/%v?versionId=%v",
+					bucket, srcObj, delMarker)),
+			})
+			cancel()
+			if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrCopySourceDeleteMarker)); err != nil {
+				return err
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			out, err := s3client.CopyObject(ctx, &s3.CopyObjectInput{
+				Bucket: &dstBucket,
+				Key:    &dstObj,
+				CopySource: getPtr(fmt.Sprintf("%v/%v?versionId=%v",
+					bucket, srcObj, getString(srcObjVersions[0].VersionId))),
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if getString(out.CopySourceVersionId) != getString(srcObjVersions[0].VersionId) {
+				return fmt.Errorf("expected the copy-source-version-id to be %v, instead got %v",
+					getString(srcObjVersions[0].VersionId), getString(out.CopySourceVersionId))
+			}
+
+			return nil
+		})
+		if err != nil {
+			return err
+		}
+
+		return teardown(s, dstBucket)
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
+// A copy of an object onto itself in a versioned bucket is an ordinary
+// write: it creates a new version and leaves the one it replaces untouched.
+// Without a metadata directive there is nothing to replace, so it's rejected.
+func Versioning_CopyObject_to_itself(s *S3Conf) error {
+	testName := "Versioning_CopyObject_to_itself"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		return forEachKey([]string{"my-obj", "my-dir/"}, func(obj string) error {
+			// directory objects always carry the directory content-type
+			srcContentType, dstContentType := "text/plain", "application/json"
+			if strings.HasSuffix(obj, "/") {
+				srcContentType, dstContentType = directoryContentType, directoryContentType
+			}
+
+			srcMeta := map[string]string{"key": "value"}
+			r, err := putObjectWithData(objDataLen(obj, 1234), &s3.PutObjectInput{
+				Bucket:      &bucket,
+				Key:         &obj,
+				ContentType: getPtr("text/plain"),
+				Metadata:    srcMeta,
+			}, s3client)
+			if err != nil {
+				return err
+			}
+
+			srcVersionId := getString(r.res.VersionId)
+
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.CopyObject(ctx, &s3.CopyObjectInput{
+				Bucket:     &bucket,
+				Key:        &obj,
+				CopySource: getPtr(fmt.Sprintf("%v/%v", bucket, obj)),
+			})
+			cancel()
+			if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidCopyDest)); err != nil {
+				return err
+			}
+
+			dstMeta := map[string]string{"new-key": "new-value"}
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			out, err := s3client.CopyObject(ctx, &s3.CopyObjectInput{
+				Bucket:            &bucket,
+				Key:               &obj,
+				CopySource:        getPtr(fmt.Sprintf("%v/%v", bucket, obj)),
+				MetadataDirective: types.MetadataDirectiveReplace,
+				ContentType:       getPtr("application/json"),
+				Metadata:          dstMeta,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			dstVersionId := getString(out.VersionId)
+			if dstVersionId == "" {
+				return fmt.Errorf("expected non empty versionId")
+			}
+			if dstVersionId == srcVersionId {
+				return fmt.Errorf("expected a new versionId, instead got %v", dstVersionId)
+			}
+
+			// the replaced version keeps its own metadata
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			res, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+				Bucket:    &bucket,
+				Key:       &obj,
+				VersionId: &srcVersionId,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if getString(res.ContentType) != srcContentType {
+				return fmt.Errorf("expected the source version content-type to be %v, instead got %v",
+					srcContentType, getString(res.ContentType))
+			}
+			if !areMapsSame(res.Metadata, srcMeta) {
+				return fmt.Errorf("expected the source version metadata to be %v, instead got %v",
+					srcMeta, res.Metadata)
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			res, err = s3client.HeadObject(ctx, &s3.HeadObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if getString(res.VersionId) != dstVersionId {
+				return fmt.Errorf("expected the current versionId to be %v, instead got %v",
+					dstVersionId, getString(res.VersionId))
+			}
+			if getString(res.ContentType) != dstContentType {
+				return fmt.Errorf("expected the new version content-type to be %v, instead got %v",
+					dstContentType, getString(res.ContentType))
+			}
+			if !areMapsSame(res.Metadata, dstMeta) {
+				return fmt.Errorf("expected the new version metadata to be %v, instead got %v",
+					dstMeta, res.Metadata)
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			vRes, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+				Bucket: &bucket,
+				Prefix: &obj,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if len(vRes.Versions) != 2 {
+				return fmt.Errorf("expected 2 object versions, instead got %v", len(vRes.Versions))
+			}
+
+			return nil
+		})
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
+// Naming the current version in the copy source makes a copy onto the same
+// key a regular copy, so it is accepted even without a metadata directive.
+func Versioning_CopyObject_to_itself_from_the_current_version(s *S3Conf) error {
+	testName := "Versioning_CopyObject_to_itself_from_the_current_version"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		return forEachKey([]string{"my-obj", "my-dir/"}, func(obj string) error {
+			versions, err := createObjVersions(s3client, bucket, obj, 1)
+			if err != nil {
+				return err
+			}
+
+			srcVersionId := getString(versions[0].VersionId)
+
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			out, err := s3client.CopyObject(ctx, &s3.CopyObjectInput{
+				Bucket:     &bucket,
+				Key:        &obj,
+				CopySource: getPtr(fmt.Sprintf("%v/%v?versionId=%v", bucket, obj, srcVersionId)),
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if getString(out.CopySourceVersionId) != srcVersionId {
+				return fmt.Errorf("expected the copy-source-version-id to be %v, instead got %v",
+					srcVersionId, getString(out.CopySourceVersionId))
+			}
+			if getString(out.VersionId) == srcVersionId {
+				return fmt.Errorf("expected a new versionId, instead got %v", getString(out.VersionId))
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			res, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+				Bucket:    &bucket,
+				Key:       &obj,
+				VersionId: &srcVersionId,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if getString(res.VersionId) != srcVersionId {
+				return fmt.Errorf("expected the source version to remain, instead got %v",
+					getString(res.VersionId))
+			}
+
+			return nil
+		})
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
 func Versioning_CopyObject_special_chars(s *S3Conf) error {
 	testName := "Versioning_CopyObject_special_chars"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -3009,6 +3257,85 @@ func Versioning_UploadPartCopy_from_an_object_version(s *S3Conf) error {
 	}, withVersioning(types.BucketVersioningStatusEnabled))
 }
 
+// A copy source that resolves to a delete marker is rejected: the key has no
+// current version when the marker is the latest, and naming the marker by
+// version id is an invalid request. Versions the marker hides stay copyable.
+func Versioning_UploadPartCopy_from_a_delete_marker(s *S3Conf) error {
+	testName := "Versioning_UploadPartCopy_from_a_delete_marker"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		return forEachKey([]string{"my-obj", "my-dir/"}, func(srcObj string) error {
+			dstBucket, dstObj := getBucketName(), "dst-obj"
+			if err := setup(s, dstBucket); err != nil {
+				return err
+			}
+
+			srcObjVersions, err := createObjVersions(s3client, bucket, srcObj, 1)
+			if err != nil {
+				return err
+			}
+
+			delMarker, err := createDeleteMarker(s3client, bucket, srcObj)
+			if err != nil {
+				return err
+			}
+
+			mp, err := createMp(s3client, dstBucket, dstObj)
+			if err != nil {
+				return err
+			}
+
+			partNumber := int32(1)
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.UploadPartCopy(ctx, &s3.UploadPartCopyInput{
+				Bucket:     &dstBucket,
+				Key:        &dstObj,
+				UploadId:   mp.UploadId,
+				PartNumber: &partNumber,
+				CopySource: getPtr(fmt.Sprintf("%v/%v", bucket, srcObj)),
+			})
+			cancel()
+			if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchKey)); err != nil {
+				return err
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.UploadPartCopy(ctx, &s3.UploadPartCopyInput{
+				Bucket:     &dstBucket,
+				Key:        &dstObj,
+				UploadId:   mp.UploadId,
+				PartNumber: &partNumber,
+				CopySource: getPtr(fmt.Sprintf("%v/%v?versionId=%v",
+					bucket, srcObj, delMarker)),
+			})
+			cancel()
+			if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrCopySourceDeleteMarker)); err != nil {
+				return err
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			out, err := s3client.UploadPartCopy(ctx, &s3.UploadPartCopyInput{
+				Bucket:     &dstBucket,
+				Key:        &dstObj,
+				UploadId:   mp.UploadId,
+				PartNumber: &partNumber,
+				CopySource: getPtr(fmt.Sprintf("%v/%v?versionId=%v",
+					bucket, srcObj, getString(srcObjVersions[0].VersionId))),
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if getString(out.CopySourceVersionId) != getString(srcObjVersions[0].VersionId) {
+				return fmt.Errorf("expected the copy-source-version-id to be %v, instead got %v",
+					getString(srcObjVersions[0].VersionId), getString(out.CopySourceVersionId))
+			}
+
+			return teardown(s, dstBucket)
+		})
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
 func Versioning_Enable_object_lock(s *S3Conf) error {
 	testName := "Versioning_Enable_object_lock"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -4099,6 +4426,105 @@ func Versioning_WORM_CopyObject_overwrite_locked_object(s *S3Conf) error {
 	}, withLock())
 }
 
+// A copy of a locked object onto itself creates a new version, leaving the
+// locked one and its legal hold in place.
+func Versioning_WORM_CopyObject_to_itself_locked_object(s *S3Conf) error {
+	testName := "Versioning_WORM_CopyObject_to_itself_locked_object"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		return forEachKey([]string{"my-obj", "my-dir/"}, func(obj string) error {
+			versions, err := createObjVersions(s3client, bucket, obj, 1)
+			if err != nil {
+				return err
+			}
+
+			v := versions[0]
+			v.IsLatest = getPtr(false)
+			lockedVersionId := getString(v.VersionId)
+
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.PutObjectLegalHold(ctx, &s3.PutObjectLegalHoldInput{
+				Bucket: &bucket,
+				Key:    &obj,
+				LegalHold: &types.ObjectLockLegalHold{
+					Status: types.ObjectLockLegalHoldStatusOn,
+				},
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			copyResult, err := s3client.CopyObject(ctx, &s3.CopyObjectInput{
+				Bucket:            &bucket,
+				Key:               &obj,
+				CopySource:        getPtr(fmt.Sprintf("%v/%v", bucket, obj)),
+				MetadataDirective: types.MetadataDirectiveReplace,
+				ContentType:       getPtr("application/json"),
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if getString(copyResult.VersionId) == lockedVersionId {
+				return fmt.Errorf("expected a new versionId, instead got %v",
+					getString(copyResult.VersionId))
+			}
+
+			version := types.ObjectVersion{
+				ETag:         copyResult.CopyObjectResult.ETag,
+				IsLatest:     getPtr(true),
+				Key:          &obj,
+				Size:         v.Size,
+				VersionId:    copyResult.VersionId,
+				StorageClass: types.ObjectVersionStorageClassStandard,
+				ChecksumType: copyResult.CopyObjectResult.ChecksumType,
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			out, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+				Bucket: &bucket,
+				Prefix: &obj,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if !compareVersions([]types.ObjectVersion{version, v}, out.Versions) {
+				return fmt.Errorf("expected the object versions to be %v, instead got %v",
+					[]types.ObjectVersion{version, v}, out.Versions)
+			}
+
+			// the legal hold stays on the version it was set on
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			lhRes, err := s3client.GetObjectLegalHold(ctx, &s3.GetObjectLegalHoldInput{
+				Bucket:    &bucket,
+				Key:       &obj,
+				VersionId: &lockedVersionId,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if lhRes.LegalHold.Status != types.ObjectLockLegalHoldStatusOn {
+				return fmt.Errorf("expected the legal hold status to be %v, instead got %v",
+					types.ObjectLockLegalHoldStatusOn, lhRes.LegalHold.Status)
+			}
+
+			return cleanupLockedObjects(s3client, bucket, []objToDelete{
+				{
+					key:                obj,
+					versionId:          lockedVersionId,
+					removeOnlyLeglHold: true,
+				},
+			})
+		})
+	}, withLock())
+}
+
 func Versioning_WORM_CompleteMultipartUpload_overwrite_locked_object(s *S3Conf) error {
 	testName := "Versioning_WORM_CompleteMultipartUpload_overwrite_locked_object"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {