From 296c9710b16467e53fd4d861bf0b01f35fe6e555 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Tue, 10 Nov 2026 21:51:47 +0800 Subject: [PATCH] fix(security): close release blockers Signed-off-by: Feng Ruohang --- cmd/bucket-handlers.go | 22 +- cmd/bucket-metadata-cache-security_test.go | 125 +++++++ cmd/bucket-metadata-sys.go | 11 +- cmd/bucket-object-lock.go | 21 +- cmd/erasure-multipart.go | 3 + cmd/erasure-object.go | 14 + ...rasure-server-pool-put-conditional_test.go | 9 +- cmd/erasure-server-pool.go | 6 + cmd/object-api-interface.go | 2 + cmd/replica-content.go | 205 ++++++++++++ cmd/replica-content_test.go | 314 ++++++++++++++++++ cmd/replication-signed-headers_test.go | 94 ++++++ cmd/replication-ssec-retransmit_test.go | 16 +- cmd/replication-trust.go | 23 +- cmd/s3-zip-handlers.go | 34 +- cmd/s3-zip-readonly_test.go | 109 ++++++ cmd/streaming-auth-boundary_test.go | 114 +++++++ cmd/streaming-signature-v4.go | 3 + cmd/streaming-v4-unsigned.go | 8 +- 19 files changed, 1080 insertions(+), 53 deletions(-) create mode 100644 cmd/bucket-metadata-cache-security_test.go create mode 100644 cmd/replica-content.go create mode 100644 cmd/replica-content_test.go create mode 100644 cmd/replication-signed-headers_test.go create mode 100644 cmd/s3-zip-readonly_test.go create mode 100644 cmd/streaming-auth-boundary_test.go diff --git a/cmd/bucket-handlers.go b/cmd/bucket-handlers.go index 23dd5b9cd..07d1a06b3 100644 --- a/cmd/bucket-handlers.go +++ b/cmd/bucket-handlers.go @@ -478,15 +478,11 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter, getObjectInfoFn := objectAPI.GetObjectInfo var ( - hasLockEnabled bool - dsc ReplicateDecision - goi ObjectInfo - gerr error + dsc ReplicateDecision + goi ObjectInfo + gerr error ) replicateDeletes := hasReplicationRules(ctx, bucket, deleteObjectsReq.Objects) - if rcfg, _ := globalBucketObjectLockSys.Get(bucket); rcfg.LockEnabled { - hasLockEnabled = true - } type deleteResult struct { delInfo DeletedObject @@ -537,10 +533,10 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter, VersionSuspended: vc.Suspended(), } - // Physical version deletes in a locked bucket are checked by the - // single-object path below, while its mutation lock is held. A lookup - // here could miss a later protection update or a protected pool copy. - if hasLockEnabled && object.VersionID != "" { + // Every physical version delete checks the version's own protection + // under its mutation lock. Bucket metadata may be stale or unavailable + // and cannot establish that a stored version is unprotected. + if object.VersionID != "" { if first, ok := objectsToDelete[object]; ok { duplicateOf[index] = first } else { @@ -605,7 +601,7 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter, bulk := make([]ObjectToDelete, 0, len(deleteList)) bulkIndex := make([]int, 0, len(deleteList)) for i, object := range deleteList { - if !hasLockEnabled || object.VersionID == "" { + if object.VersionID == "" { bulk = append(bulk, object) bulkIndex = append(bulkIndex, i) continue @@ -663,7 +659,7 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter, // under the mutation lock, and a missing version has no stored identity. dindex := objectsToDelete[deleteList[i]] if errs[i] == nil || isErrObjectNotFound(errs[i]) || isErrVersionNotFound(errs[i]) { - if replicateDeletes && (!hasLockEnabled || deleteList[i].VersionID == "") { + if replicateDeletes && deleteList[i].VersionID == "" { dObjects[i].ReplicationState = deleteList[i].ReplicationState() } deleteResults[dindex].delInfo = dObjects[i] diff --git a/cmd/bucket-metadata-cache-security_test.go b/cmd/bucket-metadata-cache-security_test.go new file mode 100644 index 000000000..0f511f470 --- /dev/null +++ b/cmd/bucket-metadata-cache-security_test.go @@ -0,0 +1,125 @@ +// Copyright (c) 2026 Feng Ruohang +// +// This file is part of Silo 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. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "context" + "encoding/xml" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/minio/minio/internal/auth" +) + +type metadataCacheLoadBarrier struct { + ObjectLayer + key string + read, resume chan struct{} + once sync.Once +} + +func (b *metadataCacheLoadBarrier) GetObjectNInfo(ctx context.Context, bucket, key string, rs *HTTPRangeSpec, h http.Header, opts ObjectOptions) (*GetObjectReader, error) { + r, err := b.ObjectLayer.GetObjectNInfo(ctx, bucket, key, rs, h, opts) + if bucket == minioMetaBucket && key == b.key { + b.once.Do(func() { close(b.read); <-b.resume }) + } + return r, err +} + +func TestBucketMetadataLoadPreservesConcurrentPublication(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: func(obj ObjectLayer, backend, bucket string, _ http.Handler, _ auth.Credentials, t *testing.T) { + // Disk has the old unlocked config; a concurrent publication enables + // lock and versioning after the cache miss has started loading it. + enableBucketObjectLock(t, bucket) + newer, err := globalBucketMetadataSys.Get(bucket) + if err != nil { + t.Fatal(err) + } + globalBucketMetadataSys.Remove(bucket) + barrier := &metadataCacheLoadBarrier{ObjectLayer: obj, key: pathJoin(bucketMetaPrefix, bucket, bucketMetadataFile), read: make(chan struct{}), resume: make(chan struct{})} + previous := newObjectLayerFn() + setObjectLayer(barrier) + defer setObjectLayer(previous) + release := sync.OnceFunc(func() { close(barrier.resume) }) + defer release() + type result struct { + meta BucketMetadata + reloaded bool + err error + } + done := make(chan result, 1) + go func() { + meta, reloaded, err := globalBucketMetadataSys.GetConfig(t.Context(), bucket) + done <- result{meta, reloaded, err} + }() + select { + case <-barrier.read: + case <-time.After(10 * time.Second): + t.Fatal("metadata load did not reach disk") + } + globalBucketMetadataSys.Set(bucket, newer) + release() + select { + case got := <-done: + if got.err != nil || got.reloaded || got.meta.objectLockConfig == nil || got.meta.objectLockConfig.ObjectLockEnabled != "Enabled" { + t.Fatalf("%s: stale load replaced the publication: %+v", backend, got) + } + case <-time.After(10 * time.Second): + t.Fatal("metadata load did not finish") + } + cached, err := globalBucketMetadataSys.Get(bucket) + if err != nil || cached.objectLockConfig == nil || cached.objectLockConfig.ObjectLockEnabled != "Enabled" { + t.Fatalf("cached lock lost: %+v %v", cached, err) + } + }, + }) +} + +func TestBulkDeleteVersionLockWithStaleBucketCache(t *testing.T) { + for _, pools := range []int{1, 2} { + z, _ := bulkDeletePools(t, pools) + bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{LockEnabled: true}) + if err != nil { + t.Fatal(err) + } + for _, protection := range batchLockKinds { + oi := batchPutVersion(t, z, bucket, protection, batchLockMeta(protection)) + saved, err := globalBucketMetadataSys.Get(bucket) + if err != nil { + t.Fatal(err) + } + globalBucketMetadataSys.Set(bucket, newBucketMetadata(bucket)) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, bulkDeleteLockRequest(t, bucket, protection, oi.VersionID)) + globalBucketMetadataSys.Set(bucket, saved) + var response DeleteObjectsResponse + if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if rec.Code != http.StatusOK || len(response.Errors) != 1 || len(response.DeletedObjects) != 0 || !batchVersionExists(t, z, bucket, protection, oi.VersionID) { + t.Fatalf("pools=%d %s: stale bucket cache bypassed version lock: %s", pools, protection, rec.Body.String()) + } + } + } +} diff --git a/cmd/bucket-metadata-sys.go b/cmd/bucket-metadata-sys.go index afe09e6aa..2ca8dab0f 100644 --- a/cmd/bucket-metadata-sys.go +++ b/cmd/bucket-metadata-sys.go @@ -697,7 +697,7 @@ func (sys *BucketMetadataSys) GetConfig(ctx context.Context, bucket string) (met } val, err, _ := sys.group.Do(bucket, func() (val any, err error) { - meta, err = loadBucketMetadata(ctx, objAPI, bucket) + meta, err := loadBucketMetadata(ctx, objAPI, bucket) if err != nil { if !sys.Initialized() { // bucket metadata not yet initialized @@ -707,13 +707,18 @@ func (sys *BucketMetadataSys) GetConfig(ctx context.Context, bucket string) (met return meta, err }) meta, _ = val.(BucketMetadata) + sys.Lock() + defer sys.Unlock() + // A writer may have published a newer configuration while the disk read + // was in flight. Never replace that publication with the older load. + if current, ok := sys.metadataMap[bucket]; ok { + return current, false, nil + } if err != nil { return meta, false, err } - sys.Lock() sys.metadataMap[bucket] = meta sys.clearLoadFailure(bucket) - sys.Unlock() return meta, true, nil } diff --git a/cmd/bucket-object-lock.go b/cmd/bucket-object-lock.go index f791d7a12..25fdcaa1e 100644 --- a/cmd/bucket-object-lock.go +++ b/cmd/bucket-object-lock.go @@ -491,6 +491,25 @@ func (s objectLockState) legalHoldIsOlderThan(src time.Time) bool { return olderThan(s.legalHoldTimestamp, src) } +// A peer's ordering timestamp cannot override an unexpired COMPLIANCE lock. +// Otherwise it could first remove retention with a same-content retransmit, +// then replace the now-unprotected version in a second request. +func (s objectLockState) complianceWouldWeaken(incoming objectLockState) bool { + if s.mode != string(objectlock.RetCompliance) { + return false + } + until, err := time.Parse(time.RFC3339Nano, s.retainUntil) + if err != nil { + return true + } + now, err := objectlock.UTCNowNTP() + if err == nil && !until.After(now) { + return false + } + nextUntil, nextErr := time.Parse(time.RFC3339Nano, incoming.retainUntil) + return incoming.mode != string(objectlock.RetCompliance) || nextErr != nil || nextUntil.Before(until) +} + // restoreRetention and restoreLegalHold put the stored state back into // metadata that was rebuilt from a request whose update was not applied. func (s objectLockState) restoreRetention(metadata map[string]string) { @@ -610,7 +629,7 @@ func reconcileStoredObjectLock(metadata map[string]string, stored objectLockStat incoming := storedObjectLockState(metadata) incomingRetentionTS, _ := time.Parse(time.RFC3339Nano, incoming.retentionTimestamp) - if !stored.retentionIsOlderThan(incomingRetentionTS) { + if !stored.retentionIsOlderThan(incomingRetentionTS) || stored.complianceWouldWeaken(incoming) { // 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 diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go index 12b662fe5..5049b1320 100644 --- a/cmd/erasure-multipart.go +++ b/cmd/erasure-multipart.go @@ -1722,6 +1722,9 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str } } + if err := er.checkReplicaContent(ctx, bucket, object, minioMetaMultipartBucket, uploadIDPath, partsMetadata, onlineDisks, opts); err != nil { + return oi, toObjectErr(err, bucket, object) + } er.cleanupMultipartPath(ctx, paths...) // cleanup all part.N.meta, and skipped part.N's before final rename(). defer func() { diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index 61a31aa15..fec815c73 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -148,6 +148,17 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d fi.IsLatest = true // we are creating a new version so this is latest. } } + if srcInfo.versionOnly && (dstOpts.ReplicaLockReconcile || dstOpts.ReplicationRequest) && !dstOpts.DataMovement { + candidate := slices.Clone(metaArr) + for i := range candidate { + candidate[i].VersionID = versionID + candidate[i].Metadata = srcInfo.UserDefined + } + readDisks, readMeta := shuffleDisksAndPartsMetadataByIndex(onlineDisks, candidate, fi) + if err := er.checkReplicaContent(ctx, dstBucket, dstObject, srcBucket, srcObject, readMeta, readDisks, dstOpts); err != nil { + return oi, err + } + } if !dstOpts.ReplicationRequest && !dstOpts.ReplicaLockReconcile && !dstOpts.DataMovement && (versionID == "" || versionID == nullVersionID) { @@ -1722,6 +1733,9 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st } // Rename the successfully written temporary object to final location. + if err := er.checkReplicaContent(ctx, bucket, object, minioMetaTmpBucket, tempObj, partsMetadata, onlineDisks, opts); err != nil { + return ObjectInfo{}, toObjectErr(err, bucket, object) + } onlineDisks, versions, oldDataDir, err := renameData(ctx, onlineDisks, minioMetaTmpBucket, tempObj, partsMetadata, bucket, object, writeQuorum) if err != nil { if errors.Is(err, errFileNotFound) { diff --git a/cmd/erasure-server-pool-put-conditional_test.go b/cmd/erasure-server-pool-put-conditional_test.go index cd4897921..cef91faa2 100644 --- a/cmd/erasure-server-pool-put-conditional_test.go +++ b/cmd/erasure-server-pool-put-conditional_test.go @@ -590,9 +590,9 @@ func TestSinglePoolConditionalPutHTTP(t *testing.T) { } } -// An internal replica callback without an addressed version is not a public -// condition. Preserve its availability when another pool is unreadable. An -// addressed replica already requires all pools for lock/tag reconciliation. +// Replica overwrites must establish protection across every pool, including +// the null version of an unversioned request. An unreadable pool may hold a +// protected copy, so neither addressed nor unaddressed writes may replace it. func TestPoolsConditionalPutReplicaAvailability(t *testing.T) { for _, addressed := range []bool{false, true} { t.Run(fmt.Sprintf("addressed=%t", addressed), func(t *testing.T) { @@ -615,10 +615,9 @@ func TestPoolsConditionalPutReplicaAvailability(t *testing.T) { xhttp.MinIOSourceETag: fmt.Sprintf("%x", md5.Sum([]byte("replacement"))), } url := getPutObjectURL("", bucket, object) - wantStatus, wantBody := http.StatusOK, "replacement" + wantStatus, wantBody := http.StatusServiceUnavailable, "old" if addressed { url += "?versionId=" + oi.VersionID - wantStatus, wantBody = http.StatusServiceUnavailable, "old" } put := multipartConditionRequest(t, router, http.MethodPut, url, "replacement", headers) if put.Code != wantStatus { diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index 907a88b18..6a9dbdd6e 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1195,6 +1195,7 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec if opts.replicaObjectInfo == nil && (opts.ReplicaLockReconcile || !opts.ReplicationRequest) { opts.replicaObjectInfo = z.replicaObjectInfo } + opts.replicaContentGuard = z.guardReplicaContent oi, err := z.serverPools[idx].PutObject(ctx, bucket, object, data, opts) if err == nil && opts.ReplicaLockReconcile && !opts.DataMovement { err = z.retireReplicaCopies(ctx, bucket, object, idx, oi) @@ -1749,6 +1750,9 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec return objInfo, err } + if !z.SinglePool() { + dstOpts.replicaContentGuard = z.guardReplicaContent + } if !z.SinglePool() && (dstOpts.ReplicaLockReconcile || !dstOpts.ReplicationRequest) { dstOpts.replicaObjectInfo = z.replicaObjectInfo if dstOpts.ReplicaLockReconcile && !dstOpts.DataMovement { @@ -1799,6 +1803,7 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec ReplicaLockReconcile: dstOpts.ReplicaLockReconcile, DataMovement: dstOpts.DataMovement, replicaObjectInfo: dstOpts.replicaObjectInfo, + replicaContentGuard: dstOpts.replicaContentGuard, } return z.serverPools[poolIdx].PutObject(ctx, dstBucket, dstObject, srcInfo.PutObjReader, putOpts) @@ -2524,6 +2529,7 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket opts.replicaObjectInfo = z.replicaObjectInfo } + opts.replicaContentGuard = z.guardReplicaContent for idx, pool := range z.serverPools { if z.IsSuspended(idx) { continue diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go index 2ad350169..ffd21230d 100644 --- a/cmd/object-api-interface.go +++ b/cmd/object-api-interface.go @@ -122,6 +122,8 @@ type ObjectOptions struct { // Pools-layer version resolver; the caller holds the shared object lock. replicaObjectInfo GetObjectInfoFn + // Verifies protected content in every pool before a replica commits. + replicaContentGuard replicaContentGuardFn // RefuseLockedOverwrite refuses, under the destination's write lock, a // write that would replace a stored version Object Lock protects: the diff --git a/cmd/replica-content.go b/cmd/replica-content.go new file mode 100644 index 000000000..542130c14 --- /dev/null +++ b/cmd/replica-content.go @@ -0,0 +1,205 @@ +// Copyright (c) 2026 Feng Ruohang +// +// This file is part of Silo 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. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "bytes" + "context" + "io" + "net/http" + + "github.com/minio/minio/internal/crypto" + "github.com/minio/minio/internal/hash/sha256" + xioutil "github.com/minio/minio/internal/ioutil" +) + +type replicaContentGuardFn func(context.Context, string, string, ObjectInfo, func() (*GetObjectReader, error)) error + +// verifyLockedReplica compares the actual data, never a sender's ETag or +// checksum assertion. A peer may retransmit a protected version to update its +// metadata, but may not replace its content. Call under the object write lock. +func verifyLockedReplica(ctx context.Context, incoming, stored ObjectInfo, openIncoming, openStored func() (*GetObjectReader, error)) error { + if !enforceRetentionForDeletion(ctx, stored) { + return nil + } + locked := ObjectLocked{Bucket: stored.Bucket, Object: stored.Name, VersionID: stored.VersionID} + if storedObjectLockState(stored.UserDefined).complianceWouldWeaken(storedObjectLockState(incoming.UserDefined)) { + return locked + } + inSSEC, oldSSEC := crypto.SSEC.IsEncrypted(incoming.UserDefined), crypto.SSEC.IsEncrypted(stored.UserDefined) + if inSSEC != oldSSEC { + return locked + } + if inSSEC { + inSize, inErr := incoming.GetActualSize() + oldSize, oldErr := stored.GetActualSize() + if inErr != nil || oldErr != nil || inSize != oldSize { + return locked + } + // Without the customer key, compare ciphertext and its interpretation. + // Equal ciphertext with a changed seal or multipart framing is not the + // same object, even if the supplied ETag and size are unchanged. + for _, key := range []string{ + crypto.MetaIV, crypto.MetaAlgorithm, crypto.MetaSealedKeySSEC, + ReservedMetadataPrefix + "compression", + } { + if incoming.UserDefined[key] != stored.UserDefined[key] { + return locked + } + } + if crypto.IsMultiPart(incoming.UserDefined) != crypto.IsMultiPart(stored.UserDefined) || incoming.isMultipart() != stored.isMultipart() { + return locked + } + if crypto.IsMultiPart(incoming.UserDefined) { + if len(incoming.Parts) != len(stored.Parts) { + return locked + } + for i, p := range incoming.Parts { + old := stored.Parts[i] + if p.Number != old.Number || p.Size != old.Size { + return locked + } + } + } + } + in, err := openIncoming() + if err != nil { + return err + } + defer in.Close() + old, err := openStored() + if err != nil { + return err + } + defer old.Close() + ih, oh := sha256.New(), sha256.New() + inSize, err := io.Copy(ih, in) + if err != nil { + return err + } + oldSize, err := io.Copy(oh, old) + if err != nil { + return err + } + if inSize != oldSize || !bytes.Equal(ih.Sum(nil), oh.Sum(nil)) { + return locked + } + return nil +} + +func (er erasureObjects) guardReplicaContent(ctx context.Context, bucket, object string, incoming ObjectInfo, openIncoming func() (*GetObjectReader, error)) error { + opts := ObjectOptions{VersionID: incoming.VersionID, NoLock: true, NoAuditLog: true, ReplicationRequest: true} + if opts.VersionID == "" { + opts.VersionID = nullVersionID + } + stored, err := er.getObjectInfo(ctx, bucket, object, opts) + if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isErrMethodNotAllowed(err) { + return nil + } + if err != nil { + return err + } + return verifyLockedReplica(ctx, incoming, stored, openIncoming, func() (*GetObjectReader, error) { + return er.GetObjectNInfo(ctx, bucket, object, nil, http.Header{}, opts) + }) +} + +func (z *erasureServerPools) guardReplicaContent(ctx context.Context, bucket, object string, incoming ObjectInfo, openIncoming func() (*GetObjectReader, error)) error { + opts := ObjectOptions{VersionID: incoming.VersionID, NoLock: true, NoAuditLog: true, ReplicationRequest: true} + if opts.VersionID == "" { + opts.VersionID = nullVersionID + } + copies, err := z.objectPoolInfos(ctx, bucket, object, opts) + if isErrObjectNotFound(err) || isErrVersionNotFound(err) { + return nil + } + if err != nil { + return err + } + for _, copy := range copies { + if err := verifyLockedReplica(ctx, incoming, copy.ObjInfo, openIncoming, func() (*GetObjectReader, error) { + return z.serverPools[copy.Index].GetObjectNInfo(ctx, bucket, object, nil, http.Header{}, opts) + }); err != nil { + return err + } + } + return nil +} + +// checkReplicaContent reads the staged version before rename commits it. Its +// data is decoded/decrypted just like a GET (raw ciphertext for SSE-C), so this +// covers multipart, compression and destination SSE-S3/KMS re-encryption. +// disks and metadata are in erasure shard order, as passed to renameData. +func (er erasureObjects) checkReplicaContent(ctx context.Context, bucket, object, stageBucket, stageObject string, metadata []FileInfo, disks []StorageAPI, opts ObjectOptions) error { + if (!opts.ReplicaLockReconcile && !opts.ReplicationRequest) || opts.DataMovement { + return nil + } + var fi FileInfo + for i, disk := range disks { + if disk != nil { + fi = metadata[i] + break + } + } + incoming := fi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended) + guard := opts.replicaContentGuard + if guard == nil { + guard = er.guardReplicaContent + } + return guard(ctx, bucket, object, incoming, func() (*GetObjectReader, error) { + // A marker-only peer write does not reconcile the stored protection. + // It therefore cannot replace a protected version, even with equal data. + if !opts.ReplicaLockReconcile { + return nil, ObjectLocked{Bucket: bucket, Object: decodeDirObject(object), VersionID: incoming.VersionID} + } + if incoming.IsCompressed() { + // A replica may assert the source's logical size. A compressed GET + // limits its output to that size, so verify it against the lengths + // recorded while consuming the uploaded parts before comparing GETs. + // Otherwise a forged zero size could hide a non-empty replacement. + var actual int64 + for _, part := range incoming.Parts { + actual += part.ActualSize + } + declared, err := incoming.GetActualSize() + if err != nil || declared != actual { + return nil, ObjectLocked{Bucket: bucket, Object: decodeDirObject(object), VersionID: incoming.VersionID} + } + } + readOpts := ObjectOptions{NoDecryption: crypto.SSEC.IsEncrypted(incoming.UserDefined)} + fn, off, length, err := NewGetObjectReader(nil, incoming, readOpts, http.Header{}) + if err != nil { + return nil, err + } + // The ordinary reader expects original drive order, not shard order. + readDisks, readMeta := make([]StorageAPI, len(disks)), make([]FileInfo, len(metadata)) + for i, shard := range fi.Erasure.Distribution { + readDisks[i], readMeta[i] = disks[shard-1], metadata[shard-1] + readMeta[i].Erasure.Index = shard + } + pr, pw := xioutil.WaitPipe() + go func() { + pw.CloseWithError(er.getObjectWithFileInfo(ctx, stageBucket, stageObject, off, length, pw, fi, readMeta, readDisks)) + }() + reader, err := fn(pr, http.Header{}, func() { pr.CloseWithError(nil) }) + if err != nil { + pr.CloseWithError(err) + } + return reader, err + }) +} diff --git a/cmd/replica-content_test.go b/cmd/replica-content_test.go new file mode 100644 index 000000000..5291f4b1e --- /dev/null +++ b/cmd/replica-content_test.go @@ -0,0 +1,314 @@ +// Copyright (c) 2026 Feng Ruohang +// +// This file is part of Silo 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. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "bytes" + "encoding/xml" + "fmt" + "io" + "maps" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "strings" + "testing" + "time" + + "github.com/minio/minio/internal/auth" + "github.com/minio/minio/internal/crypto" + xhttp "github.com/minio/minio/internal/http" + "github.com/minio/minio/internal/kms" +) + +func TestAPIReplicaPreservesProtectedContent(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + objAPITest: func(obj ObjectLayer, backend, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) { + oldTLS, oldKMS := globalIsTLS, GlobalKMS + globalIsTLS, GlobalKMS = true, kms.NewStub("replica-content") + defer func() { globalIsTLS, GlobalKMS = oldTLS, oldKMS }() + restoreCompression := disableCompression() + defer restoreCompression() + request := func(method, url string, data []byte, headers map[string]string) *httptest.ResponseRecorder { + t.Helper() + req, err := newTestSignedRequestV4(method, url, int64(len(data)), bytes.NewReader(data), cred.AccessKey, cred.SecretKey, headers) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + return rec + } + for _, encryption := range []string{"plain", "sse-s3", "sse-kms", "sse-c", "plain-compressed", "sse-s3-compressed"} { + restore := func() {} + if strings.HasSuffix(encryption, "-compressed") { + restore = setCopyChecksumCompression(true) + } + for _, multipart := range []bool{false, true} { + for _, protection := range []string{"COMPLIANCE", "LEGALHOLD", "GOVERNANCE"} { + key := fmt.Sprintf("protected/%s/%v/%s.txt", encryption, multipart, protection) + enc := map[string]string{} + switch encryption { + case "sse-s3", "sse-s3-compressed": + enc[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionAES + case "sse-kms": + enc[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionKMS + enc[xhttp.AmzServerSideEncryptionKmsID] = "replica-content" + case "sse-c": + enc = ssecKeyHeaders(bytes.Repeat([]byte{0x37}, 32), false) + } + headers := maps.Clone(enc) + maps.Copy(headers, batchLockMeta(protection)) + // Both inline and disk-backed writes are exercised. + body := bytes.Repeat([]byte("immutable content "), 20000) + switch protection { + case "LEGALHOLD": + body = body[:64] + case "GOVERNANCE": + body = nil + } + if !multipart { + putCopyChecksumSource(t, router, cred, bucket, key, body, headers) + } else { + initRec := request(http.MethodPost, getNewMultipartURL("", bucket, key), nil, headers) + if initRec.Code != http.StatusOK { + t.Fatalf("source initiate: %d %s", initRec.Code, initRec.Body.String()) + } + var init InitiateMultipartUploadResponse + if err := xml.Unmarshal(initRec.Body.Bytes(), &init); err != nil { + t.Fatal(err) + } + part := request(http.MethodPut, getPutObjectPartURL("", bucket, key, init.UploadID, "1"), body, enc) + if part.Code != http.StatusOK { + t.Fatalf("source part: %d %s", part.Code, part.Body.String()) + } + complete, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{{PartNumber: 1, ETag: canonicalizeETag(part.Header()[xhttp.ETag][0])}}}) + if err != nil { + t.Fatal(err) + } + rec := request(http.MethodPost, getCompleteMultipartUploadURL("", bucket, key, init.UploadID), complete, enc) + if rec.Code != http.StatusOK { + t.Fatalf("source complete: %d %s", rec.Code, rec.Body.String()) + } + } + gr, err := obj.GetObjectNInfo(t.Context(), bucket, key, nil, http.Header{}, ObjectOptions{ReplicationRequest: true}) + if err != nil { + t.Fatal(err) + } + src := gr.ObjInfo + wire, err := io.ReadAll(gr) + gr.Close() + if err != nil { + t.Fatal(err) + } + for _, altered := range []bool{true, false} { + data := bytes.Clone(wire) + if altered { + if len(data) == 0 { + data = []byte("x") + } else { + data[len(data)-1] ^= 0x20 + } + } + opts, _, err := putReplicationOpts(t.Context(), "", src) + if err != nil { + t.Fatal(err) + } + hdrs := map[string]string{} + for k, vs := range opts.Header() { + if len(vs) > 0 { + hdrs[k] = vs[0] + } + } + hdrs[xhttp.MinIOSourceReplicationRequest] = "true" + hdrs[xhttp.AmzBucketReplicationStatus] = "REPLICA" + // Forge the old ETag and size along with a newer tag revision: + // the body comparison must still reject different bytes. + hdrs[xhttp.MinIOSourceETag] = src.ETag + hdrs[xhttp.MinIOSourceTaggingTimestamp] = UTCNow().Add(time.Minute).Format(time.RFC3339Nano) + hdrs[xhttp.AmzObjectTagging] = "replica=updated" + var rec *httptest.ResponseRecorder + if !multipart { + rec = request(http.MethodPut, getPutObjectURL("", bucket, key)+"?versionId="+src.VersionID, data, hdrs) + } else { + initRec := request(http.MethodPost, getNewMultipartURL("", bucket, key)+"&versionId="+src.VersionID, nil, hdrs) + if initRec.Code != http.StatusOK { + t.Fatalf("%s %s initiate: %d %s", backend, key, initRec.Code, initRec.Body.String()) + } + var init InitiateMultipartUploadResponse + if err := xml.Unmarshal(initRec.Body.Bytes(), &init); err != nil { + t.Fatal(err) + } + partHdrs := map[string]string{xhttp.MinIOSourceReplicationRequest: "true"} + partRec := request(http.MethodPut, getPutObjectPartURL("", bucket, key, init.UploadID, "1"), data, partHdrs) + if partRec.Code != http.StatusOK && (!altered || !strings.Contains(partRec.Body.String(), "XMinioObjectTampered")) { + t.Fatalf("%s %s part: %d %s", backend, key, partRec.Code, partRec.Body.String()) + } + if partRec.Code != http.StatusOK { + rec = partRec + } else { + complete, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{{PartNumber: 1, ETag: canonicalizeETag(partRec.Header()[xhttp.ETag][0])}}}) + if err != nil { + t.Fatal(err) + } + actual, err := src.GetActualSize() + if err != nil { + t.Fatal(err) + } + partHdrs[xhttp.MinIOReplicationActualObjectSize] = strconv.FormatInt(actual, 10) + rec = request(http.MethodPost, getCompleteMultipartUploadURL("", bucket, key, init.UploadID), complete, partHdrs) + } + } + if altered { + if rec.Code < 400 || !strings.Contains(rec.Body.String(), "WORM") && !strings.Contains(rec.Body.String(), "XMinioObjectTampered") { + t.Fatalf("%s %s: different content not refused: %d %s", backend, key, rec.Code, rec.Body.String()) + } + } else if rec.Code != http.StatusOK { + t.Fatalf("%s %s: same content refused: %d %s", backend, key, rec.Code, rec.Body.String()) + } + getHdrs := map[string]string{} + if encryption == "sse-c" { + getHdrs = enc + } + got := request(http.MethodGet, getGetObjectURL("", bucket, key)+"?versionId="+src.VersionID, nil, getHdrs) + if got.Code != http.StatusOK || !bytes.Equal(got.Body.Bytes(), body) { + t.Fatalf("%s %s: protected content changed or became unreadable: %d", backend, key, got.Code) + } + } + } + } + restore() + } + }, + }) +} + +func TestPoolsReplicaProtectedContentAcrossPools(t *testing.T) { + z, bucket := consistencyPools(t) + const key = "protected-across-pools" + id := mustGetUUID() + stamp := UTCNow().Add(-time.Hour) + putConsistencyObject(t, z, bucket, key, 0, "original", ObjectOptions{Versioned: true, VersionID: id, MTime: stamp}) + putConsistencyObject(t, z, bucket, key, 1, "original", ObjectOptions{Versioned: true, VersionID: id, MTime: stamp, UserDefined: batchLockMeta("COMPLIANCE")}) + _, err := z.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, strings.NewReader("tampered"), 8, "", ""), ObjectOptions{Versioned: true, VersionID: id, ReplicationRequest: true, ReplicaLockReconcile: true, UserDefined: map[string]string{}}) + if _, ok := err.(ObjectLocked); !ok { + t.Fatalf("protected pool copy allowed overwrite: %v", err) + } + for i, pool := range z.serverPools { + gr, err := pool.GetObjectNInfo(t.Context(), bucket, key, nil, http.Header{}, ObjectOptions{VersionID: id}) + if err != nil { + t.Fatal(err) + } + data, err := io.ReadAll(gr) + gr.Close() + if err != nil || string(data) != "original" { + t.Fatalf("pool %d: changed protected data: %q %v", i, data, err) + } + } +} + +func TestLockedSSECReplicaRejectsChangedSeal(t *testing.T) { + stored := ObjectInfo{Bucket: "bucket", Name: "key", Size: 0, UserDefined: batchLockMeta("LEGALHOLD")} + stored.UserDefined[crypto.MetaSealedKeySSEC] = "stored-seal" + incoming := stored + incoming.UserDefined = maps.Clone(stored.UserDefined) + incoming.UserDefined[crypto.MetaSealedKeySSEC] = "changed-seal" + open := func() (*GetObjectReader, error) { + t.Fatal("changed seal must be rejected before reading") + return nil, nil + } + if err := verifyLockedReplica(t.Context(), incoming, stored, open, open); err == nil { + t.Fatal("changed SSE-C seal accepted") + } +} + +func TestAPIReplicaCopyPreservesProtectedVersion(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + objAPITest: func(obj ObjectLayer, backend, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) { + const key = "copy-protected-version" + original := batchPutVersion(t, obj, bucket, key, batchLockMeta("COMPLIANCE")) + data := []byte("different source version") + source, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), ObjectOptions{Versioned: true}) + if err != nil { + t.Fatal(err) + } + headers := map[string]string{ + xhttp.AmzCopySource: url.QueryEscape("/"+bucket+"/"+key) + "?versionId=" + source.VersionID, + xhttp.MinIOSourceReplicationRequest: "true", xhttp.AmzBucketReplicationStatus: "REPLICA", + xhttp.AmzMetadataDirective: replaceDirective, + } + req, err := newTestSignedRequestV4(http.MethodPut, getCopyObjectURL("", bucket, key)+"?versionId="+original.VersionID, 0, nil, cred.AccessKey, cred.SecretKey, headers) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code < 400 || !strings.Contains(rec.Body.String(), "WORM") { + t.Fatalf("%s: copy replaced protected version: %d %s", backend, rec.Code, rec.Body.String()) + } + gr, err := obj.GetObjectNInfo(t.Context(), bucket, key, nil, http.Header{}, ObjectOptions{VersionID: original.VersionID}) + if err != nil { + t.Fatal(err) + } + got, err := io.ReadAll(gr) + gr.Close() + if err != nil || string(got) != "batch lock "+key { + t.Fatalf("%s: protected body changed: %q %v", backend, got, err) + } + }, + }) +} + +func TestAPIReplicaCannotWeakenCompliance(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + objAPITest: func(obj ObjectLayer, backend, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) { + const key = "compliance-removal" + oi := batchPutVersion(t, obj, bucket, key, batchLockMeta("COMPLIANCE")) + until := oi.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] + for _, copyRequest := range []bool{false, true} { + headers := map[string]string{xhttp.MinIOSourceReplicationRequest: "true", xhttp.AmzBucketReplicationStatus: "REPLICA", xhttp.MinIOSourceObjectRetentionTimestamp: UTCNow().Add(time.Hour).Format(time.RFC3339Nano)} + body := []byte("batch lock " + key) + if copyRequest { + headers[xhttp.AmzCopySource] = url.QueryEscape("/"+bucket+"/"+key) + "?versionId=" + oi.VersionID + headers[xhttp.AmzMetadataDirective] = replaceDirective + body = nil + } + req, err := newTestSignedRequestV4(http.MethodPut, getPutObjectURL("", bucket, key)+"?versionId="+oi.VersionID, int64(len(body)), bytes.NewReader(body), cred.AccessKey, cred.SecretKey, headers) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("%s copy=%v: %d %s", backend, copyRequest, rec.Code, rec.Body.String()) + } + stored, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: oi.VersionID}) + if err != nil || stored.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] != "COMPLIANCE" || stored.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] != until { + t.Fatalf("%s copy=%v: a replica cleared COMPLIANCE: %+v %v", backend, copyRequest, stored.UserDefined, err) + } + } + }, + }) +} diff --git a/cmd/replication-signed-headers_test.go b/cmd/replication-signed-headers_test.go new file mode 100644 index 000000000..7e02a6e13 --- /dev/null +++ b/cmd/replication-signed-headers_test.go @@ -0,0 +1,94 @@ +// Copyright (c) 2026 Feng Ruohang +// +// This file is part of Silo 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. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "bytes" + "net/http" + "net/http/httptest" + "testing" + + "github.com/minio/minio/internal/auth" + xhttp "github.com/minio/minio/internal/http" +) + +func TestReplicationControlsRequireSignedHeaders(t *testing.T) { + for _, header := range replicationRequestHeaders { + t.Run(header, func(t *testing.T) { + req, err := newTestSignedRequestV4(http.MethodGet, "http://localhost/bucket/key", 0, nil, "access-key", "secret-key", nil) + if err != nil { + t.Fatal(err) + } + // Membership matters even for a present-empty internal header. + req.Header.Set(header, "") + if replicationHeadersSigned(req) { + t.Fatal("unsigned control granted replication trust") + } + if err := signRequestV4(req, "access-key", "secret-key"); err != nil { + t.Fatal(err) + } + if !replicationHeadersSigned(req) { + t.Fatal("signed control rejected") + } + }) + } + for _, authorization := range []string{"", "AWS access-key:signature", "Credential=access-key"} { + req := httptest.NewRequest(http.MethodGet, "/bucket/key", nil) + req.Header.Set(xhttp.Authorization, authorization) + req.Header.Set(xhttp.MinIOSourceReplicationRequest, "true") + if replicationHeadersSigned(req) { + t.Fatalf("non-SigV4 authorization granted trust: %q", authorization) + } + } + req := httptest.NewRequest(http.MethodGet, "http://localhost/bucket/key", nil) + req.Header.Set(xhttp.MinIOSourceReplicationRequest, "true") + presignBoundaryRequest(t, req, UTCNow(), []string{"host", "x-minio-source-replication-request"}, auth.Credentials{AccessKey: "access-key", SecretKey: "secret-key"}) + getRequestAuthType(req) + if replicationHeadersSigned(req) { + t.Fatal("presigned request granted peer trust") + } +} + +func TestAPIUnsignedReplicationControlsAreIgnored(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, endpoints: []string{"PutObject"}, + objAPITest: func(obj ObjectLayer, backend, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) { + for _, signedMarker := range []bool{false, true} { + headers := map[string]string{} + if signedMarker { + headers[xhttp.MinIOSourceReplicationRequest] = "true" + } + body := []byte("replication controls are authenticated") + req, err := newTestSignedRequestV4(http.MethodPut, getPutObjectURL("", bucket, "key"), int64(len(body)), bytes.NewReader(body), cred.AccessKey, cred.SecretKey, headers) + if err != nil { + t.Fatal(err) + } + req.Header.Set(xhttp.MinIOSourceReplicationRequest, "true") + req.Header.Set(xhttp.MinIOSourceMTime, "2100-01-01T00:00:00Z") + req.Header.Set(xhttp.MinIOSourceETag, "forged-source-etag") + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + oi, err := obj.GetObjectInfo(t.Context(), bucket, "key", ObjectOptions{}) + if rec.Code != http.StatusOK || err != nil || oi.ModTime.Year() == 2100 || oi.ETag == "forged-source-etag" { + t.Fatalf("%s signedMarker=%v: unsigned replication state was accepted: HTTP=%d info=%+v err=%v", backend, signedMarker, rec.Code, oi, err) + } + } + }, + }) +} diff --git a/cmd/replication-ssec-retransmit_test.go b/cmd/replication-ssec-retransmit_test.go index c845b4668..50114d0e7 100644 --- a/cmd/replication-ssec-retransmit_test.go +++ b/cmd/replication-ssec-retransmit_test.go @@ -1850,6 +1850,12 @@ func testReplicaLockReconcileNullVersion(obj ObjectLayer, instanceType, bucketNa _ http.Handler, _ auth.Credentials, t *testing.T, ) { ctx := t.Context() + // A protected retransmit must preserve both the source ciphertext and + // its seal. Generate it once for the stored null version and its retry. + 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) + } // completeNullMPU runs an SSE-C replica multipart upload addressing the null // version (VersionSuspended) carrying uploadLock, through the reconcile. @@ -1861,13 +1867,6 @@ 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(ciphertext.Bytes()), int64(ciphertext.Len()), "", ""), ObjectOptions{}) if err != nil { @@ -1884,8 +1883,9 @@ func testReplicaLockReconcileNullVersion(obj ObjectLayer, instanceType, bucketNa 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, "", ""), + mustGetPutObjReader(t, bytes.NewReader(ciphertext.Bytes()), int64(ciphertext.Len()), "", ""), ObjectOptions{VersionSuspended: true, MTime: time.Date(2026, 9, 3, 10, 0, 0, 0, time.UTC), UserDefined: map[string]string{ + crypto.MetaSealedKeySSEC: "dummy-sealed-key", strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000, }}); err != nil { diff --git a/cmd/replication-trust.go b/cmd/replication-trust.go index 0f8cd5320..c7ba9a674 100644 --- a/cmd/replication-trust.go +++ b/cmd/replication-trust.go @@ -13,6 +13,8 @@ package cmd import ( "context" "net/http" + "slices" + "strings" objectreplication "github.com/minio/minio/internal/bucket/replication" xhttp "github.com/minio/minio/internal/http" @@ -62,7 +64,7 @@ func isReplicaTrusted(ctx context.Context) bool { // grant must not turn client-controlled internal headers into trusted state. func replicationPermissionAllowed(ctx context.Context, r *http.Request, bucket, object string, action policy.Action) bool { reqInfo := logger.GetReqInfo(ctx) - if reqInfo == nil || reqInfo.Cred.AccessKey == "" { + if reqInfo == nil || reqInfo.Cred.AccessKey == "" || !replicationHeadersSigned(r) { return false } reqInfo.BucketName = bucket @@ -70,6 +72,25 @@ func replicationPermissionAllowed(ctx context.Context, r *http.Request, bucket, return authorizeRequest(ctx, r, action) == ErrNone } +// replicationHeadersSigned binds every internal control to the authenticated +// header signature. Query grants and SigV2 requests cannot confer peer trust. +// Call only after the request's signature has been verified. +func replicationHeadersSigned(r *http.Request) bool { + if !isRequestSignatureV4(r) || isRequestPresignedSignatureV4(r) { + return false + } + sig, errCode := parseSignV4(r.Header.Get(xhttp.Authorization), globalSite.Region(), serviceS3) + if errCode != ErrNone { + return false + } + for _, name := range replicationRequestHeaders { + if _, present := r.Header[http.CanonicalHeaderKey(name)]; present && !slices.Contains(sig.SignedHeaders, strings.ToLower(name)) { + return false + } + } + return true +} + // evaluateReplicationTrust decides whether a request may carry replication // semantics for the given action. A request that declares itself a replica // without holding the replication permission is rejected. trusted reports that diff --git a/cmd/s3-zip-handlers.go b/cmd/s3-zip-handlers.go index b7fe426bf..62ae864e1 100644 --- a/cmd/s3-zip-handlers.go +++ b/cmd/s3-zip-handlers.go @@ -144,13 +144,7 @@ func (api objectAPIHandlers) getObjectInArchiveFileHandler(ctx context.Context, zipInfo := zipObjInfo.ArchiveInfo(r.Header) if len(zipInfo) == 0 { - opts.EncryptFn, err = zipObjInfo.metadataEncryptFn(r.Header) - if err != nil { - writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) - return - } - - zipInfo, err = updateObjectMetadataWithZipInfo(ctx, objectAPI, bucket, zipPath, opts) + zipInfo, _, err = readObjectZIPInfo(ctx, objectAPI, bucket, zipPath, opts) } if err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) @@ -245,8 +239,7 @@ func listObjectsV2InArchive(ctx context.Context, objectAPI ObjectLayer, bucket, zipInfo := zipObjInfo.ArchiveInfo(h) if len(zipInfo) == 0 { - // Always update the latest version - zipInfo, err = updateObjectMetadataWithZipInfo(ctx, objectAPI, bucket, zipPath, ObjectOptions{}) + zipInfo, _, err = readObjectZIPInfo(ctx, objectAPI, bucket, zipPath, ObjectOptions{}) } if err != nil { return ListObjectsV2Info{}, err @@ -437,12 +430,7 @@ func (api objectAPIHandlers) headObjectInArchiveFileHandler(ctx context.Context, zipInfo := zipObjInfo.ArchiveInfo(r.Header) if len(zipInfo) == 0 { - opts.EncryptFn, err = zipObjInfo.metadataEncryptFn(r.Header) - if err != nil { - writeErrorResponseHeadersOnly(w, toAPIError(ctx, err)) - return - } - zipInfo, err = updateObjectMetadataWithZipInfo(ctx, objectAPI, bucket, zipPath, opts) + zipInfo, _, err = readObjectZIPInfo(ctx, objectAPI, bucket, zipPath, opts) } if err != nil { writeErrorResponseHeadersOnly(w, toAPIError(ctx, err)) @@ -482,15 +470,22 @@ func (api objectAPIHandlers) headObjectInArchiveFileHandler(ctx context.Context, w.WriteHeader(http.StatusOK) } -// Update the passed zip object metadata with the zip contents info, file name, modtime, size, etc. -// The returned zip index will de decrypted. -func updateObjectMetadataWithZipInfo(ctx context.Context, objectAPI ObjectLayer, bucket, object string, opts ObjectOptions) ([]byte, error) { +// readObjectZIPInfo builds an index without modifying the object. A GET, HEAD +// or archive listing must remain read-only even when no index was stored. +func readObjectZIPInfo(ctx context.Context, objectAPI ObjectLayer, bucket, object string, opts ObjectOptions) ([]byte, ObjectInfo, error) { files, srcInfo, err := getFilesListFromZIPObject(ctx, objectAPI, bucket, object, opts) if err != nil { - return nil, err + return nil, srcInfo, err } files.OptimizeSize() zipInfo, err := files.Serialize() + return zipInfo, srcInfo, err +} + +// updateObjectMetadataWithZipInfo persists an index only for upload operations +// that explicitly request extraction. The returned index is decrypted. +func updateObjectMetadataWithZipInfo(ctx context.Context, objectAPI ObjectLayer, bucket, object string, opts ObjectOptions) ([]byte, error) { + zipInfo, srcInfo, err := readObjectZIPInfo(ctx, objectAPI, bucket, object, opts) if err != nil { return nil, err } @@ -500,7 +495,6 @@ func updateObjectMetadataWithZipInfo(ctx context.Context, objectAPI ObjectLayer, at = archiveTypeEnc zipInfoStr = string(opts.EncryptFn(archiveTypeEnc, zipInfo)) } - srcInfo.UserDefined[archiveTypeMetadataKey] = at popts := ObjectOptions{ MTime: srcInfo.ModTime, VersionID: srcInfo.VersionID, diff --git a/cmd/s3-zip-readonly_test.go b/cmd/s3-zip-readonly_test.go new file mode 100644 index 000000000..22c2b5e78 --- /dev/null +++ b/cmd/s3-zip-readonly_test.go @@ -0,0 +1,109 @@ +// Copyright (c) 2026 Feng Ruohang +// +// This file is part of Silo 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. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "archive/zip" + "bytes" + "fmt" + "maps" + "net/http" + "net/http/httptest" + "testing" + + "github.com/minio/minio/internal/auth" +) + +func TestAPIArchiveReadsDoNotPersistMetadata(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + objAPITest: func(obj ObjectLayer, backend, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) { + var archive bytes.Buffer + zw := zip.NewWriter(&archive) + file, err := zw.Create("entry.txt") + if err != nil { + t.Fatal(err) + } + if _, err = file.Write([]byte("archive content")); err != nil { + t.Fatal(err) + } + if err = zw.Close(); err != nil { + t.Fatal(err) + } + reader := newObjectAttributesAuthzUser(t, backend, bucket, `"s3:GetObject","s3:ListBucket"`) + policy := fmt.Appendf(nil, `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":["s3:GetObject","s3:ListBucket"],"Resource":["arn:aws:s3:::%s","arn:aws:s3:::%s/*"]}]}`, bucket, bucket) + if _, err = globalBucketMetadataSys.Update(t.Context(), bucket, bucketPolicyConfig, policy); err != nil { + t.Fatal(err) + } + for _, anonymous := range []bool{false, true} { + for _, method := range []string{http.MethodGet, http.MethodHead, "LIST"} { + key := fmt.Sprintf("archive-%v-%s.zip", anonymous, method) + before, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewReader(archive.Bytes()), int64(archive.Len()), "", ""), ObjectOptions{Versioned: true, UserDefined: batchLockMeta("COMPLIANCE")}) + if err != nil { + t.Fatal(err) + } + before, err = obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: before.VersionID}) + if err != nil { + t.Fatal(err) + } + url := getGetObjectURL("", bucket, key+"/entry.txt") + verb := method + if method == "LIST" { + verb = http.MethodGet + url = "/" + bucket + "?list-type=2&prefix=" + key + "/" + } + req, err := newTestSignedRequestV4(verb, url, 0, nil, reader.AccessKey, reader.SecretKey, map[string]string{xMinIOExtract: "true"}) + if err != nil { + t.Fatal(err) + } + if anonymous { + req.Header.Del("Authorization") + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("%s anonymous=%v %s: %d %s", backend, anonymous, method, rec.Code, rec.Body.String()) + } + if method == http.MethodGet && rec.Body.String() != "archive content" { + t.Fatalf("wrong archive entry: %s", rec.Body.String()) + } + if method == "LIST" && !bytes.Contains(rec.Body.Bytes(), []byte("entry.txt")) { + t.Fatal("archive listing omitted entry") + } + after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: before.VersionID}) + if err != nil || !maps.Equal(before.UserDefined, after.UserDefined) || !after.ModTime.Equal(before.ModTime) || after.ETag != before.ETag { + t.Fatalf("%s anonymous=%v %s changed stored metadata: before=%v after=%v err=%v", backend, anonymous, method, before.UserDefined, after.UserDefined, err) + } + } + } + // Explicit extraction during an authorized upload still builds and + // persists the index for subsequent reads. + req, err := newTestSignedRequestV4(http.MethodPut, getPutObjectURL("", bucket, "indexed.zip"), int64(archive.Len()), bytes.NewReader(archive.Bytes()), cred.AccessKey, cred.SecretKey, map[string]string{xMinIOExtract: "true"}) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + oi, err := obj.GetObjectInfo(t.Context(), bucket, "indexed.zip", ObjectOptions{}) + if rec.Code != http.StatusOK || err != nil || len(oi.ArchiveInfo(nil)) == 0 { + t.Fatalf("explicit extraction lost index: %d %s %v", rec.Code, rec.Body.String(), err) + } + }, + }) +} diff --git a/cmd/streaming-auth-boundary_test.go b/cmd/streaming-auth-boundary_test.go new file mode 100644 index 000000000..2daeb43bb --- /dev/null +++ b/cmd/streaming-auth-boundary_test.go @@ -0,0 +1,114 @@ +// Copyright (c) 2026 Feng Ruohang +// +// This file is part of Silo 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. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "bytes" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/minio/minio/internal/auth" + xhttp "github.com/minio/minio/internal/http" +) + +func TestAPIStreamingAuthenticationBoundaries(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, endpoints: []string{"PutObject"}, + objAPITest: func(obj ObjectLayer, backend, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) { + data := []byte("authenticated streaming payload") + for _, mode := range []string{"valid", "missing-algorithm", "wrong-algorithm", "wrong-signature"} { + req, err := newStreamingUnsignedTrailerRequest(http.MethodPut, getPutObjectURL("", bucket, mode), data, UTCNow()) + if err != nil { + t.Fatal(err) + } + if err := signRequestV4(req, cred.AccessKey, cred.SecretKey); err != nil { + t.Fatal(err) + } + authorization := req.Header.Get(xhttp.Authorization) + switch mode { + case "missing-algorithm": + authorization = strings.TrimPrefix(authorization, signV4Algorithm+" ") + case "wrong-algorithm": + authorization = strings.Replace(authorization, signV4Algorithm, "INVALID", 1) + case "wrong-signature": + prefix, _, found := strings.Cut(authorization, "Signature=") + if !found { + t.Fatal("signed request has no signature") + } + authorization = prefix + "Signature=" + strings.Repeat("0", 64) + } + req.Header.Set(xhttp.Authorization, authorization) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + _, stored := obj.GetObjectInfo(t.Context(), bucket, mode, ObjectOptions{}) + if mode == "valid" { + if rec.Code != http.StatusOK || stored != nil { + t.Fatalf("%s %s: %d %s, object: %v", backend, mode, rec.Code, rec.Body.String(), stored) + } + } else if rec.Code < 400 || stored == nil { + t.Errorf("%s %s: malformed authentication wrote an object: %d %s", backend, mode, rec.Code, rec.Body.String()) + } + } + for _, trailers := range []bool{false, true} { + req, err := newTestStreamingSignedRequest(http.MethodPut, getPutObjectURL("", bucket, "tagged"), int64(len(data)), 64*1024, bytes.NewReader(data), cred.AccessKey, cred.SecretKey) + if err != nil { + t.Fatal(err) + } + if trailers { + req.Header.Set(xhttp.AmzContentSha256, streamingContentSHA256Trailer) + req.Header.Set(awsTrailerHeader, "x-amz-checksum-crc32") + if err := signRequestV4(req, cred.AccessKey, cred.SecretKey); err != nil { + t.Fatal(err) + } + } + getRequestAuthType(req) + if _, _, _, _, code := calculateSeedSignature(req, trailers); code != ErrNone { + t.Fatalf("%s trailers=%v: valid seed rejected: %v", backend, trailers, code) + } + req.Header.Set(xhttp.AmzObjectTagging, "unsigned=injected") + if _, code := newSignV4ChunkedReader(req, trailers); code != ErrUnsignedHeaders { + t.Errorf("%s trailers=%v: unsigned tags accepted: %v", backend, trailers, code) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if _, err := obj.GetObjectInfo(t.Context(), bucket, "tagged", ObjectOptions{}); rec.Code < 400 || err == nil { + t.Errorf("%s trailers=%v: injected metadata stored: %d", backend, trailers, rec.Code) + } + } + // A public bucket can still use the unsigned trailer decoder without + // claiming credentials; authorization remains the handler's job. + req, err := newStreamingUnsignedTrailerRequest(http.MethodPut, "/public/key", data, UTCNow()) + if err != nil { + t.Fatal(err) + } + rd, code := newUnsignedV4ChunkedReader(req, true) + if code != ErrNone { + t.Fatal(code) + } + got, err := io.ReadAll(rd) + rd.Close() + if err != nil || !bytes.Equal(got, data) { + t.Fatalf("anonymous decoder: %q %v", got, err) + } + }, + }) +} diff --git a/cmd/streaming-signature-v4.go b/cmd/streaming-signature-v4.go index 702714be0..eef479a0c 100644 --- a/cmd/streaming-signature-v4.go +++ b/cmd/streaming-signature-v4.go @@ -127,6 +127,9 @@ func calculateSeedSignature(r *http.Request, trailers bool) (cred auth.Credentia if errCode != ErrNone { return cred, "", "", time.Time{}, errCode } + if errCode := checkUnsignedHeaders(extractedSignedHeaders, r); errCode != ErrNone { + return cred, "", "", time.Time{}, errCode + } cred, _, errCode = checkKeyValid(r, signV4Values.Credential.accessKey) if errCode != ErrNone { diff --git a/cmd/streaming-v4-unsigned.go b/cmd/streaming-v4-unsigned.go index 04014c5c5..11c524a3a 100644 --- a/cmd/streaming-v4-unsigned.go +++ b/cmd/streaming-v4-unsigned.go @@ -24,6 +24,8 @@ import ( "io" "net/http" "strings" + + xhttp "github.com/minio/minio/internal/http" ) // newUnsignedV4ChunkedReader returns a new s3UnsignedChunkedReader that translates @@ -42,8 +44,10 @@ func newUnsignedV4ChunkedReader(req *http.Request, trailer bool) (io.ReadCloser, if isRequestPresignedSignatureV4(req) { return nil, ErrSignatureVersionNotSupported } - if isRequestSignatureV4(req) { - if errCode := reqSignatureV4Verify(req, globalSite.Region(), serviceS3); errCode != ErrNone { + // Any claimed header credential must pass the complete SigV4 parser. The + // streaming auth classification does not require the algorithm prefix. + if req.Header.Get(xhttp.Authorization) != "" { + if errCode := doesSignatureMatch(unsignedPayloadTrailer, req, globalSite.Region(), serviceS3); errCode != ErrNone { return nil, errCode } }