diff --git a/cmd/batch-expire-purge-authz_test.go b/cmd/batch-expire-purge-authz_test.go new file mode 100644 index 000000000..20f6815af --- /dev/null +++ b/cmd/batch-expire-purge-authz_test.go @@ -0,0 +1,475 @@ +// Copyright (c) 2026 PGSTY +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "bytes" + "context" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/minio/minio/internal/auth" + xhttp "github.com/minio/minio/internal/http" + "github.com/pgsty/silo-pkg/v3/policy" +) + +// A batch expiration that keeps no versions purges each key it matched under +// the purge's lock. What it authorizes for the submitter, and checks for +// Object Lock, are the versions the purge deletes then, not the versions the +// walk saw before. + +// batchPurgeHook runs change once, after the walk of a job and before its +// first purge of all versions. +type batchPurgeHook struct { + ObjectLayer + once sync.Once + change func() +} + +func (h *batchPurgeHook) DeleteObject(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) { + if opts.DeleteAllVersions { + h.once.Do(h.change) + } + return h.ObjectLayer.DeleteObject(ctx, bucket, object, opts) +} + +func (f *batchAuthzFixture) purgeYAML(prefix string) string { + return fmt.Sprintf(`expire: + apiVersion: v1 + bucket: %s + prefix: %q + rules: + - type: object + purge: + retainVersions: 0 +`, f.bucket, prefix) +} + +// runChanged submits a purge of prefix as cred and runs it, with change made +// between its walk and its purge, and returns the job's final state. +func (f *batchAuthzFixture) runChanged(t *testing.T, cred auth.Credentials, prefix string, change func()) *batchJobInfo { + t.Helper() + rec, job := f.submit(t, cred, f.purgeYAML(prefix)) + if job == nil { + t.Fatalf("submission status %d: %s", rec.Code, rec.Body.String()) + } + hook := &batchPurgeHook{ObjectLayer: f.obj, change: change} + if err := job.Expire.Start(f.ctx, hook, *job); err != nil { + t.Logf("start: %v", err) + } + return f.jobInfo(t, job) +} + +// putOlder writes a version of key older than every version it has, as a +// replicated or pulled version can be, with metadata meta. +func (f *batchAuthzFixture) putOlder(t *testing.T, key, versionID string, meta map[string]string) { + t.Helper() + data := []byte("older " + key) + _, err := f.obj.PutObject(f.ctx, f.bucket, key, mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), + ObjectOptions{Versioned: true, VersionID: versionID, MTime: UTCNow().Add(-time.Hour), UserDefined: meta}) + mustIAM(t, err) +} + +func batchPurgeVersionIDs(t *testing.T, f *batchAuthzFixture, key string) []string { + t.Helper() + var ids []string + marker, vmarker := "", "" + for { + res, err := f.obj.ListObjectVersions(f.ctx, f.bucket, key, marker, vmarker, "", 1000) + mustIAM(t, err) + for _, o := range res.Objects { + if o.Name == key { + ids = append(ids, o.VersionID) + } + } + if !res.IsTruncated { + return ids + } + marker, vmarker = res.NextMarker, res.NextVersionIDMarker + } +} + +// A version that the submitter may not expire, because of its tags or its +// version ID, is refused although the walk did not see it so: the job fails +// and the key keeps all its versions. +func TestBatchExpirePurgeAuthorizesCurrentVersions(t *testing.T) { + for _, tc := range []struct { + name string + action string // the action a conditional Deny denies + deny string // its condition + // change returns the version IDs that change made. + change func(t *testing.T, f *batchAuthzFixture, key, older string) []string + }{ + { + name: "tagged-after-walk", + action: "s3:GetObjectVersion", + deny: `{"StringEquals":{"s3:ExistingObjectTag/keep":["yes"]}}`, + change: func(t *testing.T, f *batchAuthzFixture, key, older string) []string { + _, err := f.obj.PutObjectTags(f.ctx, f.bucket, key, "keep=yes", ObjectOptions{VersionID: older}) + mustIAM(t, err) + return nil + }, + }, + { + name: "version-after-walk", + action: "s3:DeleteObjectVersion", + deny: `{"StringEquals":{"s3:versionid":["%s"]}}`, + change: func(t *testing.T, f *batchAuthzFixture, key, older string) []string { + f.putOlder(t, key, batchPurgeDeniedVersion, nil) + return []string{batchPurgeDeniedVersion} + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + f := newBatchAuthzFixture(t) + deny := tc.deny + if strings.Contains(deny, "%s") { + deny = fmt.Sprintf(deny, batchPurgeDeniedVersion) + } + batchAuthzPolicy(f.ctx, t, f.sys, "batch-purge", batchAuthzFullAccess+`, +{"Effect":"Deny","Action":["`+tc.action+`"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":`+deny+`}`) + cred := iamAdminUser(f.ctx, t, f.sys, "batch-purge-user", "batch-purge") + const key = "purge/key" + older := f.putVersions(t, key) + want := batchPurgeVersionIDs(t, f, key) + ri := f.runChanged(t, cred, "purge/", func() { + want = append(want, tc.change(t, f, key, older)...) + }) + for _, id := range want { + if !f.versionExists(t, key, id) { + t.Errorf("version %s was purged although the submitter may not delete it", id) + } + } + if !ri.Failed || ri.Complete || ri.FailureReason == "" { + t.Errorf("job complete=%v failed=%v reason=%q, want a refused job", ri.Complete, ri.Failed, ri.FailureReason) + } + if ri.Objects != 0 || ri.KeysRefused != 0 { + t.Errorf("job counts objects=%d refused=%d", ri.Objects, ri.KeysRefused) + } + }) + } +} + +const batchPurgeDeniedVersion = "6c4b2f8e-1d3a-4e5b-9c7d-0a1b2c3d4e5f" + +// A version that Object Lock protects refuses the whole key, also when it +// was added after the walk. The job completes with the key counted as +// refused, not failed. +func TestBatchExpirePurgeRefusesLockedAfterWalk(t *testing.T) { + for _, kind := range batchLockKinds { + t.Run(kind, func(t *testing.T) { + f := newBatchAuthzFixture(t) + enableBucketObjectLock(t, f.bucket) + batchAuthzPolicy(f.ctx, t, f.sys, "batch-full", batchAuthzFullAccess) + cred := iamAdminUser(f.ctx, t, f.sys, "batch-user", "batch-full") + const key = "purge/locked" + f.putVersions(t, key) + want := batchPurgeVersionIDs(t, f, key) + locked := mustGetUUID() + ri := f.runChanged(t, cred, "purge/", func() { + f.putOlder(t, key, locked, batchLockMeta(kind)) + want = append(want, locked) + }) + for _, id := range want { + if !f.versionExists(t, key, id) { + t.Errorf("version %s of a key with a locked version was purged", id) + } + } + if !ri.Complete || ri.Failed || ri.KeysRefused != 1 || ri.ObjectsFailed != 0 || ri.Objects != 0 { + t.Errorf("job complete=%v failed=%v refused=%d failed objects=%d objects=%d, want complete with 1 refused key", + ri.Complete, ri.Failed, ri.KeysRefused, ri.ObjectsFailed, ri.Objects) + } + }) + } +} + +// Controls: a purge by root and by an authorized user removes every version, +// including one added after the walk. +func TestBatchExpirePurgeAuthorizedControls(t *testing.T) { + for _, root := range []bool{true, false} { + t.Run(fmt.Sprintf("root=%v", root), func(t *testing.T) { + f := newBatchAuthzFixture(t) + cred := globalActiveCred + if !root { + batchAuthzPolicy(f.ctx, t, f.sys, "batch-full", batchAuthzFullAccess) + cred = iamAdminUser(f.ctx, t, f.sys, "batch-user", "batch-full") + } + const key = "purge/key" + f.putVersions(t, key) + added := mustGetUUID() + ri := f.runChanged(t, cred, "purge/", func() { + f.putOlder(t, key, added, map[string]string{xhttp.AmzObjectTagging: "keep=yes"}) + }) + if ids := batchPurgeVersionIDs(t, f, key); len(ids) != 0 { + t.Errorf("an authorized purge kept %v", ids) + } + if !ri.Complete || ri.Failed || ri.Objects != 3 || ri.KeysRefused != 0 { + t.Errorf("job complete=%v failed=%v objects=%d refused=%d, want complete with 3 objects", + ri.Complete, ri.Failed, ri.Objects, ri.KeysRefused) + } + }) + } +} + +// Cases found in the sixth review round. + +// batchPurgeBudgetChange runs change once the first purge stopped at its +// budget after removing versions. +type batchPurgeBudgetChange struct { + ObjectLayer + change func() + calls int +} + +func (h *batchPurgeBudgetChange) DeleteObject(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) { + oi, err := h.ObjectLayer.DeleteObject(ctx, bucket, object, opts) + if opts.DeleteAllVersions { + h.calls++ + if h.calls == 1 && errors.Is(err, errDeleteAllBudget) && opts.DeleteAllResult.Removed > 0 { + h.change() + } + } + return oi, err +} + +// A purge that continues after its budget authorizes the versions again, as +// they are then; the refusal and the versions already removed persist +// across another start. +func TestBatchExpirePurgeBudgetReauthorizes(t *testing.T) { + f := newBatchAuthzFixture(t) + const key = "purge/budget" + oldest := batchPutVersion(t, f.obj, f.bucket, key, nil) + middle := batchPutVersion(t, f.obj, f.bucket, key, nil) + latest := batchPutVersion(t, f.obj, f.bucket, key, nil) + batchAuthzPolicy(f.ctx, t, f.sys, "review6-budget-deny", batchAuthzFullAccess+`, +{"Effect":"Deny","Action":["s3:GetObjectVersion"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"StringEquals":{"s3:ExistingObjectTag/keep":"yes"}}}`) + cred := iamAdminUser(f.ctx, t, f.sys, "review6-budget-user", "review6-budget-deny") + rec, job := f.submit(t, cred, f.purgeYAML("purge/")) + if job == nil { + t.Fatalf("submit: %d %s", rec.Code, rec.Body.String()) + } + oldBudget := deleteAllBudget + deleteAllBudget.versions = 1 + t.Cleanup(func() { deleteAllBudget = oldBudget }) + hook := &batchPurgeBudgetChange{ObjectLayer: f.obj, change: func() { + _, err := f.obj.PutObjectTags(f.ctx, f.bucket, key, "keep=yes", ObjectOptions{VersionID: middle.VersionID}) + mustIAM(t, err) + }} + mustIAM(t, job.Expire.Start(f.ctx, hook, *job)) + ri := f.jobInfo(t, job) + if hook.calls != 2 || !ri.Failed || ri.Complete || ri.FailureReason == "" || ri.Objects != 1 || ri.KeysRefused != 0 { + t.Fatalf("continuation: calls=%d complete=%v failed=%v reason=%q objects=%d refused=%d", hook.calls, ri.Complete, ri.Failed, ri.FailureReason, ri.Objects, ri.KeysRefused) + } + if f.versionExists(t, key, oldest.VersionID) || !f.versionExists(t, key, middle.VersionID) || !f.versionExists(t, key, latest.VersionID) { + t.Fatal("only the first authorized version may be removed") + } + _, err := f.obj.DeleteObjectTags(f.ctx, f.bucket, key, ObjectOptions{VersionID: middle.VersionID}) + mustIAM(t, err) + mustIAM(t, job.Expire.Start(f.ctx, hook, *job)) + after := f.jobInfo(t, job) + if hook.calls != 2 || after.FailureReason != ri.FailureReason || after.Objects != 1 || after.Complete { + t.Fatal("persisted refusal must prevent a fresh purge after the tags change") + } +} + +// The purge's lock excludes tag writers from its authorization through its +// last delete, for a directory key in a lock domain of its own on one pool +// and for a key in a second pool. +func TestBatchExpirePurgeExcludesTagWriter(t *testing.T) { + for _, mode := range []string{ldDsync, "two-pools"} { + t.Run(mode, func(t *testing.T) { + var z *erasureServerPools + var bucket string + pool, key := 0, "purge/tags/" + if mode == "two-pools" { + z, bucket = consistencyPools(t) + pool = 1 + } else { + z, bucket = ldPool(t, mode) + for i := 0; ; i++ { + key = fmt.Sprintf("purge/tags-%d/", i) + if z.serverPools[0].getHashedSetIndex(encodeDirObject(key)) == 1 { + break + } + } + } + before := putConsistencyObject(t, z, bucket, encodeDirObject(key), pool, "", ObjectOptions{Versioned: true}) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + entered, release := make(chan struct{}), make(chan struct{}) + var released atomic.Bool + unblock := func() { + if released.CompareAndSwap(false, true) { + close(release) + } + } + defer unblock() + purged := make(chan error, 1) + go func() { + _, err := z.DeleteObject(ctx, bucket, key, ObjectOptions{ + DeleteAllVersions: true, + DeleteAllLatest: &deleteAllMatch{VersionID: before.VersionID, ModTime: before.ModTime}, + deleteAllAuthorize: func(versions []ObjectInfo) error { + close(entered) + select { + case <-release: + return nil + case <-ctx.Done(): + return ctx.Err() + } + }, + }) + purged <- err + }() + select { + case <-entered: + case <-ctx.Done(): + t.Fatal("purge never reached authorization") + } + written := make(chan error, 1) + go func() { + _, err := z.PutObjectTags(ctx, bucket, key, "keep=yes", ObjectOptions{VersionID: before.VersionID}) + written <- err + }() + select { + case err := <-written: + t.Errorf("tag writer crossed the purge's authorization lock: %v", err) + case <-time.After(ldWait): + unblock() + err := <-written + if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) { + t.Errorf("writer must observe the completed purge: %v", err) + } + } + unblock() + mustIAM(t, <-purged) + }) + } +} + +// batchTagWriteFault fails metadata updates of one key, so that a tag write +// reaches only the drives it does not wrap. +type batchTagWriteFault struct { + StorageAPI + bucket, object string +} + +func (d batchTagWriteFault) UpdateMetadata(ctx context.Context, bucket, object string, fi FileInfo, opts UpdateMetadataOpts) error { + if bucket == d.bucket && object == d.object { + return errDiskNotFound + } + return d.StorageAPI.UpdateMetadata(ctx, bucket, object, fi, opts) +} + +// A failed tag write can leave a revision that the submitter may not expire +// on a minority of drives. The purge removes it with the version, so it is +// authorized too. +func TestBatchExpirePurgeAuthorizesDriveRevisions(t *testing.T) { + resetTestGlobals() + t.Cleanup(resetTestGlobals) + z := purgeSinglePool(t, 16) + globalIAMSys.initStore(z, nil) + mustIAM(t, globalIAMSys.Load(t.Context(), true)) + f := &batchAuthzFixture{ctx: t.Context(), sys: globalIAMSys, obj: z, bucket: "review6-batch-authz"} + mustIAM(t, z.MakeBucket(f.ctx, f.bucket, MakeBucketOptions{VersioningEnabled: true})) + batchTestJobPool(f.ctx, t, z) + const key = "purge/minority-tags" + older := f.putVersions(t, key) + batchAuthzPolicy(f.ctx, t, f.sys, "review6-minority-deny", strings.ReplaceAll(batchAuthzFullAccess+`, +{"Effect":"Deny","Action":["s3:GetObjectVersion"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"StringEquals":{"s3:ExistingObjectTag/keep":"yes"}}}`, "batch-authz", f.bucket)) + cred := iamAdminUser(f.ctx, t, f.sys, "review6-minority-user", "review6-minority-deny") + sub, err := newBatchJobSubmitter(cred) + mustIAM(t, err) + if err := sub.authorize(f.ctx, batchAuthzRequest{action: policy.GetObjectVersionAction, bucket: f.bucket, object: key, versionID: older, tags: "keep=yes"}); !isBatchJobAuthzError(err) { + t.Fatal("control: this tag revision must deny the read needed for expiration") + } + set := z.serverPools[0].getHashedSet(key) + original := set.getDisks + disks := append([]StorageAPI(nil), original()...) + wrapped := append([]StorageAPI(nil), disks...) + for i := 1; i < len(wrapped); i++ { + wrapped[i] = batchTagWriteFault{StorageAPI: wrapped[i], bucket: f.bucket, object: key} + } + set.getDisks = func() []StorageAPI { return wrapped } + _, err = f.obj.PutObjectTags(f.ctx, f.bucket, key, "keep=yes", ObjectOptions{VersionID: older, UserDefined: map[string]string{ + ReservedMetadataPrefixLower + TaggingTimestamp: UTCNow().Format(time.RFC3339Nano), + }}) + set.getDisks = original + if err == nil { + t.Fatal("control: the tag write must fail its quorum") + } + fi, err := disks[0].ReadVersion(f.ctx, "", f.bucket, key, older, ReadOptions{}) + mustIAM(t, err) + if fi.Metadata[xhttp.AmzObjectTagging] != "keep=yes" { + t.Fatal("control: the successful drive must hold the denied tag revision") + } + history, err := z.deleteAllHistory(f.ctx, f.bucket, key) + mustIAM(t, err) + for _, v := range history { + for _, c := range v.copies { + if c.fi.Metadata[xhttp.AmzObjectTagging] != "" { + t.Fatal("control: the quorum history must still have the original tags") + } + } + } + ri := f.runChanged(t, cred, "purge/", func() {}) + if !ri.Failed || ri.Complete || !strings.Contains(ri.FailureReason, "is not allowed") { + t.Errorf("denied physical revision must refuse the job: complete=%v failed=%v reason=%q", ri.Complete, ri.Failed, ri.FailureReason) + } + if _, err := disks[0].ReadVersion(f.ctx, "", f.bucket, key, older, ReadOptions{}); err != nil { + t.Errorf("denied physical revision was removed: %v", err) + } + if got := batchPurgeVersionIDs(t, f, key); len(got) != 2 { + t.Errorf("denial must keep the whole key; remaining versions=%v", got) + } +} + +// Cases found in the seventh review round. + +// The time the authorization takes counts against the purge's budget: a +// late allow and a deny both delete nothing. +func TestBatchExpirePurgeBudgetIncludesAuthorization(t *testing.T) { + z := purgeSinglePool(t, 16) + p := newPurgeTest(t, z, "review7-purge-auth-time", MakeBucketOptions{VersioningEnabled: true}, "") + const key = "purge/key" + p.version(0, key, "older", 2) + p.version(0, key, "latest", 1) + exp := p.matched(key) + oldBudget := deleteAllBudget + deleteAllBudget.duration = 10 * time.Millisecond + t.Cleanup(func() { deleteAllBudget = oldBudget }) + denied := errors.New("review7: revision denied") + for _, decision := range []error{nil, denied} { + called := false + var res deleteAllResult + _, err := z.DeleteObject(t.Context(), p.bucket, key, ObjectOptions{ + DeleteAllVersions: true, + DeleteAllLatest: &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime}, + DeleteAllResult: &res, + deleteAllAuthorize: func(versions []ObjectInfo) error { + called = true + if len(versions) < 2 { + t.Error("authorization did not receive both approved versions") + } + time.Sleep(20 * time.Millisecond) + return decision + }, + }) + want := decision + if want == nil { + want = errDeleteAllBudget + } + if !called || !errors.Is(err, want) || res.Started || res.Complete || res.Removed != 0 { + t.Errorf("late authorization: called=%v err=%v result=%+v", called, err, res) + } + if got := p.held(0, key); len(got) != 2 { + t.Fatalf("purge removed versions after late authorization: %v", got) + } + } +} diff --git a/cmd/batch-expire.go b/cmd/batch-expire.go index 3f8b1e15a..3ddf86450 100644 --- a/cmd/batch-expire.go +++ b/cmd/batch-expire.go @@ -347,22 +347,10 @@ const ( batchExpireJobDefaultRetryDelay = 250 * time.Millisecond ) -// expireVersionRef is a version of a key whose versions are all purged, as -// the walk saw it. -type expireVersionRef struct { - VersionID string - UserTags string - DeleteMarker bool - Locked bool -} - type expireObjInfo struct { ObjectInfo ExpireAll bool DeleteMarkerCount int64 - // versions are the versions of an ExpireAll key the walk saw, the - // latest first. - versions []expireVersionRef } // isErrObjectLocked reports a version that Object Lock protects. @@ -485,35 +473,20 @@ func (r *BatchJobExpire) expireVersions(ctx context.Context, api ObjectLayer, vc // expireAllVersions purges every version of a key whose rule retains none, // with the purge of exactly that key (erasureServerPools.deleteAllVersions), -// while the version the walk matched is still its latest. What the purge -// removed is counted at once; a key it refused because Object Lock protects -// a version is counted in res as refused, and the versions a failed purge -// left in res as failures. A failed purge is retried up to retryAttempts -// times, each time with the version the walk matched; a purge that stopped -// at its budget after removing versions continues without spending an -// attempt. A key whose latest version changed since the walk is not -// expired. It returns an error that ends the job. +// while the version the walk matched is still its latest. Under its lock, +// before it deletes anything, the purge refuses the key when Object Lock +// protects a version and has every write it would delete, each copy and +// each revision a drive holds, authorized for the submitter, with the +// version ID and tags it has then. // -// TODO(rebase onto the delete-all purge, claude/fix-delete-all): before the -// purge removes anything, authorize every version of the history it -// approved under its lock, with version-specific conditions -// (expireAuthzRequests). Until then the checks here use the versions the -// walk saw, and do not cover a version added after the walk; the purge -// itself refuses a key that holds a locked version. +// What the purge removed is counted at once; a refused key is counted in res +// as refused, and the versions a failed purge left in res as failures. A +// failed purge is retried up to retryAttempts times, each time with the +// version the walk matched; a purge that stopped at its budget after +// removing versions continues without spending an attempt. A key whose +// latest version changed since the walk is not expired. It returns an error +// that ends the job: an authorization failure or the end of ctx. func (r *BatchJobExpire) expireAllVersions(ctx context.Context, api ObjectLayer, job BatchJobRequest, ri *batchJobInfo, exp expireObjInfo, retryAttempts int, res *batchCounts) error { - for _, v := range exp.versions { - if v.Locked { - res.keysRefused++ - return nil - } - } - var reqs []batchAuthzRequest - for _, v := range exp.versions { - reqs = append(reqs, r.expireAuthzRequests(exp.Name, v.VersionID, v.UserTags)...) - } - if err := job.Submitter.authorize(ctx, reqs...); err != nil { - return err - } var objects, markers int64 matched := deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime} for attempts := 1; attempts <= retryAttempts; attempts++ { @@ -523,11 +496,15 @@ func (r *BatchJobExpire) expireAllVersions(ctx context.Context, api ObjectLayer, stopFn := globalBatchJobsMetrics.trace(batchJobMetricExpire, ri.JobID, attempts) latest := matched var purge deleteAllResult - _, err := api.DeleteObject(ctx, exp.Bucket, encodeDirObject(exp.Name), ObjectOptions{ - DeleteAllVersions: true, - DeleteAllLatest: &latest, - DeleteAllResult: &purge, - }) + p, err := job.Submitter.principal(ctx) + if err == nil { + _, err = api.DeleteObject(ctx, exp.Bucket, encodeDirObject(exp.Name), ObjectOptions{ + DeleteAllVersions: true, + DeleteAllLatest: &latest, + DeleteAllResult: &purge, + deleteAllAuthorize: r.expireAllAuthorize(p, exp.Name), + }) + } // Removed versions are gone for any later run: count them now. if purge.Removed > 0 { ri.countRemoved(r.Bucket, int64(purge.Removed-purge.Markers), int64(purge.Markers)) @@ -538,6 +515,8 @@ func (r *BatchJobExpire) expireAllVersions(ctx context.Context, api ObjectLayer, switch { case err == nil: return nil + case isBatchJobAuthzError(err): + return err case errors.Is(err, errDeleteAllChanged): // Changed since the walk matched it: not expired. return nil @@ -562,6 +541,22 @@ func (r *BatchJobExpire) expireAllVersions(ctx context.Context, api ObjectLayer, return nil } +// expireAllAuthorize returns the authorization of a purge of all versions of +// name for p: each version the purge would delete, as it holds it, needs the +// actions of expiring it (expireAuthzRequests). +func (r *BatchJobExpire) expireAllAuthorize(p batchJobPrincipal, name string) func([]ObjectInfo) error { + return func(versions []ObjectInfo) error { + for _, v := range versions { + for _, req := range r.expireAuthzRequests(name, v.VersionID, v.UserTags) { + if err := p.authorize(req); err != nil { + return err + } + } + } + return nil + } +} + // expireAllFailed counts the versions of info, as the walk saw them, that a // failed purge of all its versions did not remove: those beyond the objects // and markers it removed, at least one. @@ -833,17 +828,12 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo toDel = append(toDel, expireObjInfo{ ObjectInfo: result.Item, ExpireAll: true, - versions: []expireVersionRef{newExpireVersionRef(ctx, result.Item)}, }) continue } } else if prevObj.Name == result.Item.Name { if matchedFilter.Purge.RetainVersions == 0 { - // The latest version in toDel purges them all. - if n := len(toDel); n > 0 && toDel[n-1].Name == result.Item.Name { - toDel[n-1].versions = append(toDel[n-1].versions, newExpireVersionRef(ctx, result.Item)) - } - continue + continue // including latest version in toDel suffices, skipping other versions } versionsCount++ } else { @@ -917,15 +907,6 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo return nil } -func newExpireVersionRef(ctx context.Context, oi ObjectInfo) expireVersionRef { - return expireVersionRef{ - VersionID: oi.VersionID, - UserTags: oi.UserTags, - DeleteMarker: oi.DeleteMarker, - Locked: enforceRetentionForDeletion(ctx, oi), - } -} - //msgp:ignore batchExpireJobError type batchExpireJobError struct { Code string diff --git a/cmd/erasure-server-pool-delete-all-options_test.go b/cmd/erasure-server-pool-delete-all-options_test.go index 4b64eeaf3..4dfac1b79 100644 --- a/cmd/erasure-server-pool-delete-all-options_test.go +++ b/cmd/erasure-server-pool-delete-all-options_test.go @@ -25,6 +25,8 @@ import ( "sync/atomic" "testing" "time" + + xhttp "github.com/minio/minio/internal/http" ) // The parts of the purge of all versions that need its own options. @@ -65,16 +67,22 @@ func TestPurgeAllVersionsOptions(t *testing.T) { p.version(0, "ea/key", "v1", 5) match := &deleteAllMatch{} expire := ExpirationOptions{Expire: true} + allow := func([]ObjectInfo) error { return nil } for name, opts := range map[string]ObjectOptions{ "no approval": {DeleteAllVersions: true}, - "two approvals": {DeleteAllVersions: true, Expiration: expire, DeleteAllLatest: match}, - "prefix": {DeleteAllVersions: true, Expiration: expire, DeletePrefix: true}, - "prefix object": {DeleteAllVersions: true, Expiration: expire, DeletePrefixObject: true}, - "version": {DeleteAllVersions: true, Expiration: expire, VersionID: mustGetUUID()}, - "movement": {DeleteAllVersions: true, Expiration: expire, DataMovement: true}, - "marker": {DeleteAllVersions: true, Expiration: expire, DeleteMarker: true}, - "versioned": {DeleteAllVersions: true, Expiration: expire, Versioned: true}, - "suspended": {DeleteAllVersions: true, Expiration: expire, VersionSuspended: true}, + "two approvals": {DeleteAllVersions: true, Expiration: expire, DeleteAllLatest: match, deleteAllAuthorize: allow}, + // A batch purge deletes as the submitter of its job, an ILM purge + // as the server, and no other delete takes an authorization. + "batch unauthorized": {DeleteAllVersions: true, DeleteAllLatest: match}, + "ilm authorized": {DeleteAllVersions: true, Expiration: expire, deleteAllAuthorize: allow}, + "authorized delete": {deleteAllAuthorize: allow}, + "prefix": {DeleteAllVersions: true, Expiration: expire, DeletePrefix: true}, + "prefix object": {DeleteAllVersions: true, Expiration: expire, DeletePrefixObject: true}, + "version": {DeleteAllVersions: true, Expiration: expire, VersionID: mustGetUUID()}, + "movement": {DeleteAllVersions: true, Expiration: expire, DataMovement: true}, + "marker": {DeleteAllVersions: true, Expiration: expire, DeleteMarker: true}, + "versioned": {DeleteAllVersions: true, Expiration: expire, Versioned: true}, + "suspended": {DeleteAllVersions: true, Expiration: expire, VersionSuspended: true}, } { if _, err := z.DeleteObject(t.Context(), p.bucket, "ea/key", opts); !errors.As(err, new(InvalidArgument)) { t.Errorf("%s: %v", name, err) @@ -134,9 +142,10 @@ func TestPurgeAllVersionsBudgetCoversReads(t *testing.T) { t.Cleanup(func() { pool.erasureDisksMu.Lock(); pool.erasureDisks = prior; pool.erasureDisksMu.Unlock() }) var res deleteAllResult _, err := z.DeleteObject(t.Context(), p.bucket, key, ObjectOptions{ - DeleteAllVersions: true, - DeleteAllLatest: &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime}, - DeleteAllResult: &res, + DeleteAllVersions: true, + DeleteAllLatest: &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime}, + DeleteAllResult: &res, + deleteAllAuthorize: func([]ObjectInfo) error { return nil }, }) if !errors.Is(err, errDeleteAllBudget) || res.Started || res.Removed != 0 { t.Errorf("purge after its budget ran out on reads: err=%v %+v", err, res) @@ -213,9 +222,10 @@ func TestPurgeAllVersionsDeletesOnCheckedDrives(t *testing.T) { }) var res deleteAllResult if _, err := z.DeleteObject(t.Context(), p.bucket, key, ObjectOptions{ - DeleteAllVersions: true, - DeleteAllLatest: &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime}, - DeleteAllResult: &res, + DeleteAllVersions: true, + DeleteAllLatest: &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime}, + DeleteAllResult: &res, + deleteAllAuthorize: func([]ObjectInfo) error { return nil }, }); err != nil || !res.Complete { t.Fatalf("purge: %v %+v", err, res) } @@ -223,3 +233,62 @@ func TestPurgeAllVersionsDeletesOnCheckedDrives(t *testing.T) { t.Errorf("reads=%d deletes on the replacement handle=%d", reads.Load(), deletes.Load()) } } + +// A batch purge has every copy of every version it approved authorized, as +// the copy is now, before it deletes any; an authorization error deletes +// nothing and is returned as it is. +func TestPurgeAllVersionsAuthorizesEveryCopy(t *testing.T) { + z, _ := consistencyPools(t) + p := newPurgeTest(t, z, "purge-authorize", MakeBucketOptions{VersioningEnabled: true}, "") + const key = "bx/authorize" + p.put(0, key, "old", daysAgo(6), ObjectOptions{}) + p.put(1, key, "new", daysAgo(5), ObjectOptions{UserDefined: map[string]string{xhttp.AmzObjectTagging: "copy=new"}}) + tagged := p.put(0, key, "v1", daysAgo(4), ObjectOptions{Versioned: true, UserDefined: map[string]string{xhttp.AmzObjectTagging: "keep=yes"}}) + exp := p.matched(key) + denied := errors.New("denied") + var seen []string // the distinct versions and tags authorized + purge := func(authorize func([]ObjectInfo) error) (deleteAllResult, error) { + var res deleteAllResult + _, err := z.DeleteObject(t.Context(), p.bucket, key, ObjectOptions{ + DeleteAllVersions: true, + DeleteAllLatest: &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime}, + DeleteAllResult: &res, + deleteAllAuthorize: authorize, + }) + return res, err + } + res, err := purge(func(versions []ObjectInfo) error { + for _, v := range versions { + if got := v.VersionID + " " + v.UserTags; !slices.Contains(seen, got) { + seen = append(seen, got) + } + } + return denied + }) + if !errors.Is(err, denied) || res.Started || res.Removed != 0 { + t.Errorf("denied purge: err=%v %+v", err, res) + } + want := []string{"null ", "null copy=new", tagged.VersionID + " keep=yes"} + slices.Sort(seen) + slices.Sort(want) + if !slices.Equal(seen, want) { + t.Errorf("authorized %q, want every copy %q", seen, want) + } + for pool := range z.serverPools { + if got := p.held(pool, key); len(got) == 0 { + t.Errorf("pool %d lost its copies to a denied purge", pool) + } + } + if got := p.get(key); got != "v1" { + t.Errorf("reads %s after a denied purge, want v1", got) + } + // Control: an allowed purge deletes them all. + if res, err = purge(func([]ObjectInfo) error { return nil }); err != nil || !res.Complete { + t.Fatalf("allowed purge: %v %+v", err, res) + } + for pool := range z.serverPools { + if got := p.held(pool, key); got != nil { + t.Errorf("pool %d keeps %v", pool, got) + } + } +} diff --git a/cmd/erasure-server-pool-delete-all.go b/cmd/erasure-server-pool-delete-all.go index 50c594270..42fc17101 100644 --- a/cmd/erasure-server-pool-delete-all.go +++ b/cmd/erasure-server-pool-delete-all.go @@ -26,6 +26,7 @@ import ( "sync" "time" + xhttp "github.com/minio/minio/internal/http" "github.com/pgsty/silo-pkg/v3/sync/errgroup" ) @@ -49,6 +50,9 @@ import ( // when any revision of an approved version there is transitioned or, // when the bucket has Object Lock, protected by retention or a legal // hold. It deletes on exactly the drives it read; +// - a batch expiry then authorizes every copy of every approved version, +// and every revision of one that a drive holds, for the submitter of the +// job, and the purge deletes nothing unless all of them are allowed; // - it deletes the oldest version first and the latest version last and // alone, once every older version is gone from every pool. A purge that // stops early keeps the latest version, so it never exposes an older @@ -118,11 +122,14 @@ func deleteAllVersionID(versionID string) string { } // deleteAllInvalid returns why opts cannot ask for a purge of all versions. -// The purge is approved either as an ILM expiry or by a batch match, and it -// only deletes: it never creates a marker, addresses a single version, -// moves data or deletes a prefix. +// The purge is approved either as an ILM expiry or by a batch match, which +// also authorizes the versions it deletes, and it only deletes: it never +// creates a marker, addresses a single version, moves data or deletes a +// prefix. func (opts ObjectOptions) deleteAllInvalid() error { switch { + case !opts.DeleteAllVersions: + return errors.New("only a purge of all versions authorizes the versions it deletes") case opts.DeletePrefix || opts.DeletePrefixObject: return errors.New("a purge of all versions is not a prefix delete") case opts.VersionID != "": @@ -137,6 +144,8 @@ func (opts ObjectOptions) deleteAllInvalid() error { return errors.New("a purge of all versions takes no options of other deletes") case opts.Expiration.Expire == (opts.DeleteAllLatest != nil): return errors.New("a purge of all versions needs exactly one approval") + case (opts.DeleteAllLatest != nil) != (opts.deleteAllAuthorize != nil): + return errors.New("a batch purge of all versions, and only it, authorizes the versions it deletes") } return nil } @@ -297,12 +306,29 @@ func (z *erasureServerPools) deleteAllVersions(ctx context.Context, bucket, obje } } } + // A batch expiry deletes as its submitter: every write it is about to + // delete, the accepted copies and every revision a drive holds, must be + // allowed with the version ID and tags it has now. + var writes *deleteAllWrites + if opts.deleteAllAuthorize != nil { + writes = &deleteAllWrites{bucket: bucket, name: name, seen: make(map[[2]string]bool, len(history))} + for _, v := range history { + for _, c := range v.copies { + writes.add(c.fi) + } + } + } drives := make([][]StorageAPI, len(z.serverPools)) for pool := range z.serverPools { if !held[pool] { continue } - if drives[pool], err = z.serverPools[pool].getHashedSet(object).deleteAllDrives(ctx, bucket, object, ids, lr.LockEnabled); err != nil { + if drives[pool], err = z.serverPools[pool].getHashedSet(object).deleteAllDrives(ctx, bucket, object, ids, lr.LockEnabled, writes); err != nil { + return fail(err) + } + } + if opts.deleteAllAuthorize != nil { + if err := opts.deleteAllAuthorize(writes.versions); err != nil { return fail(err) } } @@ -329,6 +355,28 @@ func (z *erasureServerPools) deleteAllVersions(ctx context.Context, bucket, obje return res.Latest, nil } +// deleteAllWrites collects, for the authorization of a batch purge, the +// writes it deletes: one of each distinct version ID and tags. +type deleteAllWrites struct { + bucket, name string + seen map[[2]string]bool + versions []ObjectInfo +} + +// add adds the write fi unless w is nil or holds one with its version ID and +// tags. +func (w *deleteAllWrites) add(fi FileInfo) { + if w == nil { + return + } + key := [2]string{deleteAllVersionID(fi.VersionID), fi.Metadata[xhttp.AmzObjectTagging]} + if w.seen[key] { + return + } + w.seen[key] = true + w.versions = append(w.versions, fi.ToObjectInfo(w.bucket, w.name, true)) +} + // deleteAllRefuses returns why the purge must keep the key that holds the // revision fi of an approved version: it is transitioned, whose remote // object a later release hands to the tier cleanup, or, with Object Lock, @@ -344,7 +392,8 @@ func deleteAllRefuses(ctx context.Context, bucket, object string, fi FileInfo, l } // deleteAllDrives reads object on every drive of the set and checks each -// revision of the versions ids that a drive holds with deleteAllRefuses. +// revision of the versions ids that a drive holds with deleteAllRefuses, and +// adds it to writes. // A minority of drives can hold a revision other than the accepted one, and // heal would later remove what the purge leaves on a drive, so a drive that // cannot be read or decoded refuses the key. A drive without the object @@ -353,7 +402,7 @@ func deleteAllRefuses(ctx context.Context, bucket, object string, fi FileInfo, l // handle names a drive, not a backend: a backend that replaces the drive // behind it starts empty and receives this object only from a write or a // heal of it, which take the object lock the purge holds. -func (er erasureObjects) deleteAllDrives(ctx context.Context, bucket, object string, ids map[string]bool, lockEnabled bool) ([]StorageAPI, error) { +func (er erasureObjects) deleteAllDrives(ctx context.Context, bucket, object string, ids map[string]bool, lockEnabled bool, writes *deleteAllWrites) ([]StorageAPI, error) { disks := slices.Clone(er.getDisks()) raws, errs := readAllRawFileInfo(ctx, disks, bucket, object, false) if err := ctx.Err(); err != nil { @@ -381,6 +430,7 @@ func (er erasureObjects) deleteAllDrives(ctx context.Context, bucket, object str if err := deleteAllRefuses(ctx, bucket, object, fi, lockEnabled); err != nil { return nil, err } + writes.add(fi) } } return disks, nil diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index efaf3654b..1df968d81 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1234,7 +1234,7 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob if err = checkDelObjArgs(ctx, bucket, object); err != nil { return objInfo, err } - if opts.DeleteAllVersions { + if opts.DeleteAllVersions || opts.deleteAllAuthorize != nil { if err := opts.deleteAllInvalid(); err != nil { return objInfo, InvalidArgument{Bucket: bucket, Object: object, Err: err} } diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go index fd52d307c..2ad350169 100644 --- a/cmd/object-api-interface.go +++ b/cmd/object-api-interface.go @@ -113,6 +113,12 @@ type ObjectOptions struct { // DeleteAllResult, when set, receives what a purge of all versions // confirmed, also when it fails part way. DeleteAllResult *deleteAllResult + // deleteAllAuthorize authorizes a batch purge of all versions, which + // needs it, on every copy of every version the purge approved and every + // revision of one that a drive holds, one of each distinct version ID + // and tags, before it deletes any; its error refuses the key and is + // returned. Set by batch expiration only, never from a request. + deleteAllAuthorize func(versions []ObjectInfo) error // Pools-layer version resolver; the caller holds the shared object lock. replicaObjectInfo GetObjectInfoFn