mirror of
https://github.com/pgsty/minio.git
synced 2026-10-17 15:16:00 +03:00
Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2ca0f9961a | |||
| d94b3b3c92 | |||
| 86c5a92eef | |||
| cdc3b8cb10 | |||
| 0d3ba5d4cb | |||
| 71a1631f13 | |||
| 14042b5aab | |||
| ba6e8c8d69 | |||
| a54d0c4354 | |||
| 00d4a206f7 |
@@ -102,6 +102,11 @@ jobs:
|
|||||||
- name: Run tag replication tests under race detector
|
- name: Run tag replication tests under race detector
|
||||||
run: go test -race ./cmd -run '^TestAPITagging' -count=1 -timeout=5m
|
run: go test -race ./cmd -run '^TestAPITagging' -count=1 -timeout=5m
|
||||||
|
|
||||||
|
- name: Run Object Lock authorization and overwrite tests under race detector
|
||||||
|
env:
|
||||||
|
MINIO_API_REQUESTS_MAX: "10000"
|
||||||
|
run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m
|
||||||
|
|
||||||
crosscompile:
|
crosscompile:
|
||||||
name: Cross Compile
|
name: Cross Compile
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
|
|||||||
@@ -70,6 +70,12 @@ jobs:
|
|||||||
buildscripts/verify-rebrand.sh
|
buildscripts/verify-rebrand.sh
|
||||||
dockerscripts/docker-entrypoint_test.sh
|
dockerscripts/docker-entrypoint_test.sh
|
||||||
|
|
||||||
|
# Keep this gate aligned with the Object Lock race regression in go.yml.
|
||||||
|
- name: Gate release on Object Lock authorization and overwrite tests under race detector
|
||||||
|
env:
|
||||||
|
MINIO_API_REQUESTS_MAX: "10000"
|
||||||
|
run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m
|
||||||
|
|
||||||
- name: Compute release variables
|
- name: Compute release variables
|
||||||
env:
|
env:
|
||||||
# Passed through the environment, never interpolated into the script
|
# Passed through the environment, never interpolated into the script
|
||||||
|
|||||||
@@ -332,7 +332,12 @@ func checkPutObjectLockAllowed(ctx context.Context, rq *http.Request, bucket, ob
|
|||||||
}
|
}
|
||||||
|
|
||||||
if opts.VersionID != "" && !replicaTrusted {
|
if opts.VersionID != "" && !replicaTrusted {
|
||||||
if objInfo, err := getObjectInfoFn(ctx, bucket, object, opts); err == nil {
|
objInfo, err := getObjectInfoFn(ctx, bucket, object, opts)
|
||||||
|
if err != nil {
|
||||||
|
if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) && !errors.As(err, &MethodNotAllowed{}) {
|
||||||
|
return mode, retainDate, legalHold, toAPIErrorCode(ctx, err)
|
||||||
|
}
|
||||||
|
} else {
|
||||||
r := objectlock.GetObjectRetentionMeta(objInfo.UserDefined)
|
r := objectlock.GetObjectRetentionMeta(objInfo.UserDefined)
|
||||||
t, err := objectlock.UTCNowNTP()
|
t, err := objectlock.UTCNowNTP()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -25,9 +25,12 @@ import (
|
|||||||
"encoding/base64"
|
"encoding/base64"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
"net/url"
|
"net/url"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -300,3 +303,244 @@ func TestPoolsObjectLockOverwrite(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type planReviewGatedBody struct {
|
||||||
|
io.ReadCloser
|
||||||
|
first sync.Once
|
||||||
|
ready chan struct{}
|
||||||
|
release chan struct{}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (g *planReviewGatedBody) Read(p []byte) (int, error) {
|
||||||
|
g.first.Do(func() { close(g.ready); <-g.release })
|
||||||
|
return g.ReadCloser.Read(p)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPlanReviewSlowDirectoryPut(t *testing.T) {
|
||||||
|
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
|
||||||
|
t: t,
|
||||||
|
makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
|
||||||
|
objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
|
||||||
|
writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
|
||||||
|
t.Run(instance, func(t *testing.T) {
|
||||||
|
key := "slow-body-directory/"
|
||||||
|
before, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("original"), 8, "", ""), ObjectOptions{Versioned: true, VersionID: nullVersionID})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
payload := []byte("replacement-body")
|
||||||
|
req, err := newTestSignedRequestV4(http.MethodPut, makeTestTargetURL("", bucket, key, nil), int64(len(payload)), bytes.NewReader(payload), writer.AccessKey, writer.SecretKey, nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
gate := &planReviewGatedBody{ReadCloser: req.Body, ready: make(chan struct{}), release: make(chan struct{})}
|
||||||
|
req.Body = gate
|
||||||
|
var once sync.Once
|
||||||
|
release := func() { once.Do(func() { close(gate.release) }) }
|
||||||
|
defer release()
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
done := make(chan struct{})
|
||||||
|
go func() { router.ServeHTTP(rec, req); close(done) }()
|
||||||
|
select {
|
||||||
|
case <-gate.ready:
|
||||||
|
case <-done:
|
||||||
|
t.Fatalf("PUT finished before body: %d %s", rec.Code, rec.Body.String())
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
t.Fatal("PUT did not reach body read")
|
||||||
|
}
|
||||||
|
locked := make(chan int, 1)
|
||||||
|
go func() {
|
||||||
|
locked <- putRetention(t, router, root, bucket, key, nullVersionID, "COMPLIANCE", UTCNow().Add(365*24*time.Hour), false)
|
||||||
|
}()
|
||||||
|
select {
|
||||||
|
case status := <-locked:
|
||||||
|
if status != http.StatusOK {
|
||||||
|
t.Fatalf("set lock while body paused: %d", status)
|
||||||
|
}
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
t.Fatal("retention update blocked behind paused body")
|
||||||
|
}
|
||||||
|
release()
|
||||||
|
select {
|
||||||
|
case <-done:
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
t.Fatal("PUT did not finish")
|
||||||
|
}
|
||||||
|
after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if rec.Code != http.StatusBadRequest || after.ETag != before.ETag || objectlock.GetObjectRetentionMeta(after.UserDefined).Mode != objectlock.RetCompliance {
|
||||||
|
t.Fatal("paused ordinary PUT overwrote the COMPLIANCE null version after the lock was acknowledged")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
type planReviewReadFault struct {
|
||||||
|
ObjectLayer
|
||||||
|
key string
|
||||||
|
fired bool
|
||||||
|
putCalled bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *planReviewReadFault) GetObjectInfo(ctx context.Context, bucket, key string, opts ObjectOptions) (ObjectInfo, error) {
|
||||||
|
if key == p.key && opts.VersionID == nullVersionID && !p.fired {
|
||||||
|
p.fired = true
|
||||||
|
return ObjectInfo{}, InsufficientReadQuorum{}
|
||||||
|
}
|
||||||
|
return p.ObjectLayer.GetObjectInfo(ctx, bucket, key, opts)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *planReviewReadFault) PutObject(ctx context.Context, bucket, key string, data *PutObjReader, opts ObjectOptions) (ObjectInfo, error) {
|
||||||
|
if key == p.key {
|
||||||
|
p.putCalled = true
|
||||||
|
}
|
||||||
|
return p.ObjectLayer.PutObject(ctx, bucket, key, data, opts)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPlanReviewPreflightReadFailure(t *testing.T) {
|
||||||
|
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
|
||||||
|
t: t,
|
||||||
|
makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
|
||||||
|
objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, _ auth.Credentials, t *testing.T) {
|
||||||
|
writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
|
||||||
|
t.Run(instance, func(t *testing.T) {
|
||||||
|
key := "locked-directory-read-fault/"
|
||||||
|
_, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("original"), 8, "", ""), ObjectOptions{
|
||||||
|
Versioned: true, VersionID: nullVersionID, ReplicationRequest: true,
|
||||||
|
UserDefined: map[string]string{
|
||||||
|
strings.ToLower(xhttp.AmzObjectLockMode): "COMPLIANCE",
|
||||||
|
strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): amztime.ISO8601Format(UTCNow().Add(365 * 24 * time.Hour)),
|
||||||
|
},
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
initial, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
fault := &planReviewReadFault{ObjectLayer: obj, key: key}
|
||||||
|
setObjectLayer(fault)
|
||||||
|
defer setObjectLayer(obj)
|
||||||
|
rec := retentionAuthzRequest(t, router, writer, http.MethodPut, makeTestTargetURL("", bucket, key, nil), []byte("replacement"), nil)
|
||||||
|
if !fault.fired {
|
||||||
|
t.Fatal("preflight fault was not injected")
|
||||||
|
}
|
||||||
|
if fault.putCalled {
|
||||||
|
t.Fatal("preflight read error allowed PutObject to proceed to storage layer")
|
||||||
|
}
|
||||||
|
after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if rec.Code < 400 {
|
||||||
|
t.Errorf("metadata read quorum error was treated as absence: status=%d", rec.Code)
|
||||||
|
}
|
||||||
|
retentionOverwriteUnchanged(t, initial, after)
|
||||||
|
})
|
||||||
|
},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPlanReviewLegacyMPUCompliance(t *testing.T) {
|
||||||
|
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
|
||||||
|
t: t,
|
||||||
|
makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
|
||||||
|
objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
|
||||||
|
writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
|
||||||
|
t.Run(instance, func(t *testing.T) {
|
||||||
|
key := "legacy-mpu-compliance"
|
||||||
|
initial, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("original"), 8, "", ""), ObjectOptions{Versioned: true})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
up, err := obj.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{Versioned: true, VersionID: initial.VersionID})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
payload := []byte("replacement-part")
|
||||||
|
part, err := obj.PutObjectPart(t.Context(), bucket, key, up.UploadID, 1, mustGetPutObjReader(t, bytes.NewReader(payload), int64(len(payload)), "", ""), ObjectOptions{})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if status := putRetention(t, router, root, bucket, key, initial.VersionID, "COMPLIANCE", UTCNow().Add(365*24*time.Hour), false); status != http.StatusOK {
|
||||||
|
t.Fatalf("set compliance: %d", status)
|
||||||
|
}
|
||||||
|
before, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: initial.VersionID})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
body := fmt.Appendf(nil, `<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>"%s"</ETag></Part></CompleteMultipartUpload>`, part.ETag)
|
||||||
|
rec := retentionAuthzRequest(t, router, writer, http.MethodPost, makeTestTargetURL("", bucket, key, url.Values{"uploadId": {up.UploadID}}), body, nil)
|
||||||
|
after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: initial.VersionID})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if rec.Code != http.StatusForbidden {
|
||||||
|
t.Errorf("legacy upload replaced locked UUID: status=%d", rec.Code)
|
||||||
|
}
|
||||||
|
retentionOverwriteUnchanged(t, before, after)
|
||||||
|
})
|
||||||
|
},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// A recovered upload retains an authentic server-generated version ID while a
|
||||||
|
// surviving replica reintroduces its already-completed version and lock.
|
||||||
|
// Only the fixture uses replication privileges; completion uses PutObject only.
|
||||||
|
func TestRecoveredMultipartUploadAfterReplica(t *testing.T) {
|
||||||
|
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
|
||||||
|
t: t,
|
||||||
|
makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
|
||||||
|
objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
|
||||||
|
writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
|
||||||
|
t.Run(instance, func(t *testing.T) {
|
||||||
|
z := obj.(*erasureServerPools)
|
||||||
|
key := "recovered-generated-upload"
|
||||||
|
up, err := obj.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{Versioned: true})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
er := z.serverPools[0].getHashedSet(key)
|
||||||
|
fi, _, err := er.checkUploadIDExists(t.Context(), bucket, key, up.UploadID, true)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
_, err = obj.PutObject(t.Context(), bucket, key,
|
||||||
|
mustGetPutObjReader(t, bytes.NewBufferString("protected-original"), 18, "", ""),
|
||||||
|
ObjectOptions{Versioned: true, VersionID: fi.VersionID, ReplicationRequest: true})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if status := putRetention(t, router, root, bucket, key, fi.VersionID, "COMPLIANCE", UTCNow().Add(365*24*time.Hour), false); status != http.StatusOK {
|
||||||
|
t.Fatalf("setting compliance lock: %d", status)
|
||||||
|
}
|
||||||
|
before, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: fi.VersionID})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
partRec := retentionAuthzRequest(t, router, writer, http.MethodPut,
|
||||||
|
makeTestTargetURL("", bucket, key, url.Values{"uploadId": {up.UploadID}, "partNumber": {"1"}}),
|
||||||
|
[]byte("attacker-replacement"), nil)
|
||||||
|
if partRec.Code != http.StatusOK {
|
||||||
|
t.Fatalf("upload part: %d %s", partRec.Code, partRec.Body.String())
|
||||||
|
}
|
||||||
|
etag := multipartConditionResponseETag(partRec)
|
||||||
|
body := []byte(multipartConditionCompleteBody([]CompletePart{{PartNumber: 1, ETag: etag}}))
|
||||||
|
rec := retentionAuthzRequest(t, router, writer, http.MethodPost,
|
||||||
|
makeTestTargetURL("", bucket, key, url.Values{"uploadId": {up.UploadID}}), body, nil)
|
||||||
|
after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: fi.VersionID})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if rec.Code != http.StatusForbidden {
|
||||||
|
t.Errorf("recovered upload bypassed existing UUID guard: status=%d", rec.Code)
|
||||||
|
}
|
||||||
|
retentionOverwriteUnchanged(t, before, after)
|
||||||
|
})
|
||||||
|
},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|||||||
@@ -290,6 +290,9 @@ func (s *storageRESTServer) NSScannerHandler(ctx context.Context, params *nsScan
|
|||||||
for update := range updates {
|
for update := range updates {
|
||||||
resp := storageNSScannerRPC.NewResponse()
|
resp := storageNSScannerRPC.NewResponse()
|
||||||
resp.Update = &update
|
resp.Update = &update
|
||||||
|
// Pooled responses may carry fields from a previous
|
||||||
|
// scan, always clear the fields not being set.
|
||||||
|
resp.Final = nil
|
||||||
out <- resp
|
out <- resp
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
@@ -301,6 +304,7 @@ func (s *storageRESTServer) NSScannerHandler(ctx context.Context, params *nsScan
|
|||||||
// Send final response.
|
// Send final response.
|
||||||
resp := storageNSScannerRPC.NewResponse()
|
resp := storageNSScannerRPC.NewResponse()
|
||||||
resp.Final = &ui
|
resp.Final = &ui
|
||||||
|
resp.Update = nil
|
||||||
out <- resp
|
out <- resp
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,94 @@
|
|||||||
|
// Copyright (c) 2026 Julien Laurenceau and Feng Ruohang
|
||||||
|
//
|
||||||
|
// This file is part of Silo Object Storage.
|
||||||
|
//
|
||||||
|
// 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 <http://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
package cmd
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/minio/madmin-go/v3"
|
||||||
|
"github.com/minio/minio/internal/grid"
|
||||||
|
)
|
||||||
|
|
||||||
|
type scannerResponseTestDisk struct {
|
||||||
|
StorageAPI
|
||||||
|
}
|
||||||
|
|
||||||
|
func (scannerResponseTestDisk) NSScanner(_ context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, _ madmin.HealScanMode, _ func() bool) (dataUsageCache, error) {
|
||||||
|
defer close(updates)
|
||||||
|
updates <- dataUsageEntry{Size: int64(cache.Info.NextCycle)}
|
||||||
|
cache.Info.NextCycle++
|
||||||
|
return cache, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNSScannerHandlerClearsStaleResponses(t *testing.T) {
|
||||||
|
endpoint, err := NewEndpoint(t.TempDir())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
globalLocalDrivesMu.Lock()
|
||||||
|
oldDrives, oldSets := globalLocalDrivesMap, globalLocalSetDrives
|
||||||
|
globalLocalDrivesMap = map[string]StorageAPI{endpoint.String(): scannerResponseTestDisk{}}
|
||||||
|
globalLocalSetDrives = nil
|
||||||
|
globalLocalDrivesMu.Unlock()
|
||||||
|
t.Cleanup(func() {
|
||||||
|
globalLocalDrivesMu.Lock()
|
||||||
|
globalLocalDrivesMap, globalLocalSetDrives = oldDrives, oldSets
|
||||||
|
globalLocalDrivesMu.Unlock()
|
||||||
|
})
|
||||||
|
|
||||||
|
oldRPC := storageNSScannerRPC
|
||||||
|
t.Cleanup(func() { storageNSScannerRPC = oldRPC })
|
||||||
|
// Every new allocation is dirty too: the check must not depend on
|
||||||
|
// sync.Pool returning a previously used object.
|
||||||
|
storageNSScannerRPC = grid.NewStream[*nsScannerOptions, grid.NoPayload, *nsScannerResp](grid.HandlerNSScanner,
|
||||||
|
func() *nsScannerOptions { return &nsScannerOptions{} }, nil,
|
||||||
|
func() *nsScannerResp {
|
||||||
|
return &nsScannerResp{Update: &dataUsageEntry{Size: -1}, Final: &dataUsageCache{}}
|
||||||
|
})
|
||||||
|
server := storageRESTServer{endpoint: endpoint}
|
||||||
|
for cycle := uint32(1); cycle <= 3; cycle++ {
|
||||||
|
out := make(chan *nsScannerResp, 2)
|
||||||
|
params := &nsScannerOptions{Cache: &dataUsageCache{Info: dataUsageCacheInfo{Name: "bucket", NextCycle: cycle}}}
|
||||||
|
if err := server.NSScannerHandler(t.Context(), params, out); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(out) != 2 {
|
||||||
|
t.Fatalf("cycle %d: got %d responses, want an update and a final cache", cycle, len(out))
|
||||||
|
}
|
||||||
|
for i := 0; i < 2; i++ {
|
||||||
|
resp := <-out
|
||||||
|
wire, err := resp.MarshalMsg(nil)
|
||||||
|
storageNSScannerRPC.PutResponse(resp)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
var decoded nsScannerResp
|
||||||
|
if _, err := decoded.UnmarshalMsg(wire); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if i == 0 {
|
||||||
|
if decoded.Update == nil || decoded.Final != nil || decoded.Update.Size != int64(cycle) {
|
||||||
|
t.Errorf("cycle %d: invalid update response: %+v", cycle, decoded)
|
||||||
|
}
|
||||||
|
} else if decoded.Update != nil || decoded.Final == nil || decoded.Final.Info.NextCycle != cycle+1 {
|
||||||
|
t.Errorf("cycle %d: invalid final response: %+v", cycle, decoded)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -40,6 +40,10 @@ func (x *xlMetaV2VersionHeader) unmarshalV(v uint8, bts []byte) (o []byte, err e
|
|||||||
|
|
||||||
// unmarshalV1 decodes version 1, never released.
|
// unmarshalV1 decodes version 1, never released.
|
||||||
func (x *xlMetaV2VersionHeader) unmarshalV1(bts []byte) (o []byte, err error) {
|
func (x *xlMetaV2VersionHeader) unmarshalV1(bts []byte) (o []byte, err error) {
|
||||||
|
// Fields absent from the V1 wire format must be reset since the
|
||||||
|
// header may be reused across decodes.
|
||||||
|
x.Signature = [4]byte{}
|
||||||
|
x.EcN, x.EcM = 0, 0
|
||||||
var zb0001 uint32
|
var zb0001 uint32
|
||||||
zb0001, bts, err = msgp.ReadArrayHeaderBytes(bts)
|
zb0001, bts, err = msgp.ReadArrayHeaderBytes(bts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -134,7 +138,7 @@ type xlMetaV2VersionHeaderV2 struct {
|
|||||||
|
|
||||||
// UnmarshalMsg implements msgp.Unmarshaler
|
// UnmarshalMsg implements msgp.Unmarshaler
|
||||||
func (z *xlMetaV2VersionHeaderV2) UnmarshalMsg(bts []byte) (o []byte, err error) {
|
func (z *xlMetaV2VersionHeaderV2) UnmarshalMsg(bts []byte) (o []byte, err error) {
|
||||||
z.EcN, z.EcN = 0, 0
|
z.EcN, z.EcM = 0, 0
|
||||||
var zb0001 uint32
|
var zb0001 uint32
|
||||||
zb0001, bts, err = msgp.ReadArrayHeaderBytes(bts)
|
zb0001, bts, err = msgp.ReadArrayHeaderBytes(bts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -184,7 +188,7 @@ func (z *xlMetaV2VersionHeaderV2) UnmarshalMsg(bts []byte) (o []byte, err error)
|
|||||||
|
|
||||||
// DecodeMsg implements msgp.Decodable
|
// DecodeMsg implements msgp.Decodable
|
||||||
func (z *xlMetaV2VersionHeaderV2) DecodeMsg(dc *msgp.Reader) (err error) {
|
func (z *xlMetaV2VersionHeaderV2) DecodeMsg(dc *msgp.Reader) (err error) {
|
||||||
z.EcN, z.EcN = 0, 0
|
z.EcN, z.EcM = 0, 0
|
||||||
var zb0001 uint32
|
var zb0001 uint32
|
||||||
zb0001, err = dc.ReadArrayHeader()
|
zb0001, err = dc.ReadArrayHeader()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -38,6 +38,7 @@ import (
|
|||||||
"github.com/klauspost/compress/zstd"
|
"github.com/klauspost/compress/zstd"
|
||||||
"github.com/minio/minio/internal/bucket/lifecycle"
|
"github.com/minio/minio/internal/bucket/lifecycle"
|
||||||
xhttp "github.com/minio/minio/internal/http"
|
xhttp "github.com/minio/minio/internal/http"
|
||||||
|
"github.com/tinylib/msgp/msgp"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestReadXLMetaNoData(t *testing.T) {
|
func TestReadXLMetaNoData(t *testing.T) {
|
||||||
@@ -1190,3 +1191,92 @@ func BenchmarkToFileInfoNoParts(b *testing.B) {
|
|||||||
func BenchmarkToFileInfoWithParts(b *testing.B) {
|
func BenchmarkToFileInfoWithParts(b *testing.B) {
|
||||||
benchmarkManyPartsOptionally(b, true)
|
benchmarkManyPartsOptionally(b, true)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestXLMetaV2HeaderLegacyUnmarshalResetsEc checks that decoding legacy
|
||||||
|
// version headers, which carry no erasure metadata fields, cannot leak
|
||||||
|
// stale EcN/EcM values from a previously decoded object into the same
|
||||||
|
// structure.
|
||||||
|
func TestXLMetaV2HeaderLegacyUnmarshalResetsEc(t *testing.T) {
|
||||||
|
base := xlMetaV2VersionHeader{
|
||||||
|
VersionID: [16]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16},
|
||||||
|
ModTime: time.Now().Unix(),
|
||||||
|
Signature: [4]byte{1, 2, 3, 4},
|
||||||
|
Type: ObjectType,
|
||||||
|
Flags: xlFlagInlineData,
|
||||||
|
}
|
||||||
|
|
||||||
|
// Legacy V2 header wire format: 5 fields, no EcN/EcM.
|
||||||
|
var legacy []byte
|
||||||
|
legacy = msgp.AppendArrayHeader(legacy, 5)
|
||||||
|
legacy = msgp.AppendBytes(legacy, base.VersionID[:])
|
||||||
|
legacy = msgp.AppendInt64(legacy, base.ModTime)
|
||||||
|
legacy = msgp.AppendBytes(legacy, base.Signature[:])
|
||||||
|
legacy = msgp.AppendUint8(legacy, uint8(base.Type))
|
||||||
|
legacy = msgp.AppendUint8(legacy, uint8(base.Flags))
|
||||||
|
|
||||||
|
for _, decoder := range []string{"UnmarshalMsg", "DecodeMsg", "loadIndexed"} {
|
||||||
|
t.Run(decoder, func(t *testing.T) {
|
||||||
|
header := base
|
||||||
|
header.EcN, header.EcM = 4, 8
|
||||||
|
x2 := xlMetaV2VersionHeaderV2{xlMetaV2VersionHeader: &header}
|
||||||
|
var err error
|
||||||
|
switch decoder {
|
||||||
|
case "UnmarshalMsg":
|
||||||
|
_, err = x2.UnmarshalMsg(legacy)
|
||||||
|
case "DecodeMsg":
|
||||||
|
err = x2.DecodeMsg(msgp.NewReader(bytes.NewReader(legacy)))
|
||||||
|
case "loadIndexed":
|
||||||
|
indexed := msgp.AppendUint(nil, 2)
|
||||||
|
indexed = msgp.AppendUint(indexed, xlMetaVersion)
|
||||||
|
indexed = msgp.AppendInt(indexed, 1)
|
||||||
|
indexed = msgp.AppendBytes(indexed, legacy)
|
||||||
|
indexed = msgp.AppendBytes(indexed, nil)
|
||||||
|
xl := xlMetaV2{versions: []xlMetaV2ShallowVersion{{header: header}}}
|
||||||
|
err = xl.loadIndexed(indexed, nil)
|
||||||
|
header = xl.versions[0].header
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if header != base {
|
||||||
|
t.Fatalf("legacy header: got %+v, want %+v", header, base)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// V1 headers (4 fields, never released) must zero the fields too.
|
||||||
|
var v1 []byte
|
||||||
|
v1 = msgp.AppendArrayHeader(v1, 4)
|
||||||
|
v1 = msgp.AppendBytes(v1, base.VersionID[:])
|
||||||
|
v1 = msgp.AppendInt64(v1, base.ModTime)
|
||||||
|
v1 = msgp.AppendUint8(v1, uint8(base.Type))
|
||||||
|
v1 = msgp.AppendUint8(v1, uint8(base.Flags))
|
||||||
|
|
||||||
|
header := base
|
||||||
|
header.EcN, header.EcM = 4, 8
|
||||||
|
header.Signature = [4]byte{9, 9, 9, 9}
|
||||||
|
if _, err := header.unmarshalV1(v1); err != nil {
|
||||||
|
t.Fatalf("legacy V1 header unmarshal failed: %v", err)
|
||||||
|
}
|
||||||
|
if header.EcN != 0 || header.EcM != 0 {
|
||||||
|
t.Fatalf("stale erasure metadata leaked on V1 decode: EcN %d, EcM %d", header.EcN, header.EcM)
|
||||||
|
}
|
||||||
|
if header.Signature != ([4]byte{}) {
|
||||||
|
t.Fatalf("stale signature leaked on V1 decode: %v", header.Signature)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Current headers must keep their erasure metadata.
|
||||||
|
current := base
|
||||||
|
current.EcN, current.EcM = 4, 8
|
||||||
|
wire, err := current.MarshalMsg(nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("header marshal failed: %v", err)
|
||||||
|
}
|
||||||
|
decoded := base
|
||||||
|
if _, err := decoded.UnmarshalMsg(wire); err != nil {
|
||||||
|
t.Fatalf("header unmarshal failed: %v", err)
|
||||||
|
}
|
||||||
|
if decoded.EcN != 4 || decoded.EcM != 8 {
|
||||||
|
t.Fatalf("erasure metadata lost on current decode: EcN %d, EcM %d", decoded.EcN, decoded.EcM)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user