Merge pull request #86 from pgsty/codex/issue-83-copy-null-version-release

fix: keep rewritten CopyObject data and metadata consistent

cmd/encryption-v1.go

@@ -355,6 +355,29 @@ func rotateKey(ctx context.Context, oldKey []byte, newKeyID string, newKey []byt
 	}
 }
 
+// checkSSECCopySourceKey authenticates the SSE-C copy source key against the
+// sealed object key held in metadata. GetObjectNInfo builds no decryptor for a
+// zero byte object, so a copy whose data path never decrypts anything has to
+// verify the source key explicitly. Mirrors the errors rotateKey reports.
+func checkSSECCopySourceKey(h http.Header, metadata map[string]string, bucket, object string, newKey []byte) error {
+	oldKey, err := ParseSSECopyCustomerRequest(h, metadata)
+	if err != nil {
+		return err
+	}
+	sealedKey, err := crypto.SSEC.ParseMetadata(metadata)
+	if err != nil {
+		return err
+	}
+	var objectKey crypto.ObjectKey
+	if err := objectKey.Unseal(oldKey, sealedKey, crypto.SSEC.String(), bucket, object); err != nil {
+		if subtle.ConstantTimeCompare(oldKey, newKey) == 1 {
+			return errInvalidSSEParameters
+		}
+		return crypto.ErrInvalidCustomerKey
+	}
+	return nil
+}
+
 func newEncryptMetadata(ctx context.Context, kind crypto.Type, keyID string, key []byte, bucket, object string, metadata map[string]string, cryptoCtx kms.Context) (crypto.ObjectKey, error) {
 	var sealedKey crypto.SealedKey
 	switch kind {

cmd/erasure-server-pool.go

@@ -1328,6 +1328,8 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
 		return objInfo, err
 	}
 
+	// CopyObjectHandler predicts the outcome of this decision in
+	// copyRewritesObjectData(); keep the two in sync.
 	if cpSrcDstSame && srcInfo.metadataOnly {
 		// Version ID is set for the destination and source == destination version ID.
 		if dstOpts.VersionID != "" && srcOpts.VersionID == dstOpts.VersionID {

cmd/erasure-sets.go

@@ -839,6 +839,8 @@ func (s *erasureSets) CopyObject(ctx context.Context, srcBucket, srcObject, dstB
 
 	cpSrcDstSame := srcSet == dstSet
 	// Check if this request is only metadata update.
+	// CopyObjectHandler predicts the outcome of this decision in
+	// copyRewritesObjectData(); keep the two in sync.
 	if cpSrcDstSame && srcInfo.metadataOnly {
 		// Version ID is set for the destination and source == destination version ID.
 		// perform an in-place update.

cmd/object-copy-metadata_test.go

@@ -202,3 +202,407 @@ func testAPICopyObjectSSECKeyRotationKeepsCompressionState(obj ObjectLayer, inst
 			instanceType, response.Code, response.Body.Len(), len(data), response.Body.String())
 	}
 }
+
+// TestAPICopyObjectMetadataOnlyNullVersion covers the copy whose source is a
+// null version on a bucket that gained versioning after the object was written.
+// The object layer cannot reference such a version, so it rewrites the data and
+// the recorded compression metadata has to describe the rewritten bytes.
+func TestAPICopyObjectMetadataOnlyNullVersion(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:          t,
+		objAPITest: testAPICopyObjectMetadataOnlyNullVersion,
+		endpoints:  []string{"CopyObject", "PutObject", "GetObject"},
+	})
+}
+
+func testAPICopyObjectMetadataOnlyNullVersion(obj ObjectLayer, instanceType, bucketName string,
+	apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+	restoreCompression := setCopyChecksumCompression(true)
+	compressionRestored := false
+	defer func() {
+		if !compressionRestored {
+			restoreCompression()
+		}
+	}()
+
+	data := bytes.Repeat([]byte("null-version-metadata-copy-"), 64*1024)
+	want := mustChecksum(t, hash.ChecksumCRC32, data)
+	object := "copy-metadata/null-version.txt"
+	putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data,
+		map[string]string{xhttp.AmzChecksumCRC32: want})
+
+	before, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !before.IsCompressed() || before.VersionID != "" {
+		t.Fatalf("%s: invalid null-version precondition: compressed=%v versionID=%q",
+			instanceType, before.IsCompressed(), before.VersionID)
+	}
+
+	// Versioning is enabled after the write, so the object keeps a null version.
+	if _, err := globalBucketMetadataSys.Update(t.Context(), bucketName,
+		bucketVersioningConfig, enabledBucketVersioningConfig); err != nil {
+		t.Fatalf("%s: unable to enable versioning: %v", instanceType, err)
+	}
+	if !globalBucketVersioningSys.PrefixEnabled(bucketName, object) {
+		t.Fatalf("%s: versioning did not become enabled", instanceType)
+	}
+
+	// Without compression the rewritten destination stores plaintext.
+	restoreCompression()
+	compressionRestored = true
+
+	rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, object, object,
+		map[string]string{xhttp.AmzMetadataDirective: "REPLACE"})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("%s: metadata-only CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
+	}
+
+	after := assertCopyChecksum(t, obj, bucketName, object, hash.ChecksumCRC32, data, false, nil)
+	if after.VersionID == "" {
+		t.Fatalf("%s: versioned copy did not create a new version", instanceType)
+	}
+	if got := readCopyChecksumObject(t, obj, bucketName, object, ObjectOptions{}); !bytes.Equal(got, data) {
+		t.Fatalf("%s: copied object body differs: got %d bytes, want %d", instanceType, len(got), len(data))
+	}
+}
+
+func TestAPICopyObjectMetadataOnlyNullVersionCompressesRewrite(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:          t,
+		objAPITest: testAPICopyObjectMetadataOnlyNullVersionCompressesRewrite,
+		endpoints:  []string{"CopyObject", "PutObject", "GetObject"},
+	})
+}
+
+func testAPICopyObjectMetadataOnlyNullVersionCompressesRewrite(obj ObjectLayer, instanceType, bucketName string,
+	apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+	globalCompressConfigMu.Lock()
+	previousCompression := globalCompressConfig
+	globalCompressConfig.Enabled = false
+	globalCompressConfigMu.Unlock()
+	defer func() {
+		globalCompressConfigMu.Lock()
+		globalCompressConfig = previousCompression
+		globalCompressConfigMu.Unlock()
+	}()
+
+	data := bytes.Repeat([]byte("null-version-compress-rewrite-"), 64*1024)
+	want := mustChecksum(t, hash.ChecksumCRC32, data)
+	object := "copy-metadata/null-version-compress.txt"
+	putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data,
+		map[string]string{xhttp.AmzChecksumCRC32: want})
+
+	before, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if before.IsCompressed() || before.VersionID != "" {
+		t.Fatalf("%s: invalid null-version precondition: compressed=%v versionID=%q",
+			instanceType, before.IsCompressed(), before.VersionID)
+	}
+	if _, err := globalBucketMetadataSys.Update(t.Context(), bucketName,
+		bucketVersioningConfig, enabledBucketVersioningConfig); err != nil {
+		t.Fatalf("%s: unable to enable versioning: %v", instanceType, err)
+	}
+
+	restoreCopyCompression := setCopyChecksumCompression(false)
+	defer restoreCopyCompression()
+	rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, object, object,
+		map[string]string{xhttp.AmzMetadataDirective: "REPLACE"})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("%s: metadata-only CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
+	}
+
+	after := assertCopyChecksum(t, obj, bucketName, object, hash.ChecksumCRC32, data, true, nil)
+	if after.VersionID == "" {
+		t.Fatalf("%s: versioned copy did not create a new version", instanceType)
+	}
+	if got := readCopyChecksumObject(t, obj, bucketName, object, ObjectOptions{}); !bytes.Equal(got, data) {
+		t.Fatalf("%s: copied object body differs: got %d bytes, want %d", instanceType, len(got), len(data))
+	}
+}
+
+func TestCopyRewritesObjectData(t *testing.T) {
+	tests := []struct {
+		name         string
+		metadataOnly bool
+		srcOpts      ObjectOptions
+		dstOpts      ObjectOptions
+		want         bool
+	}{
+		{
+			name: "data copy always rewrites",
+			want: true,
+		},
+		// PostRestoreObjectHandler, updateRestoreMetadata and batchKeyRotate all
+		// address the same version on both sides and never set Versioned, so they
+		// only ever reach these two cases.
+		{
+			name:         "unversioned in-place metadata update",
+			metadataOnly: true,
+		},
+		{
+			name:         "addressed version updated in place",
+			metadataOnly: true,
+			srcOpts:      ObjectOptions{VersionID: "v1"},
+			dstOpts:      ObjectOptions{VersionID: "v1"},
+		},
+		{
+			name:         "versioned self referential version",
+			metadataOnly: true,
+			srcOpts:      ObjectOptions{VersionID: "v1"},
+			dstOpts:      ObjectOptions{Versioned: true},
+		},
+		{
+			name:         "versioned null source version cannot be referenced",
+			metadataOnly: true,
+			dstOpts:      ObjectOptions{Versioned: true},
+			want:         true,
+		},
+		{
+			name:         "suspended destination with an addressed source version",
+			metadataOnly: true,
+			srcOpts:      ObjectOptions{VersionID: "v1"},
+			dstOpts:      ObjectOptions{VersionSuspended: true, VersionID: nullVersionID},
+			want:         true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := copyRewritesObjectData(tt.metadataOnly, tt.srcOpts, tt.dstOpts); got != tt.want {
+				t.Fatalf("copyRewritesObjectData() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+// TestAPICopyObjectSSECKeyRotationNullVersion covers an SSE-C key rotation whose
+// source is a null version on a bucket that gained versioning after the object
+// was written. A rotation only rewraps the object key held in metadata, so it
+// may not take the metadata-only path when the object layer stores new object
+// data; the rotation has to re-encrypt instead.
+func TestAPICopyObjectSSECKeyRotationNullVersion(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:          t,
+		objAPITest: testAPICopyObjectSSECKeyRotationNullVersion,
+		endpoints:  []string{"CopyObject", "PutObject", "GetObject"},
+	})
+}
+
+func testAPICopyObjectSSECKeyRotationNullVersion(obj ObjectLayer, instanceType, bucketName string,
+	apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+	testAPICopyObjectSSECKeyRotationNullVersionWithCompression(obj, instanceType, bucketName,
+		apiRouter, credentials, false, t)
+}
+
+func TestAPICopyObjectSSECKeyRotationNullVersionCompressesRewrite(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:          t,
+		objAPITest: testAPICopyObjectSSECKeyRotationNullVersionCompressesRewrite,
+		endpoints:  []string{"CopyObject", "PutObject", "GetObject"},
+	})
+}
+
+func testAPICopyObjectSSECKeyRotationNullVersionCompressesRewrite(obj ObjectLayer, instanceType, bucketName string,
+	apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+	testAPICopyObjectSSECKeyRotationNullVersionWithCompression(obj, instanceType, bucketName,
+		apiRouter, credentials, true, t)
+}
+
+func testAPICopyObjectSSECKeyRotationNullVersionWithCompression(obj ObjectLayer, instanceType, bucketName string,
+	apiRouter http.Handler, credentials auth.Credentials, compressAtCopy bool, t *testing.T,
+) {
+	previousTLS := globalIsTLS
+	globalIsTLS = true
+	defer func() { globalIsTLS = previousTLS }()
+
+	data := bytes.Repeat([]byte("key-rotation-null-version-"), 64*1024)
+	object := "copy-metadata/key-rotation-null.txt"
+	oldKey := bytes.Repeat([]byte{0x11}, 32)
+	oldMD5 := md5.Sum(oldKey)
+	newKey := bytes.Repeat([]byte{0x22}, 32)
+	newMD5 := md5.Sum(newKey)
+
+	putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data, map[string]string{
+		xhttp.AmzChecksumCRC32:                         mustChecksum(t, hash.ChecksumCRC32, data),
+		xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
+		xhttp.AmzServerSideEncryptionCustomerKey:       base64.StdEncoding.EncodeToString(oldKey),
+		xhttp.AmzServerSideEncryptionCustomerKeyMD5:    base64.StdEncoding.EncodeToString(oldMD5[:]),
+	})
+	before, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if before.VersionID != "" {
+		t.Fatalf("%s: invalid null-version precondition: versionID=%q", instanceType, before.VersionID)
+	}
+
+	// Versioning is enabled after the write, so the object keeps a null version.
+	if _, err := globalBucketMetadataSys.Update(t.Context(), bucketName,
+		bucketVersioningConfig, enabledBucketVersioningConfig); err != nil {
+		t.Fatalf("%s: unable to enable versioning: %v", instanceType, err)
+	}
+	if !globalBucketVersioningSys.PrefixEnabled(bucketName, object) {
+		t.Fatalf("%s: versioning did not become enabled", instanceType)
+	}
+	if compressAtCopy {
+		restoreCompression := setCopyChecksumCompression(true)
+		defer restoreCompression()
+	}
+
+	rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, object, object, map[string]string{
+		xhttp.AmzServerSideEncryptionCustomerAlgorithm:     xhttp.AmzEncryptionAES,
+		xhttp.AmzServerSideEncryptionCustomerKey:           base64.StdEncoding.EncodeToString(newKey),
+		xhttp.AmzServerSideEncryptionCustomerKeyMD5:        base64.StdEncoding.EncodeToString(newMD5[:]),
+		xhttp.AmzServerSideEncryptionCopyCustomerAlgorithm: xhttp.AmzEncryptionAES,
+		xhttp.AmzServerSideEncryptionCopyCustomerKey:       base64.StdEncoding.EncodeToString(oldKey),
+		xhttp.AmzServerSideEncryptionCopyCustomerKeyMD5:    base64.StdEncoding.EncodeToString(oldMD5[:]),
+	})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("%s: key rotation failed: %d %s", instanceType, rec.Code, rec.Body.String())
+	}
+
+	assertCopyChecksumResponse(t, rec, hash.ChecksumCRC32, data)
+
+	getHeaders := map[string]string{
+		xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
+		xhttp.AmzServerSideEncryptionCustomerKey:       base64.StdEncoding.EncodeToString(newKey),
+		xhttp.AmzServerSideEncryptionCustomerKeyMD5:    base64.StdEncoding.EncodeToString(newMD5[:]),
+	}
+	req, err := newTestSignedRequestV4(http.MethodGet, getGetObjectURL("", bucketName, object),
+		0, nil, credentials.AccessKey, credentials.SecretKey, getHeaders)
+	if err != nil {
+		t.Fatalf("failed to build GetObject request: %v", err)
+	}
+	response := httptest.NewRecorder()
+	apiRouter.ServeHTTP(response, req)
+	if response.Code != http.StatusOK || !bytes.Equal(response.Body.Bytes(), data) {
+		t.Fatalf("%s: post-rotation GetObject returned %d with %d bytes, want 200 with %d bytes: %s",
+			instanceType, response.Code, response.Body.Len(), len(data), response.Body.String())
+	}
+
+	decryptHeaders := http.Header{}
+	for key, value := range getHeaders {
+		decryptHeaders.Set(key, value)
+	}
+	after := assertCopyChecksum(t, obj, bucketName, object, hash.ChecksumCRC32, data, compressAtCopy, decryptHeaders)
+	if after.VersionID == "" {
+		t.Fatalf("%s: rotation into a versioned bucket did not create a new version", instanceType)
+	}
+	// The rotation could not be applied in place, so the object was re-encrypted
+	// under a fresh object key. That regenerates the encrypted ETag, unlike an
+	// in-place rotation which leaves the stored bytes and the ETag alone.
+	if after.ETag == before.ETag {
+		t.Fatalf("%s: re-encrypting rotation kept the source ETag %q", instanceType, after.ETag)
+	}
+}
+
+// TestAPICopyObjectSSECKeyRotationNullVersionWrongKey pins the source key
+// authentication of the re-encrypting fallback. A zero byte source has no data
+// to decrypt, so the copy would otherwise reach the destination write without
+// ever proving the caller holds the current key.
+func TestAPICopyObjectSSECKeyRotationNullVersionWrongKey(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:          t,
+		objAPITest: testAPICopyObjectSSECKeyRotationNullVersionWrongKey,
+		endpoints:  []string{"CopyObject", "PutObject", "GetObject"},
+	})
+}
+
+func testAPICopyObjectSSECKeyRotationNullVersionWrongKey(obj ObjectLayer, instanceType, bucketName string,
+	apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+	previousTLS := globalIsTLS
+	globalIsTLS = true
+	defer func() { globalIsTLS = previousTLS }()
+
+	object := "copy-metadata/key-rotation-null-empty.txt"
+	oldKey := bytes.Repeat([]byte{0x11}, 32)
+	oldMD5 := md5.Sum(oldKey)
+	wrongKey := bytes.Repeat([]byte{0x33}, 32)
+	wrongMD5 := md5.Sum(wrongKey)
+	newKey := bytes.Repeat([]byte{0x22}, 32)
+	newMD5 := md5.Sum(newKey)
+
+	putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, nil, map[string]string{
+		xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
+		xhttp.AmzServerSideEncryptionCustomerKey:       base64.StdEncoding.EncodeToString(oldKey),
+		xhttp.AmzServerSideEncryptionCustomerKeyMD5:    base64.StdEncoding.EncodeToString(oldMD5[:]),
+	})
+	before, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if before.Size != 0 || before.VersionID != "" || len(before.Checksum) != 0 {
+		t.Fatalf("%s: invalid empty null-version precondition: size=%d versionID=%q checksum=%d",
+			instanceType, before.Size, before.VersionID, len(before.Checksum))
+	}
+
+	if _, err := globalBucketMetadataSys.Update(t.Context(), bucketName,
+		bucketVersioningConfig, enabledBucketVersioningConfig); err != nil {
+		t.Fatalf("%s: unable to enable versioning: %v", instanceType, err)
+	}
+
+	rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, object, object, map[string]string{
+		xhttp.AmzServerSideEncryptionCustomerAlgorithm:     xhttp.AmzEncryptionAES,
+		xhttp.AmzServerSideEncryptionCustomerKey:           base64.StdEncoding.EncodeToString(newKey),
+		xhttp.AmzServerSideEncryptionCustomerKeyMD5:        base64.StdEncoding.EncodeToString(newMD5[:]),
+		xhttp.AmzServerSideEncryptionCopyCustomerAlgorithm: xhttp.AmzEncryptionAES,
+		xhttp.AmzServerSideEncryptionCopyCustomerKey:       base64.StdEncoding.EncodeToString(wrongKey),
+		xhttp.AmzServerSideEncryptionCopyCustomerKeyMD5:    base64.StdEncoding.EncodeToString(wrongMD5[:]),
+	})
+	if rec.Code != http.StatusForbidden {
+		t.Fatalf("%s: rotation with an incorrect source key returned %d, want %d: %s",
+			instanceType, rec.Code, http.StatusForbidden, rec.Body.String())
+	}
+
+	rec = copyChecksumRequest(t, apiRouter, credentials, bucketName, object, object, map[string]string{
+		xhttp.AmzServerSideEncryptionCustomerAlgorithm:     xhttp.AmzEncryptionAES,
+		xhttp.AmzServerSideEncryptionCustomerKey:           base64.StdEncoding.EncodeToString(newKey),
+		xhttp.AmzServerSideEncryptionCustomerKeyMD5:        base64.StdEncoding.EncodeToString(newMD5[:]),
+		xhttp.AmzServerSideEncryptionCopyCustomerAlgorithm: xhttp.AmzEncryptionAES,
+		xhttp.AmzServerSideEncryptionCopyCustomerKey:       base64.StdEncoding.EncodeToString(newKey),
+		xhttp.AmzServerSideEncryptionCopyCustomerKeyMD5:    base64.StdEncoding.EncodeToString(newMD5[:]),
+	})
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("%s: rotation with equal invalid keys returned %d, want %d: %s",
+			instanceType, rec.Code, http.StatusBadRequest, rec.Body.String())
+	}
+
+	after, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if after.VersionID != "" {
+		t.Fatalf("%s: rejected rotation still created version %q", instanceType, after.VersionID)
+	}
+
+	// The object stays readable with the key it was written under.
+	req, err := newTestSignedRequestV4(http.MethodGet, getGetObjectURL("", bucketName, object),
+		0, nil, credentials.AccessKey, credentials.SecretKey, map[string]string{
+			xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
+			xhttp.AmzServerSideEncryptionCustomerKey:       base64.StdEncoding.EncodeToString(oldKey),
+			xhttp.AmzServerSideEncryptionCustomerKeyMD5:    base64.StdEncoding.EncodeToString(oldMD5[:]),
+		})
+	if err != nil {
+		t.Fatalf("failed to build GetObject request: %v", err)
+	}
+	response := httptest.NewRecorder()
+	apiRouter.ServeHTTP(response, req)
+	if response.Code != http.StatusOK || response.Body.Len() != 0 {
+		t.Fatalf("%s: original object no longer readable: %d with %d bytes: %s",
+			instanceType, response.Code, response.Body.Len(), response.Body.String())
+	}
+}

