fix: stop posix copying data from a delete marker

`CopyObject` and `UploadPartCopy` never looked at the source's `delete-marker` attribute. A delete marker leaves the data file in place at the object path and only flags it, so a copy whose source resolved to a marker opened that file and succeeded, handing back the data of the version the marker had deleted.

AWS rejects such a copy: `NoSuchKey` when the marker is the current version of the key, and `InvalidRequest` when the marker is named by version id, the latter regardless of whether it is the latest version. Versions the marker hides stay copyable by version id.

Both copy paths now run the resolved source through `checkCopySourceDeleteMarker` once the entry has been validated, returning `NoSuchKey` for an unqualified source and the new `ErrCopySourceDeleteMarker` for one carrying a version id.

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

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,7 @@ 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)
 	if !ts.conf.windowsTests {
 		ts.Run(Versioning_CopyObject_special_chars)
 	}
@@ -2026,6 +2027,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)
@@ -3494,6 +3496,7 @@ 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_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 +3553,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,

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,78 @@ 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))
+}
+
 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 +3081,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 {