From 87746913fc2340e7620d1b43dca49ffe9f613567 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 5 Sep 2026 15:23:49 +0800 Subject: [PATCH 1/5] 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()) + } + }) +} From 109d824e5f67d365b72f843e6fa66751ab267a16 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 5 Sep 2026 17:45:12 +0800 Subject: [PATCH 2/5] fix: retransmit and re-order Object Lock for SSE-C replicas (single erasure set) Issue #120 routes an existing SSE-C replica through PutObjectHandler and NewMultipartUploadHandler. On main those handlers assigned the incoming retention and legal hold directly, without the source-timestamp ordering #111 added to CopyObjectHandler and without persisting the ordering timestamps, so in active-active replication a retransmit carrying an older value could overwrite a destination version's newer lock state. Share #111's ordering decision as applyReplicatedObjectLock in cmd/bucket-object-lock.go and call it from CopyObject, PUT and multipart initiation. A request that is not an actual trusted replica keeps ordinary write semantics (a validated value is applied and stamped now); only a real replica update is ordered against the stored version, so a marker-only peer write no longer drops a validated hold or default retention. CopyObject keeps its SSE-C key-rotation encMetadata reconciliation inline. putReplicationOpts now emits a stored retention ordering timestamp even when the value keys are absent, so a removal recorded on the retransmit PUT path still replicates onward. replicateAll marks Failed and carries the error when putReplicationOpts fails. The handler decision is made against the version as it stands then, which a concurrent lock update can outrun before the write commits, and for multipart across the whole initiation-to-completion span. Close that window under the object write lock the receiving erasure set holds: a trusted SSE-C replica full write sets ObjectOptions.ReplicaLockReconcile, and erasureObjects.PutObject and CompleteMultipartUpload re-run the ordering (reconcileStoredObjectLock, which orders retention and legal hold independently by their reserved timestamps) against the destination version read on that set before committing. Persisted upload metadata records the null version as an empty VersionID, so completion looks that up as the null version rather than the latest. The reconcile runs only against an existing version; a not-found destination keeps the write's own accepted lock, including a pre-upgrade upload that persisted values without ordering timestamps, and a non-not-found read error fails the write. Scoped to the SSE-C paths this issue enables; CopyObject is left as #111 wrote it. Scope: this orders Object Lock against the destination version under the write lock and is correct for a single erasure set. A multi-pool deployment -- where a version can have duplicate copies across pools, object ModTime ties do not track per-field lock timestamps, and the object namespace lock is per-pool -- needs a cross-pool lock-safe reconcile and is deliberately out of scope here, tracked in pgsty/silo#TBD-multipool-lock. Tests: TestAPISSECReplicaRetransmitObjectLockOrdering and its multipart sibling; TestAPIReplicaMultipartNewerHoldSurvivesCompletion and TestReplicaPutObjectLockReconcileUnderWriteLock (a hold or retention reaching the version after the handler decision, or after multipart initiation, survives the commit; a pre-upgrade upload on an absent version keeps its lock); TestReplicaLockReconcileNullVersion (a null-version completion reconciles the null version, not a coexisting UUID version, and an absent null version keeps its accepted lock); TestAPIReplicaMarkerOnlyAppliesObjectLock; TestReplicaStoredLock; the timestamp-only putReplicationOpts round trip; and the retransmit, exemption and target-head tests. The #111 CopyObject replica suite and the existing #120 suite stay green, as do the PUT/multipart handler and object-layer regression suites. Compatibility: the shared helper preserves #111's CopyObject behavior; a non-replica PUT or multipart initiation that sets Object Lock now also stamps the reserved ordering timestamp, matching CopyObject since #111; only trusted SSE-C replica writes take the in-lock reconcile. Refs pgsty/silo#120 Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01L7qJqWwy8oFA6aCXWRzXQe Signed-off-by: Feng Ruohang --- cmd/bucket-object-lock.go | 114 +++ cmd/bucket-replication.go | 19 +- cmd/erasure-multipart.go | 66 +- cmd/erasure-object.go | 32 +- cmd/object-api-interface.go | 1 + cmd/object-handlers.go | 78 +- cmd/object-multipart-handlers.go | 41 +- cmd/replication-ssec-retransmit_test.go | 1000 +++++++++++++++++++++++ 8 files changed, 1268 insertions(+), 83 deletions(-) diff --git a/cmd/bucket-object-lock.go b/cmd/bucket-object-lock.go index e8cb31114..e5828e2dd 100644 --- a/cmd/bucket-object-lock.go +++ b/cmd/bucket-object-lock.go @@ -25,6 +25,7 @@ import ( "strings" "time" + "github.com/minio/minio/internal/amztime" "github.com/minio/minio/internal/auth" objectlock "github.com/minio/minio/internal/bucket/object/lock" xhttp "github.com/minio/minio/internal/http" @@ -407,3 +408,116 @@ func (s objectLockState) restoreLegalHold(metadata map[string]string) { } metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = s.legalHold } + +// replicaStoredLock reads the Object Lock state stored on the addressed version +// so a trusted replica write can order its update against it. A missing object +// or version yields an empty state, which is correct for the first write of a +// version; any other read error is returned so the caller fails the write rather +// than ordering an incoming update against lock state it merely failed to read +// (an older incoming value must not win over a newer stored one just because the +// read timed out). +func replicaStoredLock(ctx context.Context, getObjectInfo GetObjectInfoFn, bucket, object, versionID string) (objectLockState, error) { + oi, err := getObjectInfo(ctx, bucket, object, ObjectOptions{VersionID: versionID}) + switch { + case err == nil: + return storedObjectLockState(oi.UserDefined), nil + case isErrObjectNotFound(err) || isErrVersionNotFound(err): + return objectLockState{}, nil + default: + return objectLockState{}, err + } +} + +// applyReplicatedObjectLock writes the retention and legal-hold decision into +// metadata for a PUT, CopyObject, or multipart-initiation request. A request +// that is not an actual trusted replica -- a normal user write, or a trusted +// peer that carried the replication marker without REPLICA status -- takes +// ordinary write semantics: a validated value is applied and stamped now, and a +// missing value is left as is. Only an actual replica update is ordered against +// the state already stored on the addressed version, so a stale value cannot +// overwrite a newer one and a full retransmit cannot roll a destination back. +// The stored argument is meaningful only for a replica; callers pass an empty +// state otherwise. Only the two Object Lock keys and their reserved ordering +// timestamps are touched; any encryption-metadata reconciliation stays with the +// caller. +func applyReplicatedObjectLock(metadata map[string]string, stored objectLockState, + replicaTrusted bool, + retentionMode objectlock.RetMode, retentionDate objectlock.RetentionDate, + legalHold objectlock.ObjectLegalHold, srcRetentionTimestamp, srcLegalholdTimestamp time.Time, +) { + switch { + case !replicaTrusted: + // Ordinary write semantics: apply a validated retention and stamp it now; + // a missing value carries no instruction, so leave the metadata as it is. + if retentionMode.Valid() { + metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode) + metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC()) + metadata[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = UTCNow().Format(time.RFC3339Nano) + } + case !stored.retentionIsOlderThan(srcRetentionTimestamp): + // The stored update is at least as new as this replica's, or the replica + // carries no ordering timestamp: keep what is stored. This is also how a + // stale retransmit is rejected. + stored.restoreRetention(metadata) + default: + // The replica update wins. A removal carries no value but still records + // the source timestamp that orders it. + if retentionMode.Valid() { + metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode) + metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC()) + } + metadata[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = srcRetentionTimestamp.UTC().Format(time.RFC3339Nano) + } + + // Legal hold has no removal in S3: an explicitly empty header is already + // rejected as an invalid status, so the only value-less shape that gets here + // is an absent one, which conveys no legal-hold change. Only a valid status + // can win. + switch { + case !replicaTrusted: + if legalHold.Status.Valid() { + metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status) + metadata[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = UTCNow().Format(time.RFC3339Nano) + } + case legalHold.Status.Valid() && stored.legalHoldIsOlderThan(srcLegalholdTimestamp): + metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status) + metadata[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = srcLegalholdTimestamp.UTC().Format(time.RFC3339Nano) + default: + stored.restoreLegalHold(metadata) + } +} + +// reconcileStoredObjectLock re-orders the Object Lock already written into +// metadata against the state currently stored on the destination version, both +// compared by their reserved ordering timestamps. It runs inside the object +// layer under the namespace write lock that guards the version replacement, +// after the destination version is read and before the new one is committed, so +// a replica update whose ordering was decided at handler time (or, for multipart, +// at initiation) cannot overwrite a newer lock update that reached the version in +// between. metadata already carries the incoming update with its source +// timestamps; a stored value that is not older than the incoming one is put back, +// which for a stored removal means clearing the incoming value and keeping only +// the removal's timestamp. Only the two lock keys and their reserved timestamps +// move; a non-replica write never sets the flag that invokes this. +func reconcileStoredObjectLock(metadata map[string]string, stored objectLockState) { + incoming := storedObjectLockState(metadata) + + incomingRetentionTS, _ := time.Parse(time.RFC3339Nano, incoming.retentionTimestamp) + if !stored.retentionIsOlderThan(incomingRetentionTS) { + // The stored retention is at least as new as the incoming one (or the + // incoming update is unordered): drop the incoming value and put the stored + // state back, which may itself be a removal (value keys absent, timestamp + // present). + delete(metadata, strings.ToLower(xhttp.AmzObjectLockMode)) + delete(metadata, strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)) + delete(metadata, ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp) + stored.restoreRetention(metadata) + } + + incomingLegalHoldTS, _ := time.Parse(time.RFC3339Nano, incoming.legalHoldTimestamp) + if incoming.legalHold == "" || !stored.legalHoldIsOlderThan(incomingLegalHoldTS) { + delete(metadata, strings.ToLower(xhttp.AmzObjectLockLegalHold)) + delete(metadata, ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp) + stored.restoreLegalHold(metadata) + } +} diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go index e78ab2edd..1b42e970c 100644 --- a/cmd/bucket-replication.go +++ b/cmd/bucket-replication.go @@ -877,8 +877,8 @@ func putReplicationOpts(ctx context.Context, sc string, objInfo ObjectInfo) (put if hasMode { putOpts.Mode = minio.RetentionMode(mode) } - // 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. + // A removed retention is stored as an empty or absent 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 { @@ -886,10 +886,14 @@ func putReplicationOpts(ctx context.Context, sc string, objInfo ObjectInfo) (put } putOpts.RetainUntilDate = rdate } - if hasMode || hasRetainDate { - // set retention timestamp in opts + retainTmstampStr, hasRetainTmstamp := objInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] + if hasMode || hasRetainDate || hasRetainTmstamp { + // Send the ordering timestamp whenever the version carries one, even for a + // removal whose value keys are absent (the shape a retransmit PUT leaves), + // so the next hop can order the removal instead of keeping obsolete + // retention. retTimestamp := objInfo.ModTime - if retainTmstampStr, ok := objInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp]; ok { + if hasRetainTmstamp { var err error retTimestamp, err = time.Parse(time.RFC3339Nano, retainTmstampStr) if err != nil { @@ -1618,8 +1622,11 @@ applyAction: } else { putOpts, isMP, err := putReplicationOpts(ctx, tgt.StorageClass, objInfo) if err != nil { - rinfo.Err = err + // rinfo was primed Completed above; a failure to build the write + // options means nothing reached the target, so mark it Failed and + // carry the error instead of reporting a phantom success. rinfo.ReplicationStatus = replication.Failed + rinfo.Err = err replLogIf(ctx, fmt.Errorf("failed to set replicate options for object %s/%s(%s) (target %s) err:%w", bucket, objInfo.Name, objInfo.VersionID, tgt.EndpointURL(), err)) sendEvent(eventArgs{ EventName: event.ObjectReplicationNotTracked, diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go index 35c790805..c50c44b0f 100644 --- a/cmd/erasure-multipart.go +++ b/cmd/erasure-multipart.go @@ -1114,7 +1114,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str auditObjectErasureSet(ctx, "CompleteMultipartUpload", object, &er) } - if opts.CheckPrecondFn != nil { + if opts.CheckPrecondFn != nil || opts.ReplicaLockReconcile { if !opts.NoLock { ns := er.NewNSLock(bucket, object) lkctx, err := ns.GetLock(ctx, globalOperationTimeout) @@ -1126,18 +1126,24 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str opts.NoLock = true } - obj, err := er.getObjectInfo(ctx, bucket, object, opts) - if err == nil && opts.CheckPrecondFn(obj) { - return ObjectInfo{}, PreConditionFailed{} - } - if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { - return ObjectInfo{}, err - } + // The Object Lock reconcile below needs the version being committed, read + // after checkUploadIDExists, so only the precondition read happens here; + // both run under this same write lock, held until the version is renamed + // into place. + if opts.CheckPrecondFn != nil { + obj, err := er.getObjectInfo(ctx, bucket, object, opts) + if err == nil && opts.CheckPrecondFn(obj) { + return ObjectInfo{}, PreConditionFailed{} + } + if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { + return ObjectInfo{}, err + } - // if object doesn't exist return error for If-Match conditional requests - // If-None-Match should be allowed to proceed for non-existent objects - if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) { - return ObjectInfo{}, err + // if object doesn't exist return error for If-Match conditional requests + // If-None-Match should be allowed to proceed for non-existent objects + if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) { + return ObjectInfo{}, err + } } } @@ -1149,6 +1155,42 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str return oi, toObjectErr(err, bucket, object, uploadID) } + // A trusted SSE-C replica completion re-orders the Object Lock carried in the + // upload metadata against the version it is about to replace, read on this + // erasure set under the write lock held above, so a hold or retention that + // reached the version after this upload was initiated is not rolled back at + // completion (issue #120). Scoped to SSE-C uploads, the only ones this issue + // routes through completion. + // + // Scope: correct for a single erasure set. A multi-pool deployment (duplicate + // versions across pools, ModTime ties, cross-pool lock authority) is out of + // scope and tracked in pgsty/silo#TBD-multipool-lock. + if opts.ReplicaLockReconcile && crypto.SSEC.IsEncrypted(fi.Metadata) { + // A persisted upload records the null version as an empty VersionID; look + // it up as the null version so the reconcile reads the addressed version's + // stored lock, not the latest version's. + lookupVersionID := fi.VersionID + if lookupVersionID == "" { + lookupVersionID = nullVersionID + } + curr, gerr := er.getObjectInfo(ctx, bucket, object, ObjectOptions{ + VersionID: lookupVersionID, + Versioned: opts.Versioned, + VersionSuspended: opts.VersionSuspended, + NoLock: true, + }) + switch { + case gerr == nil: + reconcileStoredObjectLock(fi.Metadata, storedObjectLockState(curr.UserDefined)) + case isErrVersionNotFound(gerr) || isErrObjectNotFound(gerr): + // No existing version to order against: keep the upload's own accepted + // lock, including a pre-upgrade upload that persisted values without + // their ordering timestamps. + default: + return oi, toObjectErr(gerr, bucket, object) + } + } + uploadIDPath := er.getUploadIDDir(bucket, object, uploadID) onlineDisks := er.getDisks() writeQuorum := fi.WriteQuorum(er.defaultWQuorum()) diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index db5e77ed2..a6293edc2 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -1268,7 +1268,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st data := r.Reader - if opts.CheckPrecondFn != nil { + if opts.CheckPrecondFn != nil || opts.ReplicaLockReconcile { if !opts.NoLock { ns := er.NewNSLock(bucket, object) lkctx, err := ns.GetLock(ctx, globalOperationTimeout) @@ -1281,17 +1281,33 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st } obj, err := er.getObjectInfo(ctx, bucket, object, opts) - if err == nil && opts.CheckPrecondFn(obj) { - return objInfo, PreConditionFailed{} - } + // A destination read that fails for a reason other than not-found must not + // be taken as a passed precondition or as absent lock state. if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { return objInfo, err } + if opts.CheckPrecondFn != nil { + if err == nil && opts.CheckPrecondFn(obj) { + return objInfo, PreConditionFailed{} + } + // if object doesn't exist return error for If-Match conditional requests + // If-None-Match should be allowed to proceed for non-existent objects + if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) { + return objInfo, err + } + } - // if object doesn't exist return error for If-Match conditional requests - // If-None-Match should be allowed to proceed for non-existent objects - if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) { - return objInfo, err + // Order this trusted SSE-C replica's Object Lock against the addressed + // version's stored state, read on this erasure set under the write lock, + // so a value that lost the ordering cannot overwrite a newer one committed + // after the handler decided (issue #120). Only reconcile against an + // existing version; on not-found the write's own accepted lock is kept. + // + // Scope: correct for a single erasure set. A multi-pool deployment + // (duplicate versions across pools, ModTime ties, cross-pool lock + // authority) is out of scope and tracked in pgsty/silo#TBD-multipool-lock. + if opts.ReplicaLockReconcile && err == nil { + reconcileStoredObjectLock(opts.UserDefined, storedObjectLockState(obj.UserDefined)) } } diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go index f8664310d..2f64ded4f 100644 --- a/cmd/object-api-interface.go +++ b/cmd/object-api-interface.go @@ -99,6 +99,7 @@ type ObjectOptions struct { ReplicationSourceTaggingTimestamp time.Time // set if MinIOSourceTaggingTimestamp received ReplicationSourceLegalholdTimestamp time.Time // set if MinIOSourceObjectLegalholdTimestamp received ReplicationSourceRetentionTimestamp time.Time // set if MinIOSourceObjectRetentionTimestamp received + ReplicaLockReconcile bool // set for a trusted SSE-C replica full write/completion: re-order Object Lock against the destination version read under the write lock (single erasure set; see pgsty/silo#TBD-multipool-lock) DeletePrefix bool // set true to enforce a prefix deletion, only application for DeleteObject API, DeletePrefixObject bool // set true when object's erasure set is resolvable by object name (using getHashedSetIndex) diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go index fcb834821..789d0fe2a 100644 --- a/cmd/object-handlers.go +++ b/cmd/object-handlers.go @@ -1771,50 +1771,9 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re // apply default bucket configuration/governance headers for dest side. retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, dstBucket, dstObject, getObjectInfo, retPerms, holdPerms, replicaTrusted) if s3Err == ErrNone { - // A replica update is ordered by its source timestamp alone, whether or - // not it still carries a value: an update newer than the stored state - // applies, and a removal is just an update with no value. An update that - // is stale, or that carries no source timestamp at all and is therefore - // unordered, leaves the stored state in place instead of erasing it. - switch { - case !dstOpts.ReplicationRequest: - if retentionMode.Valid() { - srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode) - srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC()) - srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = UTCNow().Format(time.RFC3339Nano) - } - case !replicaTrusted && !retentionMode.Valid(): - // A trusted peer that did not mark this request as a replica sends - // no replicated state, so a missing value carries no instruction and - // the rebuilt metadata is left as it is. - case !storedLock.retentionIsOlderThan(dstOpts.ReplicationSourceRetentionTimestamp): - storedLock.restoreRetention(srcInfo.UserDefined) - default: - if retentionMode.Valid() { - srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode) - srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC()) - } - srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = dstOpts.ReplicationSourceRetentionTimestamp.UTC().Format(time.RFC3339Nano) - } - - // Legal hold has no removal in S3: an explicitly empty header is already - // rejected as an invalid status, so the only value-less shape that gets - // here is an absent one, and that conveys no legal-hold change even when - // an orphaned timestamp comes with it. Only a valid status can win. - switch { - case !dstOpts.ReplicationRequest: - if legalHold.Status.Valid() { - srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status) - srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = UTCNow().Format(time.RFC3339Nano) - } - case !replicaTrusted && !legalHold.Status.Valid(): - // As above: a marker-only request carries no legal-hold update. - case legalHold.Status.Valid() && storedLock.legalHoldIsOlderThan(dstOpts.ReplicationSourceLegalholdTimestamp): - srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status) - srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = dstOpts.ReplicationSourceLegalholdTimestamp.UTC().Format(time.RFC3339Nano) - default: - storedLock.restoreLegalHold(srcInfo.UserDefined) - } + applyReplicatedObjectLock(srcInfo.UserDefined, storedLock, replicaTrusted, + retentionMode, retentionDate, legalHold, + dstOpts.ReplicationSourceRetentionTimestamp, dstOpts.ReplicationSourceLegalholdTimestamp) if replicaTrusted { // An SSE-C key rotation snapshots every stored reserved key into @@ -2282,12 +2241,31 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req getObjectInfo := objectAPI.GetObjectInfo retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms, isReplicaTrusted(ctx)) - if s3Err == ErrNone && retentionMode.Valid() { - metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode) - metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC()) - } - if s3Err == ErrNone && legalHold.Status.Valid() { - metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status) + if s3Err == ErrNone { + // A trusted replica write addressing a specific version can be a full + // retransmit over an existing version whose lock state is newer than the + // source snapshot (issue #120). Order the incoming update against what is + // stored so a stale value cannot overwrite it; a non-replica write (and a + // marker-only peer write) has no stored state to order against and takes + // the helper's ordinary-write branch. + var storedLock objectLockState + if isReplicaTrusted(ctx) && opts.VersionID != "" { + var lerr error + if storedLock, lerr = replicaStoredLock(ctx, getObjectInfo, bucket, object, opts.VersionID); lerr != nil { + writeErrorResponse(ctx, w, toAPIError(ctx, lerr), r.URL) + return + } + } + applyReplicatedObjectLock(metadata, storedLock, isReplicaTrusted(ctx), + retentionMode, retentionDate, legalHold, + opts.ReplicationSourceRetentionTimestamp, opts.ReplicationSourceLegalholdTimestamp) + // The decision above orders against the version as read here; let the + // object layer re-run it against the version read under the write lock + // that guards the replacement, so a newer hold or retention committed in + // between is not rolled back (issue #120). Scoped to the SSE-C replica + // retransmit this issue enables, keyed on the incoming write's restored + // SSE-C seal, the same predicate as the duplicate-version exemption. + opts.ReplicaLockReconcile = isReplicaTrusted(ctx) && opts.VersionID != "" && crypto.SSEC.IsEncrypted(metadata) } if s3Err != ErrNone { writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL) diff --git a/cmd/object-multipart-handlers.go b/cmd/object-multipart-handlers.go index 22d2839f1..983644223 100644 --- a/cmd/object-multipart-handlers.go +++ b/cmd/object-multipart-handlers.go @@ -34,7 +34,6 @@ import ( "github.com/minio/minio-go/v7" "github.com/minio/minio-go/v7/pkg/encrypt" "github.com/minio/minio-go/v7/pkg/tags" - "github.com/minio/minio/internal/amztime" sse "github.com/minio/minio/internal/bucket/encryption" objectlock "github.com/minio/minio/internal/bucket/object/lock" "github.com/minio/minio/internal/bucket/replication" @@ -260,12 +259,32 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r getObjectInfo := objectAPI.GetObjectInfo retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms, replicaTrusted) - if s3Err == ErrNone && retentionMode.Valid() { - metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode) - metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC()) - } - if s3Err == ErrNone && legalHold.Status.Valid() { - metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status) + if s3Err == ErrNone { + // A trusted replica NewMultipartUpload addressing a specific version can + // be a full retransmit over an existing version whose lock state is newer + // than the source snapshot (issue #120). Order the incoming update against + // what is stored so a stale value cannot overwrite it. opts is built below + // (its ServerSideEncryption depends on the encMetadata merge that has not + // happened yet), so read the replica ordering inputs the way + // putOptsFromHeaders will; a malformed timestamp fails the request when + // opts is built, so a parse error here is left as a zero time. + var ( + storedLock objectLockState + srcRetentionTS, srcLegalholdTS time.Time + ) + if replicaTrusted { + srcRetentionTS, _ = time.Parse(time.RFC3339, strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceObjectRetentionTimestamp))) + srcLegalholdTS, _ = time.Parse(time.RFC3339, strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceObjectLegalHoldTimestamp))) + if versionID := strings.TrimSpace(r.Form.Get(xhttp.VersionID)); versionID != "" { + var lerr error + if storedLock, lerr = replicaStoredLock(ctx, getObjectInfo, bucket, object, versionID); lerr != nil { + writeErrorResponse(ctx, w, toAPIError(ctx, lerr), r.URL) + return + } + } + } + applyReplicatedObjectLock(metadata, storedLock, replicaTrusted, + retentionMode, retentionDate, legalHold, srcRetentionTS, srcLegalholdTS) } if s3Err != ErrNone { writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL) @@ -1166,6 +1185,14 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite } opts.Versioned = versioned opts.VersionSuspended = suspended + // A replicated multipart completion carries the internal replication marker + // (the sender does not re-assert REPLICA status on Complete, so this is keyed + // on trusted replication, matching completeMultipartOpts). The object layer + // re-orders the Object Lock it carries against the destination version read + // under the write lock, but only for an SSE-C upload -- the scope this issue + // enables -- so a marker-only non-SSE-C completion keeps ordinary write + // semantics (issue #120). + opts.ReplicaLockReconcile = trustedReplication // First, we compute the ETag of the multipart object. // The ETag of a multi-part object is always: diff --git a/cmd/replication-ssec-retransmit_test.go b/cmd/replication-ssec-retransmit_test.go index e894b8fa0..bbb8712b9 100644 --- a/cmd/replication-ssec-retransmit_test.go +++ b/cmd/replication-ssec-retransmit_test.go @@ -12,6 +12,7 @@ package cmd import ( "bytes" + "context" "crypto/md5" "encoding/base64" "encoding/xml" @@ -792,3 +793,1002 @@ func testAPISSECReplicaWriteExemptionIsKeyedOnTheIncomingWrite(obj ObjectLayer, } }) } + +// Retain-until values in the millisecond form ISO8601Format round-trips to, so +// a value applied through the handler reads back byte-for-byte. +const ( + retransmitRetainUntilNewer = "2031-01-01T00:00:00.000Z" + retransmitRetainUntilStale = "2028-01-01T00:00:00.000Z" +) + +// TestAPISSECReplicaRetransmitObjectLockOrdering proves that the full SSE-C +// replica retransmit orders the Object Lock update it carries against the state +// already stored on the addressed version, the same way the metadata CopyObject +// path does. Before issue #120 routed these writes through PutObjectHandler the +// handler applied the incoming retention and legal hold directly and never +// persisted the ordering timestamps, so a retransmit carrying an older value +// could overwrite a destination version's newer one. See pgsty/silo#120. +func TestAPISSECReplicaRetransmitObjectLockOrdering(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testAPISSECReplicaRetransmitObjectLockOrdering, + makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + }) +} + +func testAPISSECReplicaRetransmitObjectLockOrdering(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{0x45}, 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[:]), + } + + // seed writes a fresh SSE-C source version with no Object Lock state and + // returns its version id. + seed := func(t *testing.T, object string) string { + t.Helper() + putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, + bytes.Repeat([]byte("ssec-lock-ordering-"), 64), sseHeaders) + info, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + return info.VersionID + } + + // retransmit sends the full SSE-C replica retransmit the sender emits for the + // addressed version, over its raw ciphertext, carrying exactly the given + // Object Lock headers on top of the replica seal. The credential is the admin + // user so the retention and legal-hold permission checks pass; the request is + // still a trusted replica because it carries the marker and REPLICA status. + retransmit := func(t *testing.T, object, versionID string, lockHeaders map[string]string) { + t.Helper() + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, + VersionID: versionID, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + cipher, err := io.ReadAll(gr) + gr.Close() + if err != nil { + t.Fatal(err) + } + opts, _, err := putReplicationOpts(t.Context(), "", srcInfo) + if err != nil { + t.Fatal(err) + } + opts.Internal.SourceMTime = time.Time{} + hdrs := make(map[string]string) + for name, values := range opts.Header() { + if len(values) > 0 { + hdrs[name] = values[0] + } + } + hdrs[xhttp.MinIOSourceReplicationRequest] = "true" + hdrs[xhttp.AmzBucketReplicationStatus] = "REPLICA" + hdrs[xhttp.MinIOSourceETag] = srcInfo.ETag + // The case owns the lock instruction: drop any lock header the sender + // derived from the source version. + for _, name := range []string{ + xhttp.AmzObjectLockMode, xhttp.AmzObjectLockRetainUntilDate, xhttp.AmzObjectLockLegalHold, + xhttp.MinIOSourceObjectRetentionTimestamp, xhttp.MinIOSourceObjectLegalHoldTimestamp, + } { + delete(hdrs, name) + } + maps.Copy(hdrs, lockHeaders) + + req, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectURL("", bucketName, object)+"?versionId="+versionID, int64(len(cipher)), + bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, hdrs) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("%s: replica retransmit PUT status %d, want 200: %s", instanceType, rec.Code, rec.Body.String()) + } + } + + t.Run("older-legal-hold-off-does-not-clear-newer-on", func(t *testing.T) { + object := "ssec-lock-ordering/legal-hold" + versionID := seed(t, object) + + // Establish the newer legal hold ON. Its timestamp persistence is proven + // by the retention sibling test, so here only require the value took + // effect before the older OFF arrives, so the clobber below is what tells + // a fixed handler from a broken one. + retransmit(t, object, versionID, map[string]string{ + xhttp.AmzObjectLockLegalHold: "ON", + xhttp.MinIOSourceObjectLegalHoldTimestamp: objectLockTestStamp1000, + }) + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got.legalHold != "ON" { + t.Fatalf("%s: seeding legal hold ON failed: got %+v", instanceType, got) + } + + // A retransmit carrying an older legal-hold OFF must not clear it. + retransmit(t, object, versionID, map[string]string{ + xhttp.AmzObjectLockLegalHold: "OFF", + xhttp.MinIOSourceObjectLegalHoldTimestamp: objectLockTestStamp0900, + }) + want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000} + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: an older legal-hold OFF cleared the newer ON: got %+v, want %+v", instanceType, got, want) + } + }) + + t.Run("newer-retention-applies-and-persists-its-timestamp", func(t *testing.T) { + object := "ssec-lock-ordering/retention-applies" + versionID := seed(t, object) + + retransmit(t, object, versionID, map[string]string{ + xhttp.AmzObjectLockMode: "GOVERNANCE", + xhttp.AmzObjectLockRetainUntilDate: retransmitRetainUntilNewer, + xhttp.MinIOSourceObjectRetentionTimestamp: objectLockTestStamp1000, + }) + // The value applies and, crucially, the ordering timestamp is persisted so + // a later stale update can be recognized as older. + want := objectLockFields{ + mode: "GOVERNANCE", retainUntil: retransmitRetainUntilNewer, retentionStamp: objectLockTestStamp1000, + } + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: newer retention did not apply and persist its timestamp: got %+v, want %+v", instanceType, got, want) + } + }) + + t.Run("stale-retention-update-is-ignored", func(t *testing.T) { + object := "ssec-lock-ordering/retention-stale" + versionID := seed(t, object) + + // Establish the newer retention first; its timestamp persistence is proven + // by the sibling test above, so here only require the value took effect, so + // the stale overwrite below is what tells a fixed handler from a broken one. + retransmit(t, object, versionID, map[string]string{ + xhttp.AmzObjectLockMode: "GOVERNANCE", + xhttp.AmzObjectLockRetainUntilDate: retransmitRetainUntilNewer, + xhttp.MinIOSourceObjectRetentionTimestamp: objectLockTestStamp1000, + }) + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got.mode != "GOVERNANCE" || got.retainUntil != retransmitRetainUntilNewer { + t.Fatalf("%s: seeding the newer retention failed: got %+v", instanceType, got) + } + + // A retransmit carrying an older retention with a different date must be + // ignored; the newer date and its ordering timestamp survive. + retransmit(t, object, versionID, map[string]string{ + xhttp.AmzObjectLockMode: "GOVERNANCE", + xhttp.AmzObjectLockRetainUntilDate: retransmitRetainUntilStale, + xhttp.MinIOSourceObjectRetentionTimestamp: objectLockTestStamp0900, + }) + want := objectLockFields{ + mode: "GOVERNANCE", retainUntil: retransmitRetainUntilNewer, retentionStamp: objectLockTestStamp1000, + } + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: a stale retention update overwrote the newer one: got %+v, want %+v", instanceType, got, want) + } + }) +} + +// TestAPISSECReplicaRetransmitMultipartObjectLockOrdering proves the ordering +// fix also covers the multipart initiation path #120 routes an SSE-C replica +// through: NewMultipartUploadHandler reads the addressed version's stored lock +// state and orders the incoming update against it, so a large-object retransmit +// carrying an older legal-hold OFF cannot clear a newer ON. See pgsty/silo#120. +func TestAPISSECReplicaRetransmitMultipartObjectLockOrdering(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testAPISSECReplicaRetransmitMultipartObjectLockOrdering, + makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + }) +} + +func testAPISSECReplicaRetransmitMultipartObjectLockOrdering(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{0x46}, 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[:]), + } + + // replicaSealHeaders returns the sender's replica seal and marker headers for + // the addressed version, with every Object Lock header stripped so the case + // owns the lock instruction. + replicaSealHeaders := func(t *testing.T, srcInfo ObjectInfo) map[string]string { + t.Helper() + opts, _, err := putReplicationOpts(t.Context(), "", srcInfo) + if err != nil { + t.Fatal(err) + } + opts.Internal.SourceMTime = time.Time{} + hdrs := make(map[string]string) + for name, values := range opts.Header() { + if len(values) > 0 { + hdrs[name] = values[0] + } + } + hdrs[xhttp.MinIOSourceReplicationRequest] = "true" + hdrs[xhttp.AmzBucketReplicationStatus] = "REPLICA" + hdrs[xhttp.MinIOSourceETag] = srcInfo.ETag + for _, name := range []string{ + xhttp.AmzObjectLockMode, xhttp.AmzObjectLockRetainUntilDate, xhttp.AmzObjectLockLegalHold, + xhttp.MinIOSourceObjectRetentionTimestamp, xhttp.MinIOSourceObjectLegalHoldTimestamp, + } { + delete(hdrs, name) + } + return hdrs + } + + object := "ssec-lock-ordering/multipart" + // A single-part SSE-C source is enough; the retransmit re-uploads its bytes + // as one multipart part and commits the addressed version. + putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, + bytes.Repeat([]byte("ssec-multipart-lock-ordering-"), 64), sseHeaders) + base, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + versionID := base.VersionID + + // Establish the newer legal hold ON through the single-part PUT retransmit. + { + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, VersionID: versionID, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + cipher, rerr := io.ReadAll(gr) + gr.Close() + if rerr != nil { + t.Fatal(rerr) + } + hdrs := replicaSealHeaders(t, srcInfo) + hdrs[xhttp.AmzObjectLockLegalHold] = "ON" + hdrs[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp1000 + req, rerr := newTestSignedRequestV4(http.MethodPut, + getPutObjectURL("", bucketName, object)+"?versionId="+versionID, int64(len(cipher)), + bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, hdrs) + if rerr != nil { + t.Fatal(rerr) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("%s: seeding legal hold ON via PUT status %d: %s", instanceType, rec.Code, rec.Body.String()) + } + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got.legalHold != "ON" { + t.Fatalf("%s: seeding legal hold ON failed: got %+v", instanceType, got) + } + } + + // A multipart retransmit carrying an older legal-hold OFF must not clear it. + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, VersionID: versionID, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + cipher, err := io.ReadAll(gr) + gr.Close() + if err != nil { + t.Fatal(err) + } + actualSize, err := srcInfo.GetActualSize() + if err != nil { + t.Fatal(err) + } + + newHdrs := replicaSealHeaders(t, srcInfo) + newHdrs[xhttp.AmzObjectLockLegalHold] = "OFF" + newHdrs[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp0900 + newReq, err := newTestSignedRequestV4(http.MethodPost, + getNewMultipartURL("", bucketName, object)+"&versionId="+versionID, 0, nil, + credentials.AccessKey, credentials.SecretKey, newHdrs) + if err != nil { + t.Fatal(err) + } + newRec := httptest.NewRecorder() + apiRouter.ServeHTTP(newRec, newReq) + if newRec.Code != http.StatusOK { + t.Fatalf("%s: replica NewMultipartUpload status %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(cipher)), + bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, + map[string]string{xhttp.MinIOSourceReplicationRequest: "true"}) + if err != nil { + t.Fatal(err) + } + partRec := httptest.NewRecorder() + apiRouter.ServeHTTP(partRec, partReq) + if partRec.Code != http.StatusOK { + t.Fatalf("%s: replica PutObjectPart status %d: %s", instanceType, partRec.Code, partRec.Body.String()) + } + + completeBody, 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(completeBody)), + bytes.NewReader(completeBody), credentials.AccessKey, credentials.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) + } + completeRec := httptest.NewRecorder() + apiRouter.ServeHTTP(completeRec, completeReq) + if completeRec.Code != http.StatusOK { + t.Fatalf("%s: replica CompleteMultipartUpload status %d: %s", instanceType, completeRec.Code, completeRec.Body.String()) + } + + want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000} + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: an older legal-hold OFF via multipart cleared the newer ON: got %+v, want %+v", instanceType, got, want) + } +} + +// TestReplicaStoredLock verifies how a replica write reads the destination +// version's stored lock state before ordering its update: a present version +// yields its state, a missing object or version yields an empty state so a first +// write is not blocked, and any other read error (a quorum loss, a timeout) is +// returned so the caller fails the write instead of ordering an incoming update +// against lock state it merely could not read. See pgsty/silo#120. +func TestReplicaStoredLock(t *testing.T) { + fixed := func(oi ObjectInfo, err error) GetObjectInfoFn { + return func(context.Context, string, string, ObjectOptions) (ObjectInfo, error) { return oi, err } + } + stored := ObjectInfo{UserDefined: map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000, + }} + + t.Run("present-version-returns-its-state", func(t *testing.T) { + got, err := replicaStoredLock(context.Background(), fixed(stored, nil), "b", "o", "v") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got.legalHold != "ON" || got.legalHoldTimestamp != objectLockTestStamp1000 { + t.Fatalf("stored lock state = %+v, want legal hold ON stamped %q", got, objectLockTestStamp1000) + } + }) + + for name, notFound := range map[string]error{ + "object-not-found": ObjectNotFound{Bucket: "b", Object: "o"}, + "version-not-found": VersionNotFound{Bucket: "b", Object: "o", VersionID: "v"}, + } { + t.Run(name+"-is-empty-state", func(t *testing.T) { + got, err := replicaStoredLock(context.Background(), fixed(ObjectInfo{}, notFound), "b", "o", "v") + if err != nil { + t.Fatalf("a not-found read must not error: %v", err) + } + if got != (objectLockState{}) { + t.Fatalf("a not-found read must yield empty state, got %+v", got) + } + }) + } + + t.Run("transient-read-error-propagates", func(t *testing.T) { + boom := InsufficientReadQuorum{} + got, err := replicaStoredLock(context.Background(), fixed(ObjectInfo{}, boom), "b", "o", "v") + if err == nil { + t.Fatal("a transient read error must propagate, not be treated as absent lock state") + } + if isErrObjectNotFound(err) || isErrVersionNotFound(err) { + t.Fatalf("transient error misclassified as not-found: %v", err) + } + if got != (objectLockState{}) { + t.Fatalf("on a read error the caller must get empty state and fail the write, got %+v", got) + } + }) +} + +// TestPutReplicationOptsRetentionRemovalTimestampOnly asserts that a version +// whose retention was removed on the retransmit PUT path -- stored as an +// ordering timestamp with the value keys absent, not empty -- still builds +// replication options that carry the removal timestamp, so the next hop can +// order the removal instead of keeping obsolete retention. See pgsty/silo#120. +func TestPutReplicationOptsRetentionRemovalTimestampOnly(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{ + // Only the reserved ordering timestamp; no mode/date keys at all. + ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: removedAt.Format(time.RFC3339Nano), + }, + } + opts, _, err := putReplicationOpts(t.Context(), "", oi) + if err != nil { + t.Fatalf("putReplicationOpts on a timestamp-only removal: %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) + } +} + +// TestAPIReplicaMarkerOnlyAppliesObjectLock guards the regression the shared +// ordering helper could introduce: a trusted peer write that carries the +// internal replication marker but NOT REPLICA status (replicationRequest true, +// replicaTrusted false) must keep ordinary write semantics and apply its +// validated Object Lock, not restore an empty stored state. It covers an +// explicit legal hold and a bucket-default retention, on both the PUT and the +// multipart-initiation paths. See pgsty/silo#120 (Codex finding 3). +func TestAPIReplicaMarkerOnlyAppliesObjectLock(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testAPIReplicaMarkerOnlyAppliesObjectLock, + makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + }) +} + +func testAPIReplicaMarkerOnlyAppliesObjectLock(obj ObjectLayer, instanceType, bucketName string, + apiRouter http.Handler, credentials auth.Credentials, t *testing.T, +) { + // A trusted peer credential holding ReplicateObject plus the lock permissions + // checkPutObjectLockAllowed enforces even for a marker-only write. + peer := newObjectAttributesAuthzUser(t, instanceType, bucketName, + `"s3:GetObject","s3:PutObject","s3:ReplicateObject","s3:PutObjectRetention","s3:PutObjectLegalHold"`) + + // Bucket default retention, so a marker-only write with no lock headers still + // has a validated retention to apply. + lockCfg := []byte(`EnabledGOVERNANCE30`) + if _, err := globalBucketMetadataSys.Update(t.Context(), bucketName, objectLockConfig, lockCfg); err != nil { + t.Fatalf("%s: configure bucket default retention: %v", instanceType, err) + } + + // markerOnly returns the trusted-but-not-REPLICA header set plus extra: the + // internal marker with no REPLICA status. + markerOnly := func(extra map[string]string) map[string]string { + h := map[string]string{xhttp.MinIOSourceReplicationRequest: "true"} + maps.Copy(h, extra) + return h + } + data := bytes.Repeat([]byte("marker-only-lock-"), 32) + + markerOnlyMPU := func(t *testing.T, object string, lockHeaders map[string]string) { + t.Helper() + newReq, err := newTestSignedRequestV4(http.MethodPost, getNewMultipartURL("", bucketName, object), 0, nil, + peer.AccessKey, peer.SecretKey, markerOnly(lockHeaders)) + if err != nil { + t.Fatal(err) + } + newRec := httptest.NewRecorder() + apiRouter.ServeHTTP(newRec, newReq) + if newRec.Code != http.StatusOK { + t.Fatalf("%s: marker-only NewMultipartUpload %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), peer.AccessKey, peer.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + partRec := httptest.NewRecorder() + apiRouter.ServeHTTP(partRec, partReq) + if partRec.Code != http.StatusOK { + t.Fatalf("%s: marker-only PutObjectPart %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), peer.AccessKey, peer.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + completeRec := httptest.NewRecorder() + apiRouter.ServeHTTP(completeRec, completeReq) + if completeRec.Code != http.StatusOK { + t.Fatalf("%s: marker-only CompleteMultipartUpload %d: %s", instanceType, completeRec.Code, completeRec.Body.String()) + } + } + markerOnlyPUT := func(t *testing.T, object string, lockHeaders map[string]string) { + t.Helper() + req, err := newTestSignedRequestV4(http.MethodPut, getPutObjectURL("", bucketName, object), int64(len(data)), + bytes.NewReader(data), peer.AccessKey, peer.SecretKey, markerOnly(lockHeaders)) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("%s: marker-only PUT %d: %s", instanceType, rec.Code, rec.Body.String()) + } + } + lockOf := func(t *testing.T, object string) objectLockState { + t.Helper() + info, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + return storedObjectLockState(info.UserDefined) + } + + // An explicit legal hold on a marker-only write must be applied, not dropped. + // A legal-hold request suppresses the bucket default retention, so the version + // carries only the hold. + explicitHold := map[string]string{xhttp.AmzObjectLockLegalHold: "ON"} + + t.Run("put-explicit-legal-hold", func(t *testing.T) { + object := "marker-only/put-legal-hold" + markerOnlyPUT(t, object, explicitHold) + if got := lockOf(t, object); got.legalHold != "ON" { + t.Fatalf("%s: marker-only PUT dropped the explicit legal hold: got %+v", instanceType, got) + } + }) + t.Run("mpu-explicit-legal-hold", func(t *testing.T) { + object := "marker-only/mpu-legal-hold" + markerOnlyMPU(t, object, explicitHold) + if got := lockOf(t, object); got.legalHold != "ON" { + t.Fatalf("%s: marker-only multipart dropped the explicit legal hold: got %+v", instanceType, got) + } + }) + t.Run("put-bucket-default-retention", func(t *testing.T) { + object := "marker-only/put-default-retention" + markerOnlyPUT(t, object, nil) + if got := lockOf(t, object); got.mode != "GOVERNANCE" || got.retainUntil == "" { + t.Fatalf("%s: marker-only PUT dropped the bucket default retention: got %+v", instanceType, got) + } + }) + t.Run("mpu-bucket-default-retention", func(t *testing.T) { + object := "marker-only/mpu-default-retention" + markerOnlyMPU(t, object, nil) + if got := lockOf(t, object); got.mode != "GOVERNANCE" || got.retainUntil == "" { + t.Fatalf("%s: marker-only multipart dropped the bucket default retention: got %+v", instanceType, got) + } + }) +} + +// TestAPIReplicaMultipartNewerHoldSurvivesCompletion verifies that a legal hold +// that reaches a destination version AFTER a replica multipart upload was +// initiated is not rolled back when that upload completes. The initiation +// resolves the lock against the version as it then stands, but completion +// re-orders the carried lock against the version read under the namespace write +// lock that guards the replacement, so a newer hold (with its newer timestamp) +// survives. See pgsty/silo#120 (Codex finding 1). +func TestAPIReplicaMultipartNewerHoldSurvivesCompletion(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testAPIReplicaMultipartNewerHoldSurvivesCompletion, + makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + }) +} + +func testAPIReplicaMultipartNewerHoldSurvivesCompletion(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{0x47}, 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[:]), + } + object := "mpu-lock-boundary/obj" + + // seal returns the sender's SSE-C replica seal and marker headers for the + // addressed version, with every Object Lock header stripped so the case owns + // the lock instruction. + seal := func(t *testing.T, srcInfo ObjectInfo) map[string]string { + t.Helper() + opts, _, err := putReplicationOpts(t.Context(), "", srcInfo) + if err != nil { + t.Fatal(err) + } + opts.Internal.SourceMTime = time.Time{} + hdrs := make(map[string]string) + for name, values := range opts.Header() { + if len(values) > 0 { + hdrs[name] = values[0] + } + } + hdrs[xhttp.MinIOSourceReplicationRequest] = "true" + hdrs[xhttp.AmzBucketReplicationStatus] = "REPLICA" + hdrs[xhttp.MinIOSourceETag] = srcInfo.ETag + for _, name := range []string{ + xhttp.AmzObjectLockMode, xhttp.AmzObjectLockRetainUntilDate, xhttp.AmzObjectLockLegalHold, + xhttp.MinIOSourceObjectRetentionTimestamp, xhttp.MinIOSourceObjectLegalHoldTimestamp, + } { + delete(hdrs, name) + } + return hdrs + } + rawOf := func(t *testing.T, versionID string) (ObjectInfo, []byte) { + t.Helper() + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, VersionID: versionID, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + cipher, rerr := io.ReadAll(gr) + gr.Close() + if rerr != nil { + t.Fatal(rerr) + } + return srcInfo, cipher + } + + // A destination SSE-C version with no lock. + putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, + bytes.Repeat([]byte("mpu-lock-boundary-"), 64), sseHeaders) + base, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + versionID := base.VersionID + srcInfo, cipher := rawOf(t, versionID) + + // Initiate an SSE-C replica multipart carrying legal hold OFF stamped 09:00. + // The destination has no lock yet, so the decision made now is OFF@09:00. + newHdrs := seal(t, srcInfo) + newHdrs[xhttp.AmzObjectLockLegalHold] = "OFF" + newHdrs[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp0900 + newReq, err := newTestSignedRequestV4(http.MethodPost, + getNewMultipartURL("", bucketName, object)+"&versionId="+versionID, 0, nil, + credentials.AccessKey, credentials.SecretKey, newHdrs) + if err != nil { + t.Fatal(err) + } + newRec := httptest.NewRecorder() + apiRouter.ServeHTTP(newRec, newReq) + if newRec.Code != http.StatusOK { + t.Fatalf("%s: replica NewMultipartUpload %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) + } + + // After initiation, a newer legal hold ON@10:00 lands on the same version + // through an independent SSE-C replica PUT retransmit. + putHdrs := seal(t, srcInfo) + putHdrs[xhttp.AmzObjectLockLegalHold] = "ON" + putHdrs[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp1000 + putReq, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectURL("", bucketName, object)+"?versionId="+versionID, int64(len(cipher)), + bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, putHdrs) + if err != nil { + t.Fatal(err) + } + putRec := httptest.NewRecorder() + apiRouter.ServeHTTP(putRec, putReq) + if putRec.Code != http.StatusOK { + t.Fatalf("%s: interleaving replica PUT %d: %s", instanceType, putRec.Code, putRec.Body.String()) + } + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got.legalHold != "ON" { + t.Fatalf("%s: the interleaving PUT did not set the newer ON: got %+v", instanceType, got) + } + + // Finish the multipart upload the sender's way and prove the newer ON survives. + partReq, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectPartURL("", bucketName, object, init.UploadID, "1"), int64(len(cipher)), + bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, + map[string]string{xhttp.MinIOSourceReplicationRequest: "true"}) + if err != nil { + t.Fatal(err) + } + partRec := httptest.NewRecorder() + apiRouter.ServeHTTP(partRec, partReq) + if partRec.Code != http.StatusOK { + t.Fatalf("%s: replica PutObjectPart %d: %s", instanceType, partRec.Code, partRec.Body.String()) + } + actualSize, err := srcInfo.GetActualSize() + if err != nil { + t.Fatal(err) + } + 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, 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) + } + completeRec := httptest.NewRecorder() + apiRouter.ServeHTTP(completeRec, completeReq) + if completeRec.Code != http.StatusOK { + t.Fatalf("%s: replica CompleteMultipartUpload %d: %s", instanceType, completeRec.Code, completeRec.Body.String()) + } + + // The completion re-ordered the carried OFF@09:00 against the ON@10:00 that + // reached the version after initiation: the newer hold and its timestamp win. + want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000} + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: a newer legal hold that arrived after initiation was rolled back at completion: got %+v, want %+v", + instanceType, got, want) + } +} + +// TestReplicaPutObjectLockReconcileUnderWriteLock exercises the PUT counterpart +// of the multipart reconcile: a replica full write reaches the object layer +// carrying the Object Lock its handler resolved, but the addressed version has +// since taken a newer lock update. PutObject must re-order the incoming lock +// against the version read under the write lock, so the newer stored value is +// kept. Driving the object layer directly stands in for the handler-read / +// backend-commit interleave without a timing race. See pgsty/silo#120. +func TestReplicaPutObjectLockReconcileUnderWriteLock(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testReplicaPutObjectLockReconcileUnderWriteLock, + makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + }) +} + +func testReplicaPutObjectLockReconcileUnderWriteLock(obj ObjectLayer, instanceType, bucketName string, + _ http.Handler, _ auth.Credentials, t *testing.T, +) { + ctx := t.Context() + + seedVersion := func(t *testing.T, object string, meta map[string]string) string { + t.Helper() + info, err := obj.PutObject(ctx, bucketName, object, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), + ObjectOptions{Versioned: true, UserDefined: meta}) + if err != nil { + t.Fatal(err) + } + return info.VersionID + } + // replicaWrite overwrites the addressed version the way a replica retransmit + // reaches the object layer, with the in-lock reconcile enabled. + replicaWrite := func(t *testing.T, object, versionID string, meta map[string]string) { + t.Helper() + _, err := obj.PutObject(ctx, bucketName, object, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), + ObjectOptions{Versioned: true, VersionID: versionID, ReplicaLockReconcile: true, UserDefined: meta}) + if err != nil { + t.Fatal(err) + } + } + + t.Run("older-legal-hold-off-does-not-clear-newer-on", func(t *testing.T) { + object := "reconcile/put-legal-hold" + versionID := seedVersion(t, object, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000, + }) + replicaWrite(t, object, versionID, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp0900, + }) + want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000} + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: in-lock reconcile let an older OFF overwrite the newer stored ON: got %+v, want %+v", + instanceType, got, want) + } + }) + + t.Run("stale-retention-does-not-overwrite-newer", func(t *testing.T) { + object := "reconcile/put-retention" + versionID := seedVersion(t, object, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockMode): "GOVERNANCE", + strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): retransmitRetainUntilNewer, + ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: objectLockTestStamp1000, + }) + replicaWrite(t, object, versionID, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockMode): "GOVERNANCE", + strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): retransmitRetainUntilStale, + ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: objectLockTestStamp0900, + }) + want := objectLockFields{ + mode: "GOVERNANCE", retainUntil: retransmitRetainUntilNewer, retentionStamp: objectLockTestStamp1000, + } + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: in-lock reconcile let a stale retention overwrite the newer stored one: got %+v, want %+v", + instanceType, got, want) + } + }) + + t.Run("newer-incoming-hold-applies", func(t *testing.T) { + object := "reconcile/put-newer-applies" + versionID := seedVersion(t, object, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp0900, + }) + replicaWrite(t, object, versionID, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000, + }) + want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000} + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: in-lock reconcile did not apply the newer incoming hold: got %+v, want %+v", + instanceType, got, want) + } + }) + + t.Run("pre-upgrade-shape-on-absent-version-is-preserved", func(t *testing.T) { + // A pre-upgrade upload persisted validated lock values WITHOUT their + // ordering timestamps. Completing it while the destination version is + // absent must keep those values: there is nothing to order against, so the + // reconcile is skipped rather than deleting the accepted lock. The same + // not-found handling guards CompleteMultipartUpload. + object := "reconcile/put-absent-version" + absentVersion := mustGetUUID() + _, err := obj.PutObject(ctx, bucketName, object, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), + ObjectOptions{Versioned: true, VersionID: absentVersion, ReplicaLockReconcile: true, UserDefined: map[string]string{ + strings.ToLower(xhttp.AmzObjectLockMode): "GOVERNANCE", + strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): retransmitRetainUntilNewer, + strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", + }}) + if err != nil { + t.Fatal(err) + } + want := objectLockFields{mode: "GOVERNANCE", retainUntil: retransmitRetainUntilNewer, legalHold: "ON"} + if got := readObjectLockFields(t, obj, bucketName, object, absentVersion); got != want { + t.Fatalf("%s: a pre-upgrade lock (no ordering timestamps) on an absent version was stripped: got %+v, want %+v", + instanceType, got, want) + } + }) +} + +// TestReplicaLockReconcileNullVersion covers the null version, which persisted +// upload metadata records as an empty VersionID. The completion reconcile must +// order the incoming lock against the null version's own stored state -- looked +// up as the null version, not the latest -- so a retransmit addressing the null +// version cannot be reconciled against an unrelated UUID version, and a null +// version absent while a UUID version exists is not mistaken for present. Single +// erasure set. See pgsty/silo#120. +func TestReplicaLockReconcileNullVersion(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testReplicaLockReconcileNullVersion, + makeBucketOptions: MakeBucketOptions{VersioningEnabled: true}, + }) +} + +func testReplicaLockReconcileNullVersion(obj ObjectLayer, instanceType, bucketName string, + _ http.Handler, _ auth.Credentials, t *testing.T, +) { + ctx := t.Context() + + // completeNullMPU runs an SSE-C replica multipart upload addressing the null + // version (VersionSuspended) carrying uploadLock, through the reconcile. + completeNullMPU := func(t *testing.T, object string, uploadLock map[string]string) { + t.Helper() + meta := map[string]string{crypto.MetaSealedKeySSEC: "dummy-sealed-key"} + maps.Copy(meta, uploadLock) + res, err := obj.NewMultipartUpload(ctx, bucketName, object, ObjectOptions{VersionSuspended: true, UserDefined: meta}) + if err != nil { + t.Fatal(err) + } + part, err := obj.PutObjectPart(ctx, bucketName, object, res.UploadID, 1, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + if _, err := obj.CompleteMultipartUpload(ctx, bucketName, object, res.UploadID, + []CompletePart{{PartNumber: 1, ETag: part.ETag}}, + ObjectOptions{VersionSuspended: true, ReplicaLockReconcile: true}); err != nil { + t.Fatal(err) + } + } + + t.Run("null-and-uuid-present-reconciles-the-null-version", func(t *testing.T) { + object := "reconcile/null-vs-uuid" + // The null version holds a newer legal hold ON@10:00. + if _, err := obj.PutObject(ctx, bucketName, object, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), + ObjectOptions{VersionSuspended: true, MTime: time.Date(2026, 9, 3, 10, 0, 0, 0, time.UTC), UserDefined: map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000, + }}); err != nil { + t.Fatal(err) + } + // A later UUID version holds an unrelated OFF@11:00 and is the latest. + uuidInfo, err := obj.PutObject(ctx, bucketName, object, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), + ObjectOptions{Versioned: true, MTime: time.Date(2026, 9, 3, 11, 0, 0, 0, time.UTC), UserDefined: map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1100, + }}) + if err != nil { + t.Fatal(err) + } + + // A null-version SSE-C retransmit carrying an older OFF@09:00. + completeNullMPU(t, object, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp0900, + }) + + // The null version keeps its own newer ON@10:00; the UUID is untouched. + wantNull := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000} + if got := readObjectLockFields(t, obj, bucketName, object, nullVersionID); got != wantNull { + t.Fatalf("%s: null-version completion reconciled against the wrong version: got %+v, want %+v", instanceType, got, wantNull) + } + wantUUID := objectLockFields{legalHold: "OFF", legalHoldStamp: objectLockTestStamp1100} + if got := readObjectLockFields(t, obj, bucketName, object, uuidInfo.VersionID); got != wantUUID { + t.Fatalf("%s: the UUID version was changed by a null-version completion: got %+v, want %+v", instanceType, got, wantUUID) + } + }) + + t.Run("absent-null-with-uuid-keeps-accepted-lock", func(t *testing.T) { + object := "reconcile/null-absent" + // Only a UUID version exists; there is no null version. + if _, err := obj.PutObject(ctx, bucketName, object, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), + ObjectOptions{Versioned: true, MTime: time.Date(2026, 9, 3, 11, 0, 0, 0, time.UTC), UserDefined: map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1100, + }}); err != nil { + t.Fatal(err) + } + + // A null-version retransmit with its own validated legal hold ON. The null + // version does not exist, so the write must keep its accepted lock rather + // than order against the unrelated UUID version. + completeNullMPU(t, object, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", + }) + if got := readObjectLockFields(t, obj, bucketName, object, nullVersionID); got.legalHold != "ON" { + t.Fatalf("%s: an absent null version was reconciled against the UUID version: got %+v", instanceType, got) + } + }) +} From 34cbca97ea210130e1b00b0add0c137e5da3d2e1 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sun, 6 Sep 2026 08:37:06 +0800 Subject: [PATCH 3/5] docs: point the multi-pool lock follow-up at pgsty/silo#133 Fill the tracked-issue number into the scope comments of the single erasure set Object Lock reconcile added for SSE-C replica retransmit. No behaviour change. Refs pgsty/silo#120 Refs pgsty/silo#133 Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01L7qJqWwy8oFA6aCXWRzXQe Signed-off-by: Feng Ruohang --- cmd/erasure-multipart.go | 2 +- cmd/erasure-object.go | 2 +- cmd/object-api-interface.go | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go index c50c44b0f..9de330e28 100644 --- a/cmd/erasure-multipart.go +++ b/cmd/erasure-multipart.go @@ -1164,7 +1164,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str // // Scope: correct for a single erasure set. A multi-pool deployment (duplicate // versions across pools, ModTime ties, cross-pool lock authority) is out of - // scope and tracked in pgsty/silo#TBD-multipool-lock. + // scope and tracked in pgsty/silo#133. if opts.ReplicaLockReconcile && crypto.SSEC.IsEncrypted(fi.Metadata) { // A persisted upload records the null version as an empty VersionID; look // it up as the null version so the reconcile reads the addressed version's diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index a6293edc2..bade406e6 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -1305,7 +1305,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st // // Scope: correct for a single erasure set. A multi-pool deployment // (duplicate versions across pools, ModTime ties, cross-pool lock - // authority) is out of scope and tracked in pgsty/silo#TBD-multipool-lock. + // authority) is out of scope and tracked in pgsty/silo#133. if opts.ReplicaLockReconcile && err == nil { reconcileStoredObjectLock(opts.UserDefined, storedObjectLockState(obj.UserDefined)) } diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go index 2f64ded4f..3361a9ced 100644 --- a/cmd/object-api-interface.go +++ b/cmd/object-api-interface.go @@ -99,7 +99,7 @@ type ObjectOptions struct { ReplicationSourceTaggingTimestamp time.Time // set if MinIOSourceTaggingTimestamp received ReplicationSourceLegalholdTimestamp time.Time // set if MinIOSourceObjectLegalholdTimestamp received ReplicationSourceRetentionTimestamp time.Time // set if MinIOSourceObjectRetentionTimestamp received - ReplicaLockReconcile bool // set for a trusted SSE-C replica full write/completion: re-order Object Lock against the destination version read under the write lock (single erasure set; see pgsty/silo#TBD-multipool-lock) + ReplicaLockReconcile bool // set for a trusted SSE-C replica full write/completion: re-order Object Lock against the destination version read under the write lock (single erasure set; see pgsty/silo#133) DeletePrefix bool // set true to enforce a prefix deletion, only application for DeleteObject API, DeletePrefixObject bool // set true when object's erasure set is resolvable by object name (using getHashedSetIndex) From 7220210e8d89cf687f88406e9d0a03e63c4a53a8 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sun, 6 Sep 2026 11:55:38 +0800 Subject: [PATCH 4/5] test: reconcile #134 fixtures with #119 and refresh the rebrand baseline Rebased onto current main. #119 made PutObjectPart derive an encrypted part's plaintext length and reject a part that cannot be a valid sio stream, so TestReplicaLockReconcileNullVersion's completeNullMPU fixture (a 4-byte plaintext part under SSE-C metadata) no longer stores; build it with sio.Encrypt like the other encrypted-part fixtures. Also regenerate the rebrand-guard baseline for the replication SSE header the retransmit path reintroduces (headers 84 -> 85). Mechanical integration only. Signed-off-by: Feng Ruohang --- buildscripts/rebrand-guard/compat-baseline.json | 1 + cmd/replication-ssec-retransmit_test.go | 11 ++++++++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/buildscripts/rebrand-guard/compat-baseline.json b/buildscripts/rebrand-guard/compat-baseline.json index 84ee7a4bc..8ae0f32dc 100644 --- a/buildscripts/rebrand-guard/compat-baseline.json +++ b/buildscripts/rebrand-guard/compat-baseline.json @@ -631,6 +631,7 @@ "x-minio-replication-encrypted-multipart", "x-minio-replication-ready", "x-minio-replication-reset-status", + "x-minio-replication-server-side-encryption", "x-minio-replication-server-side-encryption-iv", "x-minio-replication-server-side-encryption-seal-algorithm", "x-minio-replication-server-side-encryption-sealed-key", diff --git a/cmd/replication-ssec-retransmit_test.go b/cmd/replication-ssec-retransmit_test.go index bbb8712b9..8d4186627 100644 --- a/cmd/replication-ssec-retransmit_test.go +++ b/cmd/replication-ssec-retransmit_test.go @@ -30,6 +30,7 @@ import ( "github.com/minio/minio/internal/auth" "github.com/minio/minio/internal/crypto" xhttp "github.com/minio/minio/internal/http" + "github.com/minio/sio" ) // TestAPISSECReplicationTargetHead pins what the replication sender's target @@ -549,6 +550,7 @@ func testAPISSECReplicaRetransmitOverExistingVersion(obj ObjectLayer, instanceTy } assertRetransmittedVersion(t, obj, apiRouter, credentials, bucketName, object, srcInfo.VersionID, "retransmit=multipart", data, sseHeaders) }) + } // assertRetransmittedVersion checks that a retransmit landed on the addressed @@ -1718,8 +1720,15 @@ func testReplicaLockReconcileNullVersion(obj ObjectLayer, instanceType, bucketNa if err != nil { t.Fatal(err) } + // PutObjectPart now validates the DARE stream length (#119). This + // object-layer fixture needs an encrypted body before it can exercise + // the completion-time null-version metadata reconciliation. + var ciphertext bytes.Buffer + if _, err := sio.Encrypt(&ciphertext, bytes.NewReader([]byte("data")), sio.Config{Key: bytes.Repeat([]byte{1}, 32)}); err != nil { + t.Fatal(err) + } part, err := obj.PutObjectPart(ctx, bucketName, object, res.UploadID, 1, - mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), ObjectOptions{}) + mustGetPutObjReader(t, bytes.NewReader(ciphertext.Bytes()), int64(ciphertext.Len()), "", ""), ObjectOptions{}) if err != nil { t.Fatal(err) } From 236e163c0b76c121bd10894a32b0f5798eead040 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sun, 6 Sep 2026 11:55:38 +0800 Subject: [PATCH 5/5] fix: repair an undecodable SSE-C replica on retransmit PutObjectHandler's precondition callback ran DecryptObjectInfo on the stored object before checkPreconditionsPUT, so an authenticated raw SSE-C replica overwrite was rejected when the stored version could not decrypt. A replica a pre-fix destination (issue #109) left as compress(ciphertext) or a re-encrypted body has an invalid decrypted length, so DecryptObjectInfo returned errObjectTampered and the retransmission that repairs it never ran -- the version stayed damaged through resync. #134's raw-replica exemption only covered the version/ETag duplicate check inside checkPreconditionsPUT, one step too late. Skip the stored object's decryption precondition only for a PURE raw SSE-C replica overwrite (a trusted SSE-C replica write with no public precondition), keyed on the incoming request's restored SSE-C metadata, the same predicate checkPreconditionsPUT uses. Such a write fully replaces the object, so requiring the damaged stored version to decrypt is both wrong and unnecessary. A conditional request keeps the check: DecryptObjectInfo also normalizes the stored sealed ETag to the client-visible one, and If-Match/If-None-Match must compare against that, not the sealed ETag -- skipping it for every replica inverted both conditions. Ordinary writes and non-SSE-C replicas are unchanged. Adds red/green regressions: a raw retransmit over a version staged as an undecodable body returns 500 XMinioObjectTampered before this change and 200 with full customer-key recovery after; and a conditional replica PUT (If-Match / If-None-Match) on the client-visible ETag is honoured rather than inverted. Fixes the single-PUT compression-damage recovery gap in Signed-off-by: Feng Ruohang #120. --- cmd/object-handlers.go | 18 ++- cmd/replication-ssec-retransmit_test.go | 141 ++++++++++++++++++++++++ 2 files changed, 156 insertions(+), 3 deletions(-) diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go index 789d0fe2a..331393292 100644 --- a/cmd/object-handlers.go +++ b/cmd/object-handlers.go @@ -2227,9 +2227,21 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req r.Header.Get(xhttp.IfMatch) != "" || r.Header.Get(xhttp.IfNoneMatch) != "" { opts.CheckPrecondFn = func(oi ObjectInfo) bool { - if _, err := DecryptObjectInfo(&oi, r); err != nil { - writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) - return true + // A pure raw SSE-C replica overwrite (no public precondition) fully + // replaces the stored object, so the destination must not first + // require the stored object to decrypt: a replica an older destination + // bug left as compress(ciphertext) or double-encrypted has an invalid + // decrypted length, and requiring it here blocks the retransmission + // that repairs it. The predicate is the incoming request's restored + // SSE-C metadata, the same one checkPreconditionsPUT uses to exempt the + // version/ETag duplicate. A conditional request (If-Match/If-None-Match) + // still needs the decrypted, client-visible ETag, so it keeps the check. + ssecReplica := isReplicaTrusted(ctx) && crypto.SSEC.IsEncrypted(opts.UserDefined) + if !ssecReplica || r.Header.Get(xhttp.IfMatch) != "" || r.Header.Get(xhttp.IfNoneMatch) != "" { + if _, err := DecryptObjectInfo(&oi, r); err != nil { + writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) + return true + } } return checkPreconditionsPUT(ctx, w, r, oi, opts) } diff --git a/cmd/replication-ssec-retransmit_test.go b/cmd/replication-ssec-retransmit_test.go index 8d4186627..c845b4668 100644 --- a/cmd/replication-ssec-retransmit_test.go +++ b/cmd/replication-ssec-retransmit_test.go @@ -551,6 +551,147 @@ func testAPISSECReplicaRetransmitOverExistingVersion(obj ObjectLayer, instanceTy assertRetransmittedVersion(t, obj, apiRouter, credentials, bucketName, object, srcInfo.VersionID, "retransmit=multipart", data, sseHeaders) }) + t.Run("repairs-an-undecodable-existing-version", func(t *testing.T) { + // A pre-fix destination (issue #109) could persist an SSE-C replica as + // compress(ciphertext) or a re-encrypted body, leaving a stored length + // that is not a valid encryption stream. Resync repairs such a version by + // retransmitting the source's raw ciphertext, but the write must not first + // require the stored, damaged object to decrypt. See issue #120. + data := bytes.Repeat([]byte("SILO raw SSE-C recovery\n"), 400) + object := "ssec-duplicate/undecodable" + 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) + } + + // Stage the damage: overwrite the version with a body too short to be a + // valid encryption stream, standing in for the compression/re-encryption + // an old destination left behind. The staging write itself is a raw + // replica over the still-valid version, so it stores verbatim. + damaged := []byte("dmg!!") + if _, derr := sio.DecryptedSize(uint64(len(damaged))); derr == nil { + t.Fatalf("%s: fixture body of %d bytes is a valid stream length, not undecodable", instanceType, len(damaged)) + } + stageHdrs := replicaHeaders(t, srcInfo) + stageURL := getPutObjectURL("", bucketName, object) + "?versionId=" + srcInfo.VersionID + stageReq, err := newTestSignedRequestV4(http.MethodPut, stageURL, int64(len(damaged)), + bytes.NewReader(damaged), replicator.AccessKey, replicator.SecretKey, stageHdrs) + if err != nil { + t.Fatal(err) + } + stageRec := httptest.NewRecorder() + apiRouter.ServeHTTP(stageRec, stageReq) + if stageRec.Code != http.StatusOK { + t.Fatalf("%s: could not stage the damaged replica: %d %s", instanceType, stageRec.Code, stageRec.Body.String()) + } + // The staged version is genuinely undecodable at the object layer, which + // is exactly what makes DecryptObjectInfo fail during the overwrite. + staged, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{VersionID: srcInfo.VersionID}) + if err != nil { + t.Fatal(err) + } + if _, derr := staged.DecryptedSize(); derr == nil { + t.Fatalf("%s: staged replica is decodable, cannot exercise the repair path", instanceType) + } + + // Retransmit the correct ciphertext over the same version. Before the + // raw-replica precondition exemption this failed with XMinioObjectTampered + // because the damaged object could not decrypt; it must now repair. + srcInfo.UserTags = "retransmit=undecodable" + 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: retransmit over an undecodable version status %d, want 200: %s", instanceType, rec.Code, rec.Body.String()) + } + assertRetransmittedVersion(t, obj, apiRouter, credentials, bucketName, object, srcInfo.VersionID, "retransmit=undecodable", data, sseHeaders) + }) + + // setupHealthySSECVersion writes a normal SSE-C object and returns its + // ObjectInfo (for building replica headers), its ciphertext, and the + // client-visible ETag a keyed reader sees -- the decrypted ETag, which is + // distinct from the stored sealed ETag. + setupHealthySSECVersion := func(t *testing.T, object string, data []byte) (srcInfo ObjectInfo, cipher []byte, clientETag string) { + t.Helper() + 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 client-visible ETag is the decrypted one, which the handler derives + // with the customer key through DecryptObjectInfo; compute it the same way. + keyHeader := http.Header{} + for k, v := range sseHeaders { + keyHeader.Set(k, v) + } + clientETag = getDecryptedETag(keyHeader, srcInfo, false) + if clientETag == "" || clientETag == srcInfo.ETag { + t.Fatalf("%s: client ETag %q is not distinct from the sealed ETag %q", instanceType, clientETag, srcInfo.ETag) + } + return srcInfo, cipher, clientETag + } + + // A conditional replica PUT must compare the public precondition against the + // client-visible ETag, not the stored sealed one. Skipping DecryptObjectInfo + // for every raw SSE-C replica (not only a pure overwrite) left oi.ETag sealed + // and inverted both conditions. + t.Run("if-match-on-the-client-etag-proceeds", func(t *testing.T) { + object := "ssec-duplicate/cond-if-match" + srcInfo, cipher, clientETag := setupHealthySSECVersion(t, object, bytes.Repeat([]byte("cond-if-match-"), 64)) + hdrs := replicaHeaders(t, srcInfo) + hdrs[xhttp.IfMatch] = "\"" + clientETag + "\"" + req, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectURL("", bucketName, object)+"?versionId="+srcInfo.VersionID, + 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: If-Match on the client ETag status %d, want 200: %s", instanceType, rec.Code, rec.Body.String()) + } + }) + + t.Run("if-none-match-on-the-client-etag-fails", func(t *testing.T) { + object := "ssec-duplicate/cond-if-none-match" + srcInfo, cipher, clientETag := setupHealthySSECVersion(t, object, bytes.Repeat([]byte("cond-if-none-"), 64)) + hdrs := replicaHeaders(t, srcInfo) + hdrs[xhttp.IfNoneMatch] = "\"" + clientETag + "\"" + req, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectURL("", bucketName, object)+"?versionId="+srcInfo.VersionID, + 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.StatusPreconditionFailed { + t.Fatalf("%s: If-None-Match on the client ETag status %d, want 412: %s", instanceType, rec.Code, rec.Body.String()) + } + }) } // assertRetransmittedVersion checks that a retransmit landed on the addressed