cmd/object-handlers.go

@@ -1179,6 +1179,31 @@ func isRemoteCallRequired(ctx context.Context, bucket string, objAPI ObjectLayer
 	return false
 }
 
+// copyRewritesObjectData reports whether the object layer stores new object data
+// for this copy instead of updating metadata in place or adding a
+// self-referential version. It mirrors the metadata-only decision taken by
+// erasureServerPools.CopyObject and erasureSets.CopyObject. CopyObjectHandler
+// has to predict that decision because the compression metadata it records must
+// describe whichever bytes are finally stored. metadataOnly already excludes
+// legacy sources, which the object layer always rewrites.
+func copyRewritesObjectData(metadataOnly bool, srcOpts, dstOpts ObjectOptions) bool {
+	if !metadataOnly {
+		return true
+	}
+	switch {
+	case dstOpts.VersionID != "" && srcOpts.VersionID == dstOpts.VersionID:
+		// In-place update of the addressed version.
+		return false
+	case !dstOpts.Versioned && srcOpts.VersionID == "":
+		// In-place update of an unversioned object.
+		return false
+	case dstOpts.Versioned && srcOpts.VersionID != dstOpts.VersionID:
+		// A new version referencing the existing data.
+		return false
+	}
+	return true
+}
+
 // CopyObjectHandler - Copy Object
 // ----------
 // This implementation of the PUT operation adds an object to a bucket
