From 87746913fc2340e7620d1b43dca49ffe9f613567 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 5 Sep 2026 15:23:49 +0800 Subject: [PATCH] fix: retransmit existing SSE-C replicas instead of metadata-copying them The replication sender's target HEAD carries no SSE-C customer key, so for an SSE-C object the target answers 400 and replicateAll fell into a metadata-only CopyObject that fails on any non-empty SSE-C object (the undecryptable source checksum makes the target recompute one and rewrite the data with a plaintext-sized reader). Once a non-empty SSE-C replica existed, tag, retention and legal-hold changes never reached it, a heal never retransmitted, and a resync neither repaired the replica nor counted it correctly. Forcing a full retransmit alone was not enough: checkPreconditionsPUT rejects a write whose PreserveETag and VersionID match the stored version, only the single-part sealed ETag is truncated before that comparison, so a multipart SSE-C retransmit answered 412, which the sender turns into success. Inherited from upstream ad04afe38. Select replicateAll when the SSE-C HEAD cannot answer (the two previous assignments were dead: rAction still forced the metadata path), exempt an authenticated replica write that carries an SSE-C seal from the duplicate version and ETag rejection (the predicate is the incoming write's restored SSE-C metadata, not what the destination holds), and send the internal replication marker on the resync accounting HEAD for SSE-C objects so a peer answers with the replica metadata instead of 400. Tests: TestAPISSECReplicaRetransmitOverExistingVersion (multipart replica initiation over the same version and ETag answered 412 on main, now 200 with parts sent and plaintext readback; single-part and zero-byte writes unchanged), TestAPISSECReplicaWriteExemptionIsKeyedOnTheIncomingWrite (plaintext replica over an SSE-C version still 412; SSE-C replica over a plaintext version exempted and readable), and TestAPISSECReplicationTargetHead (keyless HEAD 400, missing key 404, marked HEAD 200 with metadata, metadata CopyObject ExcessData on a non-empty object) on ErasureSD and Erasure. Compatibility: every update of an SSE-C object now retransmits its bytes; a peer that rejects the internal marker fails the accounting HEAD as before; the #109 destination fix must be deployed first or a retransmitted replica is transformed again. Fixes pgsty/silo#120 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01L7qJqWwy8oFA6aCXWRzXQe Signed-off-by: Feng Ruohang --- cmd/bucket-replication.go | 25 +- cmd/object-handlers-common.go | 11 +- cmd/replication-ssec-retransmit_test.go | 794 ++++++++++++++++++++++++ 3 files changed, 821 insertions(+), 9 deletions(-) create mode 100644 cmd/replication-ssec-retransmit_test.go diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go index 1b8c846ad..e78ab2edd 100644 --- a/cmd/bucket-replication.go +++ b/cmd/bucket-replication.go @@ -872,19 +872,25 @@ func putReplicationOpts(ctx context.Context, sc string, objInfo ObjectInfo) (put if cc, ok := lkMap.Lookup(xhttp.CacheControl); ok { putOpts.CacheControl = cc } - if mode, ok := lkMap.Lookup(xhttp.AmzObjectLockMode); ok { - rmode := minio.RetentionMode(mode) - putOpts.Mode = rmode + mode, hasMode := lkMap.Lookup(xhttp.AmzObjectLockMode) + retainDateStr, hasRetainDate := lkMap.Lookup(xhttp.AmzObjectLockRetainUntilDate) + if hasMode { + putOpts.Mode = minio.RetentionMode(mode) } - if retainDateStr, ok := lkMap.Lookup(xhttp.AmzObjectLockRetainUntilDate); ok { + // A removed retention is stored as an empty mode and date; it is sent as + // a value-less update that still carries its ordering timestamp. + if hasRetainDate && retainDateStr != "" { rdate, err := amztime.ISO8601Parse(retainDateStr) if err != nil { return putOpts, false, err } putOpts.RetainUntilDate = rdate + } + if hasMode || hasRetainDate { // set retention timestamp in opts retTimestamp := objInfo.ModTime if retainTmstampStr, ok := objInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp]; ok { + var err error retTimestamp, err = time.Parse(time.RFC3339Nano, retainTmstampStr) if err != nil { return putOpts, false, err @@ -1520,11 +1526,14 @@ func (ri ReplicateObjectInfo) replicateAll(ctx context.Context, objectAPI Object return rinfo } } else { - // SSEC objects will refuse HeadObject without the decryption key. - // Ignore the error, since we know the object exists and versioning prevents overwriting existing versions. + // The sender holds no customer key, so the target refuses HeadObject on + // an SSE-C object and the replica cannot be compared. The metadata-only + // CopyObject that a replicateMetadata action would run then fails on any + // non-empty object, because the undecryptable source checksum makes the + // target recompute one and rewrite the data. A full retransmit is the + // only action that completes. if isSSEC && strings.Contains(cerr.Error(), errorCodes[ErrSSEEncryptedObject].Description) { - rinfo.ReplicationStatus = replication.Completed - rinfo.ReplicationAction = replicateNone + rAction = replicateAll goto applyAction } // if target returns error other than NoSuchKey, defer replication attempt diff --git a/cmd/object-handlers-common.go b/cmd/object-handlers-common.go index 90009112a..f8de11ab8 100644 --- a/cmd/object-handlers-common.go +++ b/cmd/object-handlers-common.go @@ -28,6 +28,7 @@ import ( "github.com/minio/minio/internal/amztime" "github.com/minio/minio/internal/bucket/lifecycle" + "github.com/minio/minio/internal/crypto" "github.com/minio/minio/internal/event" "github.com/minio/minio/internal/hash" xhttp "github.com/minio/minio/internal/http" @@ -193,7 +194,15 @@ func checkPreconditionsPUT(ctx context.Context, w http.ResponseWriter, r *http.R etagMatch := opts.PreserveETag != "" && isETagEqual(objInfo.ETag, opts.PreserveETag) vidMatch := opts.VersionID != "" && opts.VersionID == objInfo.VersionID - if etagMatch && vidMatch { + // A matching version and ETag normally mean the destination already holds + // this version, so the write is skipped. They do not establish that for an + // authenticated SSE-C replica write: the destination cannot decrypt or + // re-encrypt the body without the customer key, so it cannot verify the + // replica, and this retransmission is how such a replica is repaired or + // updated. The predicate is the incoming request's restored SSE-C metadata, + // not what the destination happens to hold. + ssecReplica := isReplicaTrusted(r.Context()) && crypto.SSEC.IsEncrypted(opts.UserDefined) + if etagMatch && vidMatch && !ssecReplica { writeHeaders() writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrPreconditionFailed), r.URL) return true diff --git a/cmd/replication-ssec-retransmit_test.go b/cmd/replication-ssec-retransmit_test.go new file mode 100644 index 000000000..e894b8fa0 --- /dev/null +++ b/cmd/replication-ssec-retransmit_test.go @@ -0,0 +1,794 @@ +// Copyright (c) 2015-2026 MinIO, Inc. +// Copyright (c) 2026 PGSTY +// +// This file is part of MinIO Object Storage stack +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +package cmd + +import ( + "bytes" + "crypto/md5" + "encoding/base64" + "encoding/xml" + "io" + "maps" + "net/http" + "net/http/httptest" + "net/url" + "slices" + "strconv" + "strings" + "testing" + "time" + + "github.com/minio/minio/internal/auth" + "github.com/minio/minio/internal/crypto" + xhttp "github.com/minio/minio/internal/http" +) + +// TestAPISSECReplicationTargetHead pins what the replication sender's target +// HEAD sees for an SSE-C object, which is what replicateAll's dispatch relies +// on: a keyless HEAD answers 400 InvalidRequest (so the sender must retransmit +// rather than compare), a missing key still answers 404 NoSuchKey (so a missing +// replica keeps healing), a HEAD carrying the internal replication marker +// answers with the replica metadata (what the resync accounting HEAD now +// sends), and the metadata-only CopyObject the sender used to fall into fails +// with ExcessData on any non-empty object. See pgsty/silo#120. +func TestAPISSECReplicationTargetHead(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testAPISSECReplicationTargetHead, + makeBucketOptions: MakeBucketOptions{VersioningEnabled: true}, + }) +} + +func testAPISSECReplicationTargetHead(obj ObjectLayer, instanceType, bucketName string, + apiRouter http.Handler, credentials auth.Credentials, t *testing.T, +) { + previousTLS := globalIsTLS + globalIsTLS = true + defer func() { globalIsTLS = previousTLS }() + + key := bytes.Repeat([]byte{0x42}, 32) + keyMD5 := md5.Sum(key) + data := bytes.Repeat([]byte("ssec-keyless-head-"), 512) + sseHeaders := map[string]string{ + xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES, + xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key), + xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]), + } + object := "ssec-keyless-head/replica" + putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data, sseHeaders) + + // The replication target credential holds the standard replication actions. + replicator := newObjectAttributesAuthzUser(t, instanceType, bucketName, + `"s3:GetObject","s3:PutObject","s3:ReplicateObject","s3:ReplicateDelete","s3:ReplicateTags"`) + + // Exactly the header set replicateAll's StatObject sends today. + senderHeaders := map[string]string{ + "X-Minio-Source-Proxy-Request": "false", + xhttp.AmzTagDirective: "ACCESS", + } + // The same request with the internal replication marker added. + markedHeaders := map[string]string{ + "X-Minio-Source-Proxy-Request": "false", + xhttp.AmzTagDirective: "ACCESS", + xhttp.MinIOSourceReplicationRequest: "true", + } + + head := func(t *testing.T, creds auth.Credentials, obj, versionID string, headers map[string]string) *httptest.ResponseRecorder { + t.Helper() + headURL := getGetObjectURL("", bucketName, obj) + if versionID != "" { + // replicateAll addresses the source version (minio-go + // api-get-options.go toQueryValues). + headURL += "?versionId=" + versionID + } + req, err := newTestSignedRequestV4(http.MethodHead, headURL, 0, nil, + creds.AccessKey, creds.SecretKey, headers) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + return rec + } + + sseDesc := errorCodes[ErrSSEEncryptedObject].Description + + baseInfo, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + sourceVersion := baseInfo.VersionID + + t.Run("sender-head-today-is-rejected", func(t *testing.T) { + rec := head(t, replicator, object, sourceVersion, senderHeaders) + if rec.Code != http.StatusBadRequest { + t.Fatalf("%s: keyless HEAD status %d, want 400", instanceType, rec.Code) + } + if got := rec.Header().Get("x-minio-error-code"); got != "InvalidRequest" { + t.Fatalf("%s: error code %q, want InvalidRequest", instanceType, got) + } + desc := strings.Trim(rec.Header().Get("x-minio-error-desc"), `"`) + if !strings.Contains(desc, sseDesc) { + t.Fatalf("%s: error desc %q does not carry %q", instanceType, desc, sseDesc) + } + t.Logf("%s: keyless HEAD -> %d %s / %s", instanceType, rec.Code, + rec.Header().Get("x-minio-error-code"), desc) + }) + + t.Run("missing-object-head-is-distinguishable", func(t *testing.T) { + rec := head(t, replicator, "ssec-keyless-head/absent", "", senderHeaders) + if rec.Code != http.StatusNotFound { + t.Fatalf("%s: missing-object HEAD status %d, want 404", instanceType, rec.Code) + } + if got := rec.Header().Get("x-minio-error-code"); got != "NoSuchKey" { + t.Fatalf("%s: missing-object error code %q, want NoSuchKey", instanceType, got) + } + }) + + t.Run("marked-head-answers-with-metadata", func(t *testing.T) { + rec := head(t, replicator, object, sourceVersion, markedHeaders) + if rec.Code != http.StatusOK { + t.Fatalf("%s: marked keyless HEAD status %d, want 200: %s", instanceType, rec.Code, rec.Body.String()) + } + // setObjectHeaders assigns the ETag through the raw header map with the + // non canonical key "ETag", so read it the same way. + etag := "" + if v := rec.Header()[xhttp.ETag]; len(v) > 0 { + etag = strings.Trim(v[0], `"`) + } + clen := rec.Header().Get(xhttp.ContentLength) + lastMod := rec.Header().Get(xhttp.LastModified) + if etag == "" || clen == "" || lastMod == "" { + t.Fatalf("%s: marked HEAD lacks comparison metadata etag=%q len=%q mtime=%q", + instanceType, etag, clen, lastMod) + } + + // What replicateAll would compare this against: the source ObjectInfo it + // obtained from GetObjectNInfo(..., ReplicationRequest: true). + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + gr.Close() + srcSize, err := srcInfo.GetActualSize() + if err != nil { + t.Fatal(err) + } + headSize, err := strconv.ParseInt(clen, 10, 64) + if err != nil { + t.Fatal(err) + } + t.Logf("%s: source ETag=%q (len %d) size=%d ; target HEAD ETag=%q (len %d) size=%d", + instanceType, srcInfo.ETag, len(srcInfo.ETag), srcSize, etag, len(etag), headSize) + if headSize != srcSize { + t.Errorf("%s: getReplicationAction size mismatch: source %d target %d", instanceType, srcSize, headSize) + } + if srcInfo.ETag != etag { + t.Logf("%s: NOTE getReplicationAction would see an ETag mismatch (source keeps the sealed ETag, "+ + "the target HEAD returns the last 32 bytes) and therefore return replicateAll", instanceType) + } + }) + + t.Run("zero-byte-metadata-copy-succeeds", func(t *testing.T) { + // A zero-byte SSE-C object has nothing for the plaintext-sized reader to + // overrun, so the same copy request succeeds. The ExcessData failure is a + // property of non-empty objects. + zero := "ssec-keyless-head/zero" + putCopyChecksumSource(t, apiRouter, credentials, bucketName, zero, nil, sseHeaders) + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, zero, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, + }) + if err != nil { + t.Fatal(err) + } + zi := gr.ObjInfo + gr.Close() + + copySrc := url.QueryEscape(SlashSeparator+bucketName+SlashSeparator+zero) + "?versionId=" + zi.VersionID + headers := map[string]string{ + xhttp.AmzCopySource: copySrc, + xhttp.MinIOSourceReplicationRequest: "true", + } + for k, v := range getCopyObjMetadata(zi, "") { + if strings.EqualFold(k, "content-length") { + continue + } + headers[k] = v + } + headers[xhttp.AmzObjectTagging] = "keyless-head=zero" + req, err := newTestSignedRequestV4(http.MethodPut, + getCopyObjectURL("", bucketName, zero)+"?versionId="+zi.VersionID, 0, nil, + replicator.AccessKey, replicator.SecretKey, headers) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + after, gerr := obj.GetObjectInfo(t.Context(), bucketName, zero, ObjectOptions{}) + if gerr != nil { + t.Fatal(gerr) + } + t.Logf("%s: zero-byte metadata CopyObject -> %d; stored SSE-C=%v size=%d tags=%q", + instanceType, rec.Code, crypto.SSEC.IsEncrypted(after.UserDefined), after.Size, after.UserTags) + if rec.Code != http.StatusOK { + t.Fatalf("%s: zero-byte metadata CopyObject status %d: %s", instanceType, rec.Code, rec.Body.String()) + } + }) + + t.Run("metadata-copy-is-what-resync-runs", func(t *testing.T) { + // Exactly what replicateAll runs at cmd/bucket-replication.go:1582 once + // the keyless HEAD has been misread: a same bucket, same key CopyObject + // built from getCopyObjMetadata plus the replication marker, carrying no + // customer key. getCopyObjMetadata already sets REPLICA status and + // x-amz-tagging-directive: REPLACE, so the request is replica trusted. + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + gr.Close() + + // minio-go addresses the source version on the copy source and in the + // destination query (api-compose-object.go:307,314). + copySrc := url.QueryEscape(SlashSeparator+bucketName+SlashSeparator+object) + "?versionId=" + srcInfo.VersionID + headers := map[string]string{ + xhttp.AmzCopySource: copySrc, + xhttp.MinIOSourceReplicationRequest: "true", + } + copyMeta := getCopyObjMetadata(srcInfo, "") + for k, v := range copyMeta { + // net/http derives the request body length from a literal + // Content-Length header; minio-go relies on req.ContentLength, so + // drop it here to keep the in-process request faithful. + if strings.EqualFold(k, "content-length") { + t.Logf("%s: dropping content-length=%q from the copy metadata", instanceType, v) + continue + } + headers[k] = v + } + t.Logf("%s: copy metadata keys: %v", instanceType, slices.Sorted(maps.Keys(copyMeta))) + copyURL := getCopyObjectURL("", bucketName, object) + "?versionId=" + srcInfo.VersionID + req, err := newTestSignedRequestV4(http.MethodPut, copyURL, 0, nil, + replicator.AccessKey, replicator.SecretKey, headers) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + t.Logf("%s: replication metadata CopyObject (version %s) -> %d %s", instanceType, srcInfo.VersionID, rec.Code, + strings.ReplaceAll(rec.Body.String(), "\n", " ")) + if rec.Code != http.StatusBadRequest { + t.Fatalf("%s: metadata CopyObject status %d, want 400", instanceType, rec.Code) + } + if !strings.Contains(rec.Body.String(), "ExcessData") { + t.Fatalf("%s: metadata CopyObject did not fail with ExcessData: %s", instanceType, rec.Body.String()) + } + + // Whatever the status, the object must still read back with the key. + greq, gerr := newTestSignedRequestV4(http.MethodGet, getGetObjectURL("", bucketName, object), 0, nil, + credentials.AccessKey, credentials.SecretKey, sseHeaders) + if gerr != nil { + t.Fatal(gerr) + } + grec := httptest.NewRecorder() + apiRouter.ServeHTTP(grec, greq) + if grec.Code != http.StatusOK || !bytes.Equal(grec.Body.Bytes(), data) { + t.Errorf("%s: after the replication metadata CopyObject the object no longer reads back: %d (%d bytes)", + instanceType, grec.Code, grec.Body.Len()) + } else { + t.Logf("%s: object still reads back correctly with the customer key", instanceType) + } + }) +} + +// TestAPISSECReplicaRetransmitOverExistingVersion asserts that a full +// retransmit of an SSE-C object reaches the destination when the replica +// already exists with the source version and ETag. checkPreconditionsPUT used +// to reject a matching PreserveETag plus VersionID with 412 for the multipart +// path (the single-part sealed ETag is truncated before the comparison), and +// the sender turns 412 into success, so no part was ever sent and an SSE-C +// replica could never be repaired or updated. See pgsty/silo#120. +func TestAPISSECReplicaRetransmitOverExistingVersion(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testAPISSECReplicaRetransmitOverExistingVersion, + makeBucketOptions: MakeBucketOptions{VersioningEnabled: true}, + }) +} + +func testAPISSECReplicaRetransmitOverExistingVersion(obj ObjectLayer, instanceType, bucketName string, + apiRouter http.Handler, credentials auth.Credentials, t *testing.T, +) { + previousTLS := globalIsTLS + globalIsTLS = true + defer func() { globalIsTLS = previousTLS }() + + key := bytes.Repeat([]byte{0x43}, 32) + keyMD5 := md5.Sum(key) + sseHeaders := map[string]string{ + xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES, + xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key), + xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]), + } + replicator := newObjectAttributesAuthzUser(t, instanceType, bucketName, + `"s3:GetObject","s3:PutObject","s3:ReplicateObject"`) + + replicaHeaders := func(t *testing.T, oi ObjectInfo) map[string]string { + t.Helper() + opts, _, err := putReplicationOpts(t.Context(), "", oi) + if err != nil { + t.Fatal(err) + } + opts.Internal.SourceMTime = time.Time{} + out := make(map[string]string) + for name, values := range opts.Header() { + if len(values) > 0 { + out[name] = values[0] + } + } + out[xhttp.MinIOSourceReplicationRequest] = "true" + out[xhttp.AmzBucketReplicationStatus] = "REPLICA" + out[xhttp.MinIOSourceETag] = oi.ETag + return out + } + + t.Run("single-part-replica-put", func(t *testing.T) { + data := bytes.Repeat([]byte("single-part-ssec-replica-"), 400) + object := "ssec-duplicate/single" + putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data, sseHeaders) + + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + cipher, err := io.ReadAll(gr) + gr.Close() + if err != nil { + t.Fatal(err) + } + // The source changed a tag after the replica was written: the + // retransmit must carry it onto the same version. + srcInfo.UserTags = "retransmit=single" + hdrs := replicaHeaders(t, srcInfo) + putURL := getPutObjectURL("", bucketName, object) + "?versionId=" + srcInfo.VersionID + req, err := newTestSignedRequestV4(http.MethodPut, putURL, int64(len(cipher)), + bytes.NewReader(cipher), replicator.AccessKey, replicator.SecretKey, hdrs) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("%s: single-part replica PUT status %d, want 200: %s", instanceType, rec.Code, rec.Body.String()) + } + assertRetransmittedVersion(t, obj, apiRouter, credentials, bucketName, object, srcInfo.VersionID, "retransmit=single", data, sseHeaders) + }) + + t.Run("zero-byte-replica-put", func(t *testing.T) { + object := "ssec-duplicate/zero" + putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, nil, sseHeaders) + + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + cipher, err := io.ReadAll(gr) + gr.Close() + if err != nil { + t.Fatal(err) + } + srcInfo.UserTags = "retransmit=zero" + hdrs := replicaHeaders(t, srcInfo) + putURL := getPutObjectURL("", bucketName, object) + "?versionId=" + srcInfo.VersionID + req, err := newTestSignedRequestV4(http.MethodPut, putURL, int64(len(cipher)), + bytes.NewReader(cipher), replicator.AccessKey, replicator.SecretKey, hdrs) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("%s: zero-byte replica PUT status %d, want 200: %s", instanceType, rec.Code, rec.Body.String()) + } + assertRetransmittedVersion(t, obj, apiRouter, credentials, bucketName, object, srcInfo.VersionID, "retransmit=zero", nil, sseHeaders) + }) + + t.Run("multipart-replica-newmpu", func(t *testing.T) { + data := bytes.Repeat([]byte("multipart-ssec-replica-"), 4096) + object := "ssec-duplicate/multipart" + + newReq, err := newTestSignedRequestV4(http.MethodPost, getNewMultipartURL("", bucketName, object), + 0, nil, credentials.AccessKey, credentials.SecretKey, sseHeaders) + if err != nil { + t.Fatal(err) + } + newRec := httptest.NewRecorder() + apiRouter.ServeHTTP(newRec, newReq) + if newRec.Code != http.StatusOK { + t.Fatalf("%s: source NewMultipart %d: %s", instanceType, newRec.Code, newRec.Body.String()) + } + var srcInit InitiateMultipartUploadResponse + if err = xmlDecoder(newRec.Body, &srcInit, int64(newRec.Body.Len())); err != nil { + t.Fatal(err) + } + partReq, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectPartURL("", bucketName, object, srcInit.UploadID, "1"), int64(len(data)), + bytes.NewReader(data), credentials.AccessKey, credentials.SecretKey, sseHeaders) + if err != nil { + t.Fatal(err) + } + partRec := httptest.NewRecorder() + apiRouter.ServeHTTP(partRec, partReq) + if partRec.Code != http.StatusOK { + t.Fatalf("%s: source PutPart %d: %s", instanceType, partRec.Code, partRec.Body.String()) + } + body, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{ + {PartNumber: 1, ETag: canonicalizeETag(partRec.Header()[xhttp.ETag][0])}, + }}) + if err != nil { + t.Fatal(err) + } + completeReq, err := newTestSignedRequestV4(http.MethodPost, + getCompleteMultipartUploadURL("", bucketName, object, srcInit.UploadID), int64(len(body)), + bytes.NewReader(body), credentials.AccessKey, credentials.SecretKey, sseHeaders) + if err != nil { + t.Fatal(err) + } + completeRec := httptest.NewRecorder() + apiRouter.ServeHTTP(completeRec, completeReq) + if completeRec.Code != http.StatusOK { + t.Fatalf("%s: source Complete %d: %s", instanceType, completeRec.Code, completeRec.Body.String()) + } + + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + gr.Close() + + srcInfo.UserTags = "retransmit=multipart" + hdrs := replicaHeaders(t, srcInfo) + mpuURL := getNewMultipartURL("", bucketName, object) + "&versionId=" + srcInfo.VersionID + req, err := newTestSignedRequestV4(http.MethodPost, mpuURL, 0, nil, + replicator.AccessKey, replicator.SecretKey, hdrs) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + t.Logf("%s: multipart replica NewMultipartUpload over the same version+ETag -> %d %s (source ETag %q, multipart=%v)", + instanceType, rec.Code, strings.ReplaceAll(rec.Body.String(), "\n", " "), + srcInfo.ETag, crypto.IsMultiPart(srcInfo.UserDefined)) + if rec.Code == http.StatusPreconditionFailed { + t.Fatalf("%s: multipart replica upload short-circuited with 412; the sender turns that into "+ + "success, so no part is ever sent", instanceType) + } + if rec.Code != http.StatusOK { + t.Fatalf("%s: replica NewMultipart status %d: %s", instanceType, rec.Code, rec.Body.String()) + } + + // Finish the replica upload the way the sender does and prove the object + // is still readable with the customer key afterwards. + gr2, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, + }) + if err != nil { + t.Fatal(err) + } + rawPart, err := io.ReadAll(gr2) + gr2.Close() + if err != nil { + t.Fatal(err) + } + var replicaInit InitiateMultipartUploadResponse + if err = xmlDecoder(rec.Body, &replicaInit, int64(rec.Body.Len())); err != nil { + t.Fatal(err) + } + partReq2, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectPartURL("", bucketName, object, replicaInit.UploadID, "1"), int64(len(rawPart)), + bytes.NewReader(rawPart), replicator.AccessKey, replicator.SecretKey, + map[string]string{xhttp.MinIOSourceReplicationRequest: "true"}) + if err != nil { + t.Fatal(err) + } + partRec2 := httptest.NewRecorder() + apiRouter.ServeHTTP(partRec2, partReq2) + if partRec2.Code != http.StatusOK { + t.Fatalf("%s: replica PutPart %d: %s", instanceType, partRec2.Code, partRec2.Body.String()) + } + actualSize, err := srcInfo.GetActualSize() + if err != nil { + t.Fatal(err) + } + completeBody2, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{ + {PartNumber: 1, ETag: canonicalizeETag(partRec2.Header()[xhttp.ETag][0])}, + }}) + if err != nil { + t.Fatal(err) + } + completeReq2, err := newTestSignedRequestV4(http.MethodPost, + getCompleteMultipartUploadURL("", bucketName, object, replicaInit.UploadID), int64(len(completeBody2)), + bytes.NewReader(completeBody2), replicator.AccessKey, replicator.SecretKey, map[string]string{ + xhttp.MinIOSourceReplicationRequest: "true", + xhttp.MinIOSourceMTime: srcInfo.ModTime.Format(time.RFC3339Nano), + xhttp.MinIOSourceETag: srcInfo.ETag, + xhttp.MinIOReplicationActualObjectSize: strconv.FormatInt(actualSize, 10), + }) + if err != nil { + t.Fatal(err) + } + completeRec2 := httptest.NewRecorder() + apiRouter.ServeHTTP(completeRec2, completeReq2) + if completeRec2.Code != http.StatusOK { + t.Fatalf("%s: replica Complete %d: %s", instanceType, completeRec2.Code, completeRec2.Body.String()) + } + assertRetransmittedVersion(t, obj, apiRouter, credentials, bucketName, object, srcInfo.VersionID, "retransmit=multipart", data, sseHeaders) + }) +} + +// assertRetransmittedVersion checks that a retransmit landed on the addressed +// version: the changed tag is stored on it and it still reads back with the +// customer key. +func assertRetransmittedVersion(t *testing.T, obj ObjectLayer, apiRouter http.Handler, credentials auth.Credentials, + bucketName, object, versionID, wantTags string, want []byte, sseHeaders map[string]string, +) { + t.Helper() + info, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{VersionID: versionID}) + if err != nil { + t.Fatalf("version %s after retransmit: %v", versionID, err) + } + if info.UserTags != wantTags { + t.Errorf("version %s tags after retransmit %q, want %q", versionID, info.UserTags, wantTags) + } + if !crypto.SSEC.IsEncrypted(info.UserDefined) { + t.Errorf("version %s lost its SSE-C seal after retransmit", versionID) + } + getReq, err := newTestSignedRequestV4(http.MethodGet, getGetObjectURL("", bucketName, object)+"?versionId="+versionID, + 0, nil, credentials.AccessKey, credentials.SecretKey, sseHeaders) + if err != nil { + t.Fatal(err) + } + getRec := httptest.NewRecorder() + apiRouter.ServeHTTP(getRec, getReq) + if getRec.Code != http.StatusOK || !bytes.Equal(getRec.Body.Bytes(), want) { + t.Fatalf("version %s does not read back with the customer key after retransmit: %d (%d bytes, want %d)", + versionID, getRec.Code, getRec.Body.Len(), len(want)) + } +} + +// TestPutReplicationOptsRetentionRemoval asserts that a source version whose +// retention was removed (stored as an empty mode and date) still builds +// replication options, carrying the removal's ordering timestamp and no value. +func TestPutReplicationOptsRetentionRemoval(t *testing.T) { + removedAt := time.Date(2026, 9, 5, 10, 0, 0, 0, time.UTC) + oi := ObjectInfo{ + Bucket: "b", Name: "o", VersionID: "v1", ModTime: removedAt.Add(-time.Hour), + UserDefined: map[string]string{ + strings.ToLower(xhttp.AmzObjectLockMode): "", + strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): "", + ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: removedAt.Format(time.RFC3339Nano), + }, + } + opts, _, err := putReplicationOpts(t.Context(), "", oi) + if err != nil { + t.Fatalf("putReplicationOpts on a removed retention: %v", err) + } + if opts.Mode != "" || !opts.RetainUntilDate.IsZero() { + t.Errorf("removal sent as a retention: mode %q date %v", opts.Mode, opts.RetainUntilDate) + } + if !opts.Internal.RetentionTimestamp.Equal(removedAt) { + t.Errorf("removal timestamp %v, want %v", opts.Internal.RetentionTimestamp, removedAt) + } + if hdr := opts.Header(); hdr.Get(xhttp.AmzObjectLockMode) != "" || hdr.Get(xhttp.AmzObjectLockRetainUntilDate) != "" || + hdr.Get(xhttp.MinIOSourceObjectRetentionTimestamp) == "" { + t.Errorf("removal headers %v: want no lock value and a retention timestamp", hdr) + } +} + +// TestAPISSECReplicaWriteExemptionIsKeyedOnTheIncomingWrite pins the predicate +// of the duplicate version and ETag exemption: it applies to an authenticated +// replica write that carries an SSE-C seal, whatever the destination holds, and +// not to a plaintext replica write that happens to match an SSE-C destination +// version. See pgsty/silo#120. +func TestAPISSECReplicaWriteExemptionIsKeyedOnTheIncomingWrite(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testAPISSECReplicaWriteExemptionIsKeyedOnTheIncomingWrite, + makeBucketOptions: MakeBucketOptions{VersioningEnabled: true}, + }) +} + +func testAPISSECReplicaWriteExemptionIsKeyedOnTheIncomingWrite(obj ObjectLayer, instanceType, bucketName string, + apiRouter http.Handler, credentials auth.Credentials, t *testing.T, +) { + previousTLS := globalIsTLS + globalIsTLS = true + defer func() { globalIsTLS = previousTLS }() + + key := bytes.Repeat([]byte{0x44}, 32) + keyMD5 := md5.Sum(key) + sseHeaders := map[string]string{ + xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES, + xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key), + xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]), + } + replicator := newObjectAttributesAuthzUser(t, instanceType, bucketName, + `"s3:GetObject","s3:PutObject","s3:ReplicateObject"`) + + rawOf := func(t *testing.T, object string) (ObjectInfo, []byte) { + t.Helper() + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, + }) + if err != nil { + t.Fatal(err) + } + defer gr.Close() + raw, err := io.ReadAll(gr) + if err != nil { + t.Fatal(err) + } + return gr.ObjInfo, raw + } + replicaHeaders := func(t *testing.T, oi ObjectInfo) map[string]string { + t.Helper() + opts, _, err := putReplicationOpts(t.Context(), "", oi) + if err != nil { + t.Fatal(err) + } + opts.Internal.SourceMTime = time.Time{} + out := make(map[string]string) + for name, values := range opts.Header() { + if len(values) > 0 { + out[name] = values[0] + } + } + out[xhttp.MinIOSourceReplicationRequest] = "true" + out[xhttp.AmzBucketReplicationStatus] = "REPLICA" + out[xhttp.MinIOSourceETag] = oi.ETag + return out + } + replicaPut := func(t *testing.T, object, versionID string, body []byte, hdrs map[string]string) *httptest.ResponseRecorder { + t.Helper() + req, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectURL("", bucketName, object)+"?versionId="+versionID, int64(len(body)), + bytes.NewReader(body), replicator.AccessKey, replicator.SecretKey, hdrs) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + return rec + } + + // putSSECMultipart stores a one-part SSE-C multipart object: unlike a + // single-part SSE-C object, whose stored ETag is the sealed one and never + // matches the sender's, a multipart ETag compares equal, which is what + // makes the duplicate version and ETag check reachable at all. + putSSECMultipart := func(t *testing.T, object string, data []byte) { + t.Helper() + newReq, err := newTestSignedRequestV4(http.MethodPost, getNewMultipartURL("", bucketName, object), + 0, nil, credentials.AccessKey, credentials.SecretKey, sseHeaders) + if err != nil { + t.Fatal(err) + } + newRec := httptest.NewRecorder() + apiRouter.ServeHTTP(newRec, newReq) + if newRec.Code != http.StatusOK { + t.Fatalf("%s: NewMultipart %d: %s", instanceType, newRec.Code, newRec.Body.String()) + } + var init InitiateMultipartUploadResponse + if err = xmlDecoder(newRec.Body, &init, int64(newRec.Body.Len())); err != nil { + t.Fatal(err) + } + partReq, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectPartURL("", bucketName, object, init.UploadID, "1"), int64(len(data)), + bytes.NewReader(data), credentials.AccessKey, credentials.SecretKey, sseHeaders) + if err != nil { + t.Fatal(err) + } + partRec := httptest.NewRecorder() + apiRouter.ServeHTTP(partRec, partReq) + if partRec.Code != http.StatusOK { + t.Fatalf("%s: PutPart %d: %s", instanceType, partRec.Code, partRec.Body.String()) + } + body, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{ + {PartNumber: 1, ETag: canonicalizeETag(partRec.Header()[xhttp.ETag][0])}, + }}) + if err != nil { + t.Fatal(err) + } + completeReq, err := newTestSignedRequestV4(http.MethodPost, + getCompleteMultipartUploadURL("", bucketName, object, init.UploadID), int64(len(body)), + bytes.NewReader(body), credentials.AccessKey, credentials.SecretKey, sseHeaders) + if err != nil { + t.Fatal(err) + } + completeRec := httptest.NewRecorder() + apiRouter.ServeHTTP(completeRec, completeReq) + if completeRec.Code != http.StatusOK { + t.Fatalf("%s: Complete %d: %s", instanceType, completeRec.Code, completeRec.Body.String()) + } + } + + t.Run("plaintext-replica-over-ssec-version-is-still-rejected", func(t *testing.T) { + object := "ssec-exemption/ssec-destination" + putSSECMultipart(t, object, bytes.Repeat([]byte("ssec-destination-"), 4096)) + srcInfo, raw := rawOf(t, object) + hdrs := replicaHeaders(t, srcInfo) + // Without the seal the incoming write is a plaintext replica that merely + // carries the stored version and ETag: the duplicate check still applies. + for name := range hdrs { + if strings.HasPrefix(name, "X-Minio-Replication-Server-Side-Encryption-") { + delete(hdrs, name) + } + } + rec := replicaPut(t, object, srcInfo.VersionID, raw, hdrs) + if rec.Code != http.StatusPreconditionFailed { + t.Fatalf("%s: plaintext replica write over an existing SSE-C version answered %d, want 412: %s", + instanceType, rec.Code, rec.Body.String()) + } + }) + + t.Run("ssec-replica-over-plaintext-version-is-exempted", func(t *testing.T) { + object := "ssec-exemption/plain-destination" + // A plaintext version first, then an SSE-C version of the same key so + // the seal is bound to this object path. + putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, []byte("plaintext destination version"), nil) + plainInfo, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + data := bytes.Repeat([]byte("ssec-source-"), 400) + putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data, sseHeaders) + srcInfo, raw := rawOf(t, object) + + // The incoming write carries the SSE-C seal and addresses the plaintext + // version with its ETag: the exemption is decided by the incoming + // write, not by what the destination holds. + hdrs := replicaHeaders(t, srcInfo) + hdrs[xhttp.MinIOSourceETag] = plainInfo.ETag + rec := replicaPut(t, object, plainInfo.VersionID, raw, hdrs) + if rec.Code != http.StatusOK { + t.Fatalf("%s: SSE-C replica write over a matching plaintext version answered %d, want 200: %s", + instanceType, rec.Code, rec.Body.String()) + } + getReq, err := newTestSignedRequestV4(http.MethodGet, + getGetObjectURL("", bucketName, object)+"?versionId="+plainInfo.VersionID, 0, nil, + credentials.AccessKey, credentials.SecretKey, sseHeaders) + if err != nil { + t.Fatal(err) + } + getRec := httptest.NewRecorder() + apiRouter.ServeHTTP(getRec, getReq) + if getRec.Code != http.StatusOK || !bytes.Equal(getRec.Body.Bytes(), data) { + t.Fatalf("%s: the retransmitted version does not read back with the customer key: %d (%d bytes)", + instanceType, getRec.Code, getRec.Body.Len()) + } + }) +}