@@ -1463,12 +1488,39 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 		}
 	}
 
+	// Name the source version explicitly so a metadata-only copy into a
+	// versioned bucket adds a self-referential version instead of rewriting the
+	// object data. A null source version cannot be referenced this way.
+	copySrcOpts := srcOpts
+	if dstOpts.Versioned && copySrcOpts.VersionID == "" {
+		copySrcOpts.VersionID = srcInfo.VersionID
+	}
+
+	// A key rotation rewraps the object key held in metadata; it never
+	// re-encrypts the stored bytes. When the object layer stores new object
+	// data instead, the rotation has to go through the regular re-encrypting
+	// copy, or the destination ends up holding plaintext under metadata that
+	// claims the object is encrypted.
+	canRotateKeyInPlace := !srcInfo.Legacy &&
+		!copyRewritesObjectData(srcInfo.metadataOnly, copySrcOpts, dstOpts)
+
+	// The rotation shortcut authenticates the source key by unsealing it. The
+	// re-encrypting fallback authenticates it only through the source decryptor,
+	// which GetObjectNInfo skips for a zero byte object, so check it here before
+	// the destination is written under the new key.
+	if cpSrcDstSame && sseCopyC && sseC && !chStorageClass && !canRotateKeyInPlace {
+		if err := checkSSECCopySourceKey(r.Header, srcInfo.UserDefined, srcBucket, srcObject, newKey); err != nil {
+			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+			return
+		}
+	}
+
 	// If src == dst and either
 	// - the object is encrypted using SSE-C and two different SSE-C keys are present
 	// - the object is encrypted using SSE-S3 and the SSE-S3 header is present
 	// - the object storage class is not changing
 	// then execute a key rotation.
-	if cpSrcDstSame && (sseCopyC && sseC) && !chStorageClass {
+	if cpSrcDstSame && (sseCopyC && sseC) && !chStorageClass && canRotateKeyInPlace {
 		oldKey, err = ParseSSECopyCustomerRequest(r.Header, srcInfo.UserDefined)
 		if err != nil {
 			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
@@ -1708,8 +1760,12 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 		srcInfo.UserDefined[ReservedMetadataPrefixLower+ReplicationStatus] = dsc.PendingStatus()
 		srcInfo.UserDefined[ReservedMetadataPrefixLower+ReplicationTimestamp] = UTCNow().Format(time.RFC3339Nano)
 	}
-	// Compression metadata must describe data that is actually rewritten.
+	// srcInfo.metadataOnly is still cleared below for legacy sources and for
-	if !srcInfo.metadataOnly || srcInfo.Legacy || dstOpts.WantServerSideChecksumType.IsSet() {
+	// server-side checksum recomputation; both of those rewrite the object data.
+	metadataOnly := srcInfo.metadataOnly && !srcInfo.Legacy && !dstOpts.WantServerSideChecksumType.IsSet()
+
+	// Compression metadata must describe the bytes that are actually stored.
+	if copyRewritesObjectData(metadataOnly, copySrcOpts, dstOpts) {
 		if isDstCompressed {
 			maps.Copy(srcInfo.UserDefined, compressMetadata)
 		} else {
@@ -1808,11 +1864,6 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 
 		copyObjectFn := objectAPI.CopyObject
 
-		copySrcOpts := srcOpts
-		if srcInfo.metadataOnly && dstOpts.Versioned && copySrcOpts.VersionID == "" {
-			copySrcOpts.VersionID = srcInfo.VersionID
-		}
-
 		// Copy source object to destination, if source and destination
 		// object is same then only metadata is updated.
 		objInfo, err = copyObjectFn(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, copySrcOpts, dstOpts)