diff --git a/cmd/batch-authz-review2_test.go b/cmd/batch-authz-review2_test.go
new file mode 100644
index 000000000..5692d1d52
--- /dev/null
+++ b/cmd/batch-authz-review2_test.go
@@ -0,0 +1,356 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "maps"
+ "net/http"
+ "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"
+ "github.com/pgsty/silo-pkg/v3/policy"
+)
+
+// A pull into a suspended bucket replaces the null version. Its policy
+// decision must name that version even when the source is unversioned S3.
+func TestBatchReviewPullImplicitNullVersionAuthorization(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ _, err := globalBucketMetadataSys.Update(f.ctx, f.bucket, bucketVersioningConfig, []byte(`Suspended`))
+ mustIAM(t, err)
+ data := []byte("original null version")
+ before, err := f.obj.PutObject(f.ctx, f.bucket, "pull/null", mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), ObjectOptions{VersionSuspended: true})
+ mustIAM(t, err)
+ batchAuthzPolicy(f.ctx, t, f.sys, "null-deny", batchAuthzFullAccess+`,
+{"Effect":"Deny","Action":["s3:PutObject"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"StringEquals":{"s3:versionid":"null"}}}`)
+ cred := iamAdminUser(f.ctx, t, f.sys, "null-deny-user", "null-deny")
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ job := batchPullJob(f.bucket)
+ job.Source.Type = BatchJobReplicateResourceS3
+ err = job.ReplicateFromSource(f.ctx, f.obj, batchLockRemote(t, "replacement"), ObjectInfo{
+ Name: before.Name, ModTime: UTCNow(), ETag: "batchetag", UserDefined: map[string]string{},
+ }, false, sub)
+ if !isBatchJobAuthzError(err) {
+ t.Errorf("null version policy must refuse the write, got %v", err)
+ }
+ after, gerr := f.obj.GetObjectInfo(f.ctx, f.bucket, before.Name, ObjectOptions{VersionID: nullVersionID})
+ if gerr != nil || before.ETag != after.ETag {
+ t.Errorf("denied null version must remain unchanged: read=%v unchanged=%v", gerr, before.ETag == after.ETag)
+ }
+}
+
+type batchReviewPrefixWalkFailure struct{ ObjectLayer }
+
+func (a batchReviewPrefixWalkFailure) Walk(ctx context.Context, bucket, prefix string, results chan<- itemOrErr[ObjectInfo], opts WalkOptions) error {
+ if prefix != "a/" {
+ return a.ObjectLayer.Walk(ctx, bucket, prefix, results, opts)
+ }
+ go func() {
+ defer close(results)
+ results <- itemOrErr[ObjectInfo]{Err: errors.New("test: incomplete first prefix")}
+ }()
+ return nil
+}
+
+type batchReviewWalkCutoff struct {
+ ObjectLayer
+ cutoff string
+}
+
+func (a batchReviewWalkCutoff) Walk(ctx context.Context, bucket, prefix string, results chan<- itemOrErr[ObjectInfo], opts WalkOptions) error {
+ inner := make(chan itemOrErr[ObjectInfo], 1)
+ if err := a.ObjectLayer.Walk(ctx, bucket, prefix, inner, opts); err != nil {
+ close(results)
+ return err
+ }
+ go func() {
+ defer close(results)
+ stopped := false
+ for result := range inner {
+ if stopped {
+ continue
+ }
+ if result.Err == nil && result.Item.Name >= a.cutoff {
+ stopped = true
+ result = itemOrErr[ObjectInfo]{Err: errors.New("test: listing stopped before remaining child keys")}
+ }
+ select {
+ case results <- result:
+ case <-ctx.Done():
+ stopped = true
+ }
+ }
+ }()
+ return nil
+}
+
+// Listing a directory prefix returns the directory object before its child
+// keys. A durable cursor for that directory must preserve unfinished children.
+func TestBatchReviewDirectoryCursorKeepsChildrenPending(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ for range 12 {
+ batchPutVersion(t, f.obj, f.bucket, "a/", nil)
+ }
+ var pending string
+ for i := 0; i < 13; i++ {
+ older := f.putVersions(t, fmt.Sprintf("a/key-%02d", i))
+ if i == 11 {
+ pending = older
+ }
+ }
+ rec, job := f.submit(t, globalActiveCred, fmt.Sprintf(`expire:
+ apiVersion: v1
+ bucket: %s
+ prefix: ["a/"]
+ rules:
+ - type: object
+ purge:
+ retainVersions: 1
+`, f.bucket))
+ if job == nil {
+ t.Fatalf("submit: %d %s", rec.Code, rec.Body.String())
+ }
+ _ = job.Expire.Start(f.ctx, batchReviewWalkCutoff{ObjectLayer: f.obj, cutoff: "a/key-11"}, *job)
+ ri := f.jobInfo(t, job)
+ if ri.Object != "a/" {
+ t.Fatalf("control: the completed directory batch must be the cursor, got %q", ri.Object)
+ }
+ mustIAM(t, job.Expire.Start(f.ctx, f.obj, *job))
+ if f.versionExists(t, "a/key-11", pending) {
+ t.Error("a directory cursor must not suppress the unvisited child key on resume")
+ }
+}
+
+// Later prefixes must not advance the completion cursor past an earlier
+// prefix whose listing failed. This state is produced by Start itself.
+func TestBatchReviewExpireFailedPrefixKeepsWorkPending(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ older := f.putVersions(t, "a/todo")
+ for i := 0; i < 13; i++ {
+ f.putVersions(t, fmt.Sprintf("z/key-%02d", i))
+ }
+ rec, job := f.submit(t, globalActiveCred, fmt.Sprintf(`expire:
+ apiVersion: v1
+ bucket: %s
+ prefix: ["z/", "a/"]
+ rules:
+ - type: object
+ purge:
+ retainVersions: 1
+`, f.bucket))
+ if job == nil {
+ t.Fatalf("submit: %d %s", rec.Code, rec.Body.String())
+ }
+ _ = job.Expire.Start(f.ctx, batchReviewPrefixWalkFailure{f.obj}, *job)
+ ri := f.jobInfo(t, job)
+ if ri.Object >= "a/todo" {
+ t.Errorf("cursor must not cross an incomplete prefix, got %q", ri.Object)
+ }
+ mustIAM(t, job.Expire.Start(f.ctx, f.obj, *job))
+ if f.versionExists(t, "a/todo", older) {
+ t.Error("a fresh Start must revisit the key omitted by the failed listing")
+ }
+}
+
+// A repeated marker replication error must remain a failure on the second
+// attempt, including after the retry loop resets its counters.
+func TestBatchReviewMarkerRetryRemainsFailed(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchPutVersion(t, f.obj, f.bucket, "marker", nil)
+ _, err := f.obj.DeleteObject(f.ctx, f.bucket, "marker", ObjectOptions{Versioned: true})
+ mustIAM(t, err)
+ oldTransport := globalRemoteTargetTransport
+ globalRemoteTargetTransport = batchLockTransport(func(r *http.Request) (*http.Response, error) {
+ code, body := http.StatusForbidden, `AccessDeniedtest refusal`
+ if r.URL.Query().Has("location") {
+ code, body = http.StatusOK, `us-east-1`
+ }
+ return &http.Response{StatusCode: code, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(body)), Request: r}, nil
+ })
+ t.Cleanup(func() { globalRemoteTargetTransport = oldTransport })
+ job := BatchJobRequest{
+ ID: "replicate-marker-retry" + getKeySeparator() + "0", Started: UTCNow(), Submitter: batchRootSubmitter(t),
+ Replicate: &BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: f.bucket},
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: "remote", Endpoint: "https://target.invalid", Creds: BatchJobReplicateCredentials{AccessKey: "test-access", SecretKey: "test-secret"}},
+ },
+ }
+ job.Replicate.Flags.Retry.Attempts = 2
+ job.Replicate.Flags.Retry.Delay = time.Millisecond
+ mustIAM(t, job.Replicate.Start(f.ctx, f.obj, job))
+ ri := f.jobInfo(t, &job)
+ if ri.Complete || !ri.Failed || ri.DeleteMarkersFailed != 1 {
+ t.Errorf("both marker attempts failed: complete=%v failed=%v markerFailures=%d attempts=%d", ri.Complete, ri.Failed, ri.DeleteMarkersFailed, ri.RetryAttempts)
+ }
+}
+
+// A current parent mapping suppresses the OpenID policy-name fallback.
+// Background-condition validation must inspect the same effective policies.
+func TestBatchReviewMappedPolicySuppressesClaimFallback(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchAuthzPolicy(f.ctx, t, f.sys, "mapped-full", batchAuthzFullAccess)
+ batchAuthzPolicy(f.ctx, t, f.sys, "unused-claim-policy", `{"Effect":"Deny","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"Bool":{"aws:SecureTransport":"false"}}}`)
+ parent := iamAdminUser(f.ctx, t, f.sys, "mapped-parent", "mapped-full")
+ cred, err := auth.GetNewCredentialsWithMetadata(map[string]any{
+ parentClaim: parent.AccessKey, "exp": UTCNow().Add(time.Hour).Unix(), iamPolicyClaimNameOpenID(): "unused-claim-policy",
+ }, globalActiveCred.SecretKey)
+ mustIAM(t, err)
+ cred.ParentUser = parent.AccessKey
+ _, err = f.sys.SetTempUser(f.ctx, cred.AccessKey, cred, "")
+ mustIAM(t, err)
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ p, err := sub.principal(f.ctx)
+ mustIAM(t, err)
+ req := batchAuthzRequest{action: policy.GetObjectAction, bucket: f.bucket, object: "mapped/key", versionID: mustGetUUID()}
+ if !f.sys.isAllowedLocally(policy.Args{AccountName: p.accessKey, Claims: p.claims, Action: req.action, BucketName: req.bucket, ObjectName: req.object, ConditionValues: p.conditionValues(req)}) {
+ t.Fatal("control: the current mapped policy must allow this read")
+ }
+ if err := sub.authorize(f.ctx, req); err != nil {
+ t.Errorf("an unused fallback policy must not refuse a mapped-policy grant: %v", err)
+ }
+}
+
+func TestBatchReviewBindingAndExpiringServiceControls(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ root := batchRootSubmitter(t)
+ priorRoot := globalActiveCred
+ globalActiveCred.SecretKey = "different-test-root-secret"
+ _, err := root.principal(f.ctx)
+ globalActiveCred = priorRoot
+ if !isBatchJobAuthzError(err) {
+ t.Errorf("root-secret change must invalidate the binding: %v", err)
+ }
+ missing := *root
+ missing.Binding = nil
+ if _, err := missing.principal(f.ctx); !isBatchJobAuthzError(err) {
+ t.Errorf("missing fingerprint must refuse the principal: %v", err)
+ }
+ batchAuthzPolicy(f.ctx, t, f.sys, "service-parent-full", batchAuthzFullAccess)
+ parent := iamAdminUser(f.ctx, t, f.sys, "service-parent", "service-parent-full")
+ sp, err := policy.ParseConfig(strings.NewReader(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::batch-authz/*"]}]}`))
+ mustIAM(t, err)
+ expires := UTCNow().Add(time.Hour)
+ cred, _, err := f.sys.NewServiceAccount(f.ctx, parent.AccessKey, nil, newServiceAccountOpts{sessionPolicy: sp, expiration: &expires})
+ mustIAM(t, err)
+ stored, ok := f.sys.GetUser(f.ctx, cred.AccessKey)
+ if !ok {
+ t.Fatal("created service account must resolve")
+ }
+ cred = stored.Credentials
+ if !cred.IsTemp() || !cred.IsServiceAccount() {
+ t.Fatal("fixture must be an expiring service account")
+ }
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ if err := sub.authorize(f.ctx, batchAuthzRequest{action: policy.GetObjectAction, bucket: f.bucket, object: "allowed"}); err != nil {
+ t.Errorf("expiring service account must retain its allowed read: %v", err)
+ }
+ if err := sub.authorize(f.ctx, batchAuthzRequest{action: policy.DeleteObjectAction, bucket: f.bucket, object: "allowed"}); !isBatchJobAuthzError(err) {
+ t.Errorf("expiring service account must remain session-policy restricted: %v", err)
+ }
+}
+
+// An interrupted cross-pool copy can leave the same version in both pools.
+// A newer legal-hold field in either copy must prevent key rotation.
+func TestBatchReviewRotationUsesAuthoritativePoolHold(t *testing.T) {
+ z, bucket := consistencyPools(t)
+ const name = "rotation/split-hold"
+ previousKMS := GlobalKMS
+ GlobalKMS = kms.NewStub("batch-rotation-key")
+ t.Cleanup(func() { GlobalKMS = previousKMS })
+ dek, err := GlobalKMS.GenerateKey(t.Context(), &kms.GenerateKeyRequest{AssociatedData: kms.Context{bucket: bucket + "/" + name}})
+ mustIAM(t, err)
+ objectKey := crypto.GenerateKey(dek.Plaintext, nil)
+ sealed := objectKey.Seal(dek.Plaintext, crypto.GenerateIV(nil), crypto.S3.String(), bucket, name)
+ first := crypto.S3.CreateMetadata(nil, dek.KeyID, dek.Ciphertext, sealed)
+ first[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = "OFF"
+ first[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = UTCNow().Add(-time.Hour).Format(time.RFC3339Nano)
+ second := maps.Clone(first)
+ second[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = "ON"
+ second[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = UTCNow().Format(time.RFC3339Nano)
+ ciphertext, err := io.ReadAll(crypto.EncryptSinglePart(strings.NewReader("protected payload"), objectKey))
+ mustIAM(t, err)
+ before := putConsistencyObject(t, z, bucket, name, 0, string(ciphertext), ObjectOptions{Versioned: true, UserDefined: first})
+ putConsistencyObject(t, z, bucket, name, 1, string(ciphertext), ObjectOptions{Versioned: true, VersionID: before.VersionID, MTime: before.ModTime, UserDefined: second})
+ logical, err := z.replicaObjectInfo(t.Context(), bucket, name, ObjectOptions{VersionID: before.VersionID})
+ mustIAM(t, err)
+ if !enforceRetentionForDeletion(t.Context(), logical) {
+ t.Fatal("control: the authoritative logical version must be held")
+ }
+ r := BatchJobKeyRotateV1{Bucket: bucket, Encryption: BatchJobKeyRotateEncryption{Type: sses3}}
+ err = r.KeyRotate(t.Context(), z, before, batchRootSubmitter(t))
+ if !isErrObjectLocked(err) {
+ t.Errorf("rotation must refuse the authoritative legal hold, got %v", err)
+ }
+ after, gerr := z.serverPools[0].GetObjectInfo(t.Context(), bucket, name, ObjectOptions{VersionID: before.VersionID})
+ mustIAM(t, gerr)
+ if after.UserDefined[crypto.MetaSealedKeyS3] != before.UserDefined[crypto.MetaSealedKeyS3] {
+ t.Error("held logical version must not have its key rewritten")
+ }
+}
+
+// An encryption condition must describe what rotation actually commits,
+// even when the requested encryption type differs from the existing type.
+func TestBatchReviewRotationEncryptionConditionMatchesResult(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ previousKMS := GlobalKMS
+ GlobalKMS = kms.NewStub("required-key")
+ t.Cleanup(func() { GlobalKMS = previousKMS })
+ const name = "rotation/encryption-condition"
+ meta := make(map[string]string)
+ key, err := newEncryptMetadata(f.ctx, crypto.S3, "", nil, f.bucket, name, meta, nil)
+ mustIAM(t, err)
+ body, err := io.ReadAll(crypto.EncryptSinglePart(strings.NewReader("encrypted payload"), key))
+ mustIAM(t, err)
+ before, err := f.obj.PutObject(f.ctx, f.bucket, name, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{Versioned: true, UserDefined: meta})
+ mustIAM(t, err)
+ batchAuthzPolicy(f.ctx, t, f.sys, "kms-write-only", `
+{"Effect":"Allow","Action":["s3:GetObject","s3:GetObjectVersion"],"Resource":["arn:aws:s3:::batch-authz/*"]},
+{"Effect":"Allow","Action":["s3:PutObject"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"StringEquals":{"s3:x-amz-server-side-encryption":"aws:kms","s3:x-amz-server-side-encryption-aws-kms-key-id":"required-key"}}}`)
+ cred := iamAdminUser(f.ctx, t, f.sys, "kms-write-user", "kms-write-only")
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ r := BatchJobKeyRotateV1{Bucket: f.bucket, Encryption: BatchJobKeyRotateEncryption{Type: ssekms, Key: "required-key"}}
+ err = r.KeyRotate(f.ctx, f.obj, before, sub)
+ after, gerr := f.obj.GetObjectInfo(f.ctx, f.bucket, name, ObjectOptions{VersionID: before.VersionID})
+ mustIAM(t, gerr)
+ if err == nil && !crypto.S3KMS.IsEncrypted(after.UserDefined) {
+ t.Error("rotation allowed by an aws:kms-only write policy must produce SSE-KMS or refuse the operation")
+ }
+ if err != nil && before.UserDefined[crypto.MetaSealedKeyS3] != after.UserDefined[crypto.MetaSealedKeyS3] {
+ t.Error("a refused rotation must leave the version unchanged")
+ }
+}
+
+// An explicit null marker collision must retain a held null data version.
+func TestBatchReviewPullMarkerNullLock(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ meta := batchLockMeta("LEGALHOLD")
+ before, err := f.obj.PutObject(f.ctx, f.bucket, "pull/null", mustGetPutObjReader(t, strings.NewReader("held null"), 9, "", ""), ObjectOptions{VersionID: nullVersionID, Versioned: true, UserDefined: meta})
+ mustIAM(t, err)
+ job := BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: "remote"},
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: f.bucket},
+ }
+ err = job.ReplicateFromSource(f.ctx, f.obj, nil, ObjectInfo{
+ Name: before.Name, VersionID: nullVersionID, ModTime: UTCNow(), DeleteMarker: true,
+ }, false, batchRootSubmitter(t))
+ if !isErrObjectLocked(err) {
+ t.Errorf("marker collision with protected null data must return ObjectLocked, got %v", err)
+ }
+ if !batchVersionExists(t, f.obj, f.bucket, before.Name, nullVersionID) {
+ t.Fatal("protected null data must remain present")
+ }
+}
diff --git a/cmd/batch-authz-review3_test.go b/cmd/batch-authz-review3_test.go
new file mode 100644
index 000000000..d6d9d9313
--- /dev/null
+++ b/cmd/batch-authz-review3_test.go
@@ -0,0 +1,140 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+ "fmt"
+ "io"
+ "net/http"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/minio/minio-go/v7"
+ "github.com/minio/minio-go/v7/pkg/credentials"
+ "github.com/minio/minio/internal/auth"
+ xhttp "github.com/minio/minio/internal/http"
+ "github.com/pgsty/silo-pkg/v3/policy"
+)
+
+// A read policy must use the newest tag state of the logical version even
+// when an interrupted movement left an older copy in the preferred pool.
+func TestBatchReviewPushUsesAuthoritativePoolTags(t *testing.T) {
+ z, bucket := consistencyPools(t)
+ f := newBatchAuthzFixture(t)
+ const name = "push/tagged"
+ first := map[string]string{
+ xhttp.AmzObjectTagging: "classification=public",
+ ReservedMetadataPrefixLower + TaggingTimestamp: UTCNow().Add(-time.Hour).Format(time.RFC3339Nano),
+ }
+ second := map[string]string{
+ xhttp.AmzObjectTagging: "classification=private",
+ ReservedMetadataPrefixLower + TaggingTimestamp: UTCNow().Format(time.RFC3339Nano),
+ }
+ before := putConsistencyObject(t, z, bucket, name, 0, "test payload", ObjectOptions{Versioned: true, UserDefined: first})
+ putConsistencyObject(t, z, bucket, name, 1, "test payload", ObjectOptions{Versioned: true, VersionID: before.VersionID, MTime: before.ModTime, UserDefined: second})
+ logical, err := z.replicaObjectInfo(f.ctx, bucket, name, ObjectOptions{VersionID: before.VersionID})
+ mustIAM(t, err)
+ if logical.UserTags != "classification=private" {
+ t.Fatal("control: logical version must have the newer private tag")
+ }
+ batchAuthzPolicy(f.ctx, t, f.sys, "pooled-push", fmt.Sprintf(`
+{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::%s/*"]},
+{"Effect":"Deny","Action":["s3:GetObject","s3:GetObjectVersion"],"Resource":["arn:aws:s3:::%s/*"],"Condition":{"StringEquals":{"s3:ExistingObjectTag/classification":"private"}}}`, bucket, bucket))
+ cred := iamAdminUser(f.ctx, t, f.sys, "pooled-push-user", "pooled-push")
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ if err := sub.authorize(f.ctx, batchAuthzRequest{action: policy.GetObjectAction, bucket: bucket, object: name, versionID: before.VersionID, tags: logical.UserTags}); !isBatchJobAuthzError(err) {
+ t.Fatal("control: authoritative tags must deny the read")
+ }
+ writes := 0
+ target, err := minio.NewCore("target.invalid", &minio.Options{
+ Region: "us-east-1", Creds: credentials.NewStaticV4("test-access", "test-secret", ""),
+ Transport: batchLockTransport(func(r *http.Request) (*http.Response, error) {
+ if r.Body != nil {
+ _, _ = io.Copy(io.Discard, r.Body)
+ _ = r.Body.Close()
+ }
+ if r.Method == http.MethodPut {
+ writes++
+ }
+ return &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Etag": {"\"test-etag\""}}, Body: io.NopCloser(strings.NewReader("")), Request: r}, nil
+ }),
+ })
+ mustIAM(t, err)
+ r := BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: bucket},
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: "remote"},
+ }
+ err = r.ReplicateToTarget(f.ctx, z, target, before, false, sub)
+ if !isBatchJobAuthzError(err) || writes != 0 {
+ t.Errorf("newer authoritative tags must deny push: err=%v writes=%d", err, writes)
+ }
+}
+
+// Verified token claims must not replace the actual version being authorized.
+func TestBatchReviewVersionConditionResistsClaimCollision(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ deniedVersion := mustGetUUID()
+ batchAuthzPolicy(f.ctx, t, f.sys, "version-claims", batchAuthzFullAccess+fmt.Sprintf(`,
+{"Effect":"Deny","Action":["s3:DeleteObjectVersion"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"StringEquals":{"s3:versionid":"%s"}}}`, deniedVersion))
+ parent := iamAdminUser(f.ctx, t, f.sys, "version-claims-parent", "version-claims")
+ cred, err := auth.GetNewCredentialsWithMetadata(map[string]any{
+ parentClaim: parent.AccessKey, "exp": UTCNow().Add(time.Hour).Unix(), "versionid": "an-unrelated-claim-value",
+ }, globalActiveCred.SecretKey)
+ mustIAM(t, err)
+ cred.ParentUser = parent.AccessKey
+ _, err = f.sys.SetTempUser(f.ctx, cred.AccessKey, cred, "")
+ mustIAM(t, err)
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ req := batchAuthzRequest{action: policy.DeleteObjectVersionAction, bucket: f.bucket, object: "data", versionID: deniedVersion}
+ if err := sub.authorize(f.ctx, req); !isBatchJobAuthzError(err) {
+ t.Errorf("a token claim must not replace the actual denied version: %v", err)
+ }
+}
+
+// Retry totals describe a pass, including markers already successful before
+// another item caused that pass to retry.
+func TestBatchReviewMarkerSuccessCountResetsOnRetry(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchPutVersion(t, f.obj, f.bucket, "marker", nil)
+ _, err := f.obj.DeleteObject(f.ctx, f.bucket, "marker", ObjectOptions{Versioned: true})
+ mustIAM(t, err)
+ oldTransport := remoteInstanceTransport
+ remoteInstanceTransport = atomic.Value{}
+ setRemoteInstanceTransport(batchLockTransport(func(r *http.Request) (*http.Response, error) {
+ code, body := http.StatusForbidden, `AccessDeniedtest refusal`
+ h := make(http.Header)
+ switch {
+ case r.URL.Query().Has("location"):
+ code, body = http.StatusOK, `us-east-1`
+ case r.Method == http.MethodDelete:
+ code, body = http.StatusNoContent, ""
+ case r.Method == http.MethodHead && r.URL.Query().Get("versionId") != "":
+ code, body = http.StatusMethodNotAllowed, ""
+ h.Set("X-Amz-Delete-Marker", "true")
+ }
+ return &http.Response{StatusCode: code, Header: h, Body: io.NopCloser(strings.NewReader(body)), Request: r}, nil
+ }))
+ t.Cleanup(func() { remoteInstanceTransport = oldTransport })
+ job := BatchJobRequest{
+ ID: "marker-success-retry" + getKeySeparator() + "0", Started: UTCNow(), Submitter: batchRootSubmitter(t),
+ Replicate: &BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: f.bucket},
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: "remote", Endpoint: "https://target.invalid", Creds: BatchJobReplicateCredentials{AccessKey: "test-access", SecretKey: "test-secret"}},
+ },
+ }
+ job.Replicate.Flags.Retry.Attempts = 2
+ job.Replicate.Flags.Retry.Delay = time.Millisecond
+ mustIAM(t, job.Replicate.Start(f.ctx, f.obj, job))
+ ri := f.jobInfo(t, &job)
+ if !ri.Failed || ri.RetryAttempts != 2 {
+ t.Fatalf("control: data failure must cause two attempts: failed=%v attempts=%d", ri.Failed, ri.RetryAttempts)
+ }
+ if ri.DeleteMarkers != 1 {
+ t.Errorf("one successful marker must be counted once in the final attempt, got %d", ri.DeleteMarkers)
+ }
+}
diff --git a/cmd/batch-authz-review4_test.go b/cmd/batch-authz-review4_test.go
new file mode 100644
index 000000000..a4b3fb02a
--- /dev/null
+++ b/cmd/batch-authz-review4_test.go
@@ -0,0 +1,157 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+ "fmt"
+ "io"
+ "net/http"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/minio/minio-go/v7"
+ "github.com/minio/minio-go/v7/pkg/credentials"
+ "github.com/minio/minio/internal/auth"
+ xhttp "github.com/minio/minio/internal/http"
+ "github.com/pgsty/silo-pkg/v3/policy"
+)
+
+// A tag removal in a newer copy must govern both the authorization and the
+// metadata sent by a push. The submitter can read data but cannot read tags.
+func TestBatchReviewPushSendsOnlyAuthorizedMetadata(t *testing.T) {
+ z, bucket := consistencyPools(t)
+ f := newBatchAuthzFixture(t)
+ const name = "push/removed-tags"
+ first := map[string]string{
+ xhttp.AmzObjectTagging: "classification=private",
+ ReservedMetadataPrefixLower + TaggingTimestamp: UTCNow().Add(-time.Hour).Format(time.RFC3339Nano),
+ }
+ second := map[string]string{
+ xhttp.AmzObjectTagging: "",
+ ReservedMetadataPrefixLower + TaggingTimestamp: UTCNow().Format(time.RFC3339Nano),
+ }
+ before := putConsistencyObject(t, z, bucket, name, 0, "test payload", ObjectOptions{Versioned: true, UserDefined: first})
+ putConsistencyObject(t, z, bucket, name, 1, "test payload", ObjectOptions{Versioned: true, VersionID: before.VersionID, MTime: before.ModTime, UserDefined: second})
+ logical, err := z.replicaObjectInfo(f.ctx, bucket, name, ObjectOptions{VersionID: before.VersionID})
+ mustIAM(t, err)
+ if logical.UserTags != "" {
+ t.Fatal("control: the authoritative version must have no tags")
+ }
+ batchAuthzPolicy(f.ctx, t, f.sys, "data-without-tags", fmt.Sprintf(`
+{"Effect":"Allow","Action":["s3:GetObject","s3:GetObjectVersion"],"Resource":["arn:aws:s3:::%s/*"]},
+{"Effect":"Deny","Action":["s3:GetObjectTagging","s3:GetObjectVersionTagging"],"Resource":["arn:aws:s3:::%s/*"]}`, bucket, bucket))
+ cred := iamAdminUser(f.ctx, t, f.sys, "data-without-tags-user", "data-without-tags")
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ if err := sub.authorize(f.ctx, batchAuthzRequest{action: policy.GetObjectTaggingAction, bucket: bucket, object: name, versionID: before.VersionID}); !isBatchJobAuthzError(err) {
+ t.Fatal("control: reading tags must be denied")
+ }
+ writes := 0
+ var sentTags string
+ target, err := minio.NewCore("target.invalid", &minio.Options{
+ Region: "us-east-1", Creds: credentials.NewStaticV4("test-access", "test-secret", ""),
+ Transport: batchLockTransport(func(r *http.Request) (*http.Response, error) {
+ if r.Body != nil {
+ _, _ = io.Copy(io.Discard, r.Body)
+ _ = r.Body.Close()
+ }
+ if r.Method == http.MethodPut {
+ writes++
+ sentTags = r.Header.Get(xhttp.AmzObjectTagging)
+ }
+ return &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Etag": {"\"test-etag\""}}, Body: io.NopCloser(strings.NewReader("")), Request: r}, nil
+ }),
+ })
+ mustIAM(t, err)
+ r := BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: bucket},
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: "remote"},
+ }
+ err = r.ReplicateToTarget(f.ctx, z, target, before, false, sub)
+ if err != nil && !isBatchJobAuthzError(err) {
+ t.Fatalf("unexpected replication failure: %v", err)
+ }
+ if writes > 0 && sentTags != logical.UserTags {
+ t.Errorf("push must send only metadata covered by authorization: writes=%d sentTags=%q authorizedTags=%q err=%v", writes, sentTags, logical.UserTags, err)
+ }
+}
+
+// LDAP's short username and its parent DN are distinct identities. The
+// background merge must not turn a deny for the verified short name into
+// a non-matching statement by replacing it with the parent DN.
+func TestBatchReviewLDAPUsernameDeny(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchAuthzPolicy(f.ctx, t, f.sys, "ldap-short-name-deny", batchAuthzFullAccess+`,
+{"Effect":"Deny","Action":["s3:DeleteObjectVersion"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"StringEquals":{"ldap:username":"alice"}}}`)
+ const dn = "uid=alice,ou=people,dc=example,dc=org"
+ cred, err := auth.GetNewCredentialsWithMetadata(map[string]any{
+ "exp": UTCNow().Add(time.Hour).Unix(), ldapUser: dn, ldapActualUser: dn, ldapUserN: "alice",
+ }, globalActiveCred.SecretKey)
+ mustIAM(t, err)
+ cred.ParentUser = dn
+ _, err = f.sys.SetTempUser(f.ctx, cred.AccessKey, cred, "ldap-short-name-deny")
+ mustIAM(t, err)
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ p, err := sub.principal(f.ctx)
+ mustIAM(t, err)
+ cred.Claims = p.claims
+ req, err := http.NewRequest(http.MethodDelete, "https://localhost/batch-authz/data", nil)
+ mustIAM(t, err)
+ args := policy.Args{
+ AccountName: cred.AccessKey, Claims: p.claims, Action: policy.DeleteObjectVersionAction,
+ BucketName: f.bucket, ObjectName: "data", ConditionValues: getConditionValues(req, "", cred),
+ }
+ if args.ConditionValues["username"][0] != "alice" || f.sys.isAllowedLocally(args) {
+ t.Fatal("control: the request evaluator must deny the verified LDAP short username")
+ }
+ if err := sub.authorize(f.ctx, batchAuthzRequest{action: policy.DeleteObjectVersionAction, bucket: f.bucket, object: "data", versionID: mustGetUUID()}); !isBatchJobAuthzError(err) {
+ t.Errorf("batch authorization must preserve the LDAP username deny or fail closed, got %v", err)
+ }
+}
+
+// Failure to read authoritative metadata is incomplete work, even if one
+// pool's copy can still be opened. In particular, skipErrs=false must not
+// turn an unreadable entry into a successful archive.
+func TestBatchReviewSnowballAuthoritativeReadFailure(t *testing.T) {
+ z, bucket := consistencyPools(t)
+ const name = "archive/unreadable-copy"
+ before := putConsistencyObject(t, z, bucket, name, 0, "test payload", ObjectOptions{Versioned: true})
+ set := z.serverPools[1].sets[0]
+ getDisks := set.getDisks
+ disks := append([]StorageAPI(nil), getDisks()...)
+ for i := range disks {
+ disks[i] = consistencyReadFaultDisk{StorageAPI: disks[i], bucket: bucket, object: name}
+ }
+ set.getDisks = func() []StorageAPI { return disks }
+ defer func() { set.getDisks = getDisks }()
+ gr, err := z.GetObjectNInfo(t.Context(), bucket, name, nil, nil, ObjectOptions{VersionID: before.VersionID})
+ mustIAM(t, err)
+ _, logicalErr := batchAuthoritativeRead(t.Context(), z, bucket, name, gr.ObjInfo)
+ gr.Close()
+ if logicalErr == nil {
+ t.Fatal("control: the authoritative read must fail")
+ }
+ remote, err := minio.New("archive.invalid", &minio.Options{
+ Region: "us-east-1", Creds: credentials.NewStaticV4("test-access", "test-secret", ""),
+ Transport: batchLockTransport(func(r *http.Request) (*http.Response, error) {
+ if r.Body != nil {
+ _, _ = io.Copy(io.Discard, r.Body)
+ }
+ return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(strings.NewReader("")), Request: r}, nil
+ }),
+ })
+ mustIAM(t, err)
+ r := BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: bucket},
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: "remote"},
+ }
+ r.Source.Snowball.InMemory = ptr(true)
+ r.Source.Snowball.Compress = ptr(false)
+ r.Source.Snowball.SkipErrs = ptr(false)
+ if err := r.writeAsArchive(t.Context(), z, remote, []ObjectInfo{before}, "", batchRootSubmitter(t)); err == nil {
+ t.Error("archive with an unreadable authoritative version must not report success")
+ }
+}
diff --git a/cmd/batch-authz-review5_test.go b/cmd/batch-authz-review5_test.go
new file mode 100644
index 000000000..ccc2520e9
--- /dev/null
+++ b/cmd/batch-authz-review5_test.go
@@ -0,0 +1,142 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "maps"
+ "net/http"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/minio/minio-go/v7"
+ "github.com/minio/minio-go/v7/pkg/credentials"
+ "github.com/minio/minio/internal/crypto"
+ xhttp "github.com/minio/minio/internal/http"
+ "github.com/minio/minio/internal/kms"
+)
+
+// A suspended destination does not keep a source version ID. Authorization
+// must therefore describe the null destination version that the write really
+// replaces, and must not let a source ID escape that model.
+func TestBatchReviewPullSuspendedVersionBinding(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ _, err := globalBucketMetadataSys.Update(f.ctx, f.bucket, bucketVersioningConfig, []byte(`Suspended`))
+ mustIAM(t, err)
+
+ sourceVersion := mustGetUUID()
+ batchAuthzPolicy(f.ctx, t, f.sys, "pull-source-version-deny", batchAuthzFullAccess+fmt.Sprintf(`,
+{"Effect":"Deny","Action":["s3:PutObject"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"StringEquals":{"s3:versionid":"%s"}}}`, sourceVersion))
+ cred := iamAdminUser(f.ctx, t, f.sys, "pull-source-version-user", "pull-source-version-deny")
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ job := BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: "remote"},
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: f.bucket},
+ }
+ err = job.ReplicateFromSource(f.ctx, f.obj, batchLockRemote(t, "replacement"), ObjectInfo{
+ Name: "pull/suspended-version", VersionID: sourceVersion, ModTime: UTCNow(), ETag: "batchetag",
+ UserDefined: map[string]string{},
+ }, false, sub)
+ if !isBatchJobAuthzError(err) {
+ t.Fatalf("a source version ID must not evade the destination deny: %v", err)
+ }
+ if _, gerr := f.obj.GetObjectInfo(f.ctx, f.bucket, "pull/suspended-version", ObjectOptions{VersionID: sourceVersion}); gerr == nil {
+ t.Fatal("a denied suspended-destination write created the source version")
+ }
+}
+
+// A failed source open in a non-skipping snowball must be reported to the
+// batch rather than silently omitted from a successful archive.
+type batchReviewOpenFailure struct {
+ ObjectLayer
+ name string
+}
+
+func (a batchReviewOpenFailure) GetObjectNInfo(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, opts ObjectOptions) (*GetObjectReader, error) {
+ if object == a.name {
+ return nil, fmt.Errorf("test: source open failed")
+ }
+ return a.ObjectLayer.GetObjectNInfo(ctx, bucket, object, rs, h, opts)
+}
+
+func TestBatchReviewSnowballSourceOpenFailure(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ name := "archive/open-failure"
+ entry := batchPutVersion(t, f.obj, f.bucket, name, nil)
+ remote, err := minio.New("archive.invalid", &minio.Options{
+ Region: "us-east-1", Creds: credentials.NewStaticV4("test-access", "test-secret", ""),
+ Transport: batchLockTransport(func(r *http.Request) (*http.Response, error) {
+ if r.Body != nil {
+ _, _ = io.Copy(io.Discard, r.Body)
+ }
+ return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(strings.NewReader("")), Request: r}, nil
+ }),
+ })
+ mustIAM(t, err)
+ r := BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: f.bucket},
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: "remote"},
+ }
+ r.Source.Snowball.InMemory = ptr(true)
+ r.Source.Snowball.Compress = ptr(false)
+ r.Source.Snowball.SkipErrs = ptr(false)
+ api := batchReviewOpenFailure{ObjectLayer: f.obj, name: name}
+ if err := r.writeAsArchive(t.Context(), api, remote, []ObjectInfo{entry}, "", batchRootSubmitter(t)); err == nil {
+ t.Fatal("a source open failure must not be reported as a successful archive")
+ }
+}
+
+// Rotation must apply tag conditions to the logical version, including
+// a more recent tag update stored in a pool other than the selected copy.
+func TestBatchReviewRotationAuthoritativeTags(t *testing.T) {
+ z, bucket := consistencyPools(t)
+ f := newBatchAuthzFixture(t)
+ previousKMS := GlobalKMS
+ GlobalKMS = kms.NewStub("batch-round5-rotation-key")
+ t.Cleanup(func() { GlobalKMS = previousKMS })
+ batchAuthzPolicy(f.ctx, t, f.sys, "round5-rotation-tags", fmt.Sprintf(`
+{"Effect":"Allow","Action":["s3:GetObject","s3:GetObjectVersion","s3:PutObject"],"Resource":["arn:aws:s3:::%s/*"]},
+{"Effect":"Deny","Action":["s3:GetObject","s3:GetObjectVersion"],"Resource":["arn:aws:s3:::%s/*"],"Condition":{"StringEquals":{"s3:ExistingObjectTag/classification":"private"}}}`, bucket, bucket))
+ cred := iamAdminUser(f.ctx, t, f.sys, "round5-rotation-user", "round5-rotation-tags")
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ for _, classification := range []string{"private", "public"} {
+ t.Run(classification, func(t *testing.T) {
+ name := "rotation/authoritative-" + classification
+ first := make(map[string]string)
+ key, err := newEncryptMetadata(f.ctx, crypto.S3, "", nil, bucket, name, first, nil)
+ mustIAM(t, err)
+ first[xhttp.AmzObjectTagging] = "classification=public"
+ first[ReservedMetadataPrefixLower+TaggingTimestamp] = UTCNow().Add(-time.Hour).Format(time.RFC3339Nano)
+ second := maps.Clone(first)
+ second[xhttp.AmzObjectTagging] = "classification=" + classification
+ second[ReservedMetadataPrefixLower+TaggingTimestamp] = UTCNow().Format(time.RFC3339Nano)
+ ciphertext, err := io.ReadAll(crypto.EncryptSinglePart(strings.NewReader("test encrypted payload"), key))
+ mustIAM(t, err)
+ before := putConsistencyObject(t, z, bucket, name, 0, string(ciphertext), ObjectOptions{Versioned: true, UserDefined: first})
+ putConsistencyObject(t, z, bucket, name, 1, string(ciphertext), ObjectOptions{Versioned: true, VersionID: before.VersionID, MTime: before.ModTime, UserDefined: second})
+ logical, err := z.replicaObjectInfo(f.ctx, bucket, name, ObjectOptions{VersionID: before.VersionID})
+ mustIAM(t, err)
+ if logical.UserTags != "classification="+classification {
+ t.Fatal("control: the second pool must govern the logical tags")
+ }
+ job := BatchJobKeyRotateV1{Bucket: bucket, Encryption: BatchJobKeyRotateEncryption{Type: sses3}}
+ err = job.KeyRotate(f.ctx, z, before, sub)
+ after, readErr := z.serverPools[0].GetObjectInfo(f.ctx, bucket, name, ObjectOptions{VersionID: before.VersionID})
+ mustIAM(t, readErr)
+ changed := after.UserDefined[crypto.MetaSealedKeyS3] != before.UserDefined[crypto.MetaSealedKeyS3]
+ if classification == "private" {
+ if !isBatchJobAuthzError(err) || changed {
+ t.Fatalf("authoritative tag deny must prevent rewrapping: err=%v changed=%v", err, changed)
+ }
+ } else if err != nil || !changed {
+ t.Fatalf("authorized unlocked version must rotate: err=%v changed=%v", err, changed)
+ }
+ })
+ }
+}
diff --git a/cmd/batch-authz-review_test.go b/cmd/batch-authz-review_test.go
new file mode 100644
index 000000000..c69ee43e4
--- /dev/null
+++ b/cmd/batch-authz-review_test.go
@@ -0,0 +1,287 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "strings"
+ "testing"
+
+ "github.com/minio/minio-go/v7"
+ "github.com/minio/minio-go/v7/pkg/credentials"
+ polplugin "github.com/minio/minio/internal/config/policy/plugin"
+ "github.com/minio/minio/internal/crypto"
+ xhttp "github.com/minio/minio/internal/http"
+ xnet "github.com/pgsty/silo-pkg/v3/net"
+ "github.com/pgsty/silo-pkg/v3/policy"
+)
+
+func batchPullJob(bucket string) BatchJobReplicateV1 {
+ return BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: "remote"},
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: bucket},
+ }
+}
+
+// A pull does not read the destination version it may replace, so a policy
+// that conditions the write on that version's tags refuses it; a version
+// condition sees the version the pull writes.
+func TestBatchPullDestinationConditions(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ action policy.Action
+ marker bool
+ condition func(versionID string) string
+ }{
+ {"tags-put", policy.ReplicateObjectAction, false, func(string) string {
+ return `{"StringEquals":{"s3:ExistingObjectTag/class":"protected"}}`
+ }},
+ {"tags-marker", policy.ReplicateDeleteAction, true, func(string) string {
+ return `{"StringEquals":{"s3:ExistingObjectTag/class":"protected"}}`
+ }},
+ {"version-put", policy.PutObjectAction, false, func(v string) string {
+ return fmt.Sprintf(`{"StringEquals":{"s3:versionid":%q}}`, v)
+ }},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ before := batchPutVersion(t, f.obj, f.bucket, "pull/guarded", map[string]string{xhttp.AmzObjectTagging: "class=protected"})
+ batchAuthzPolicy(f.ctx, t, f.sys, "pull-guard", batchAuthzFullAccess+fmt.Sprintf(`,
+{"Effect":"Deny","Action":[%q],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":%s}`, tc.action, tc.condition(before.VersionID)))
+ cred := iamAdminUser(f.ctx, t, f.sys, "pull-guard-user", "pull-guard")
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ job := batchPullJob(f.bucket)
+ err = job.ReplicateFromSource(f.ctx, f.obj, batchLockRemote(t, "replacement"), ObjectInfo{
+ Name: before.Name, VersionID: before.VersionID, ModTime: UTCNow(), ETag: "batchetag",
+ DeleteMarker: tc.marker, UserDefined: map[string]string{},
+ }, false, sub)
+ if !isBatchJobAuthzError(err) {
+ t.Errorf("pull returned %v, want a refusal", err)
+ }
+ after, err := f.obj.GetObjectInfo(f.ctx, f.bucket, before.Name, ObjectOptions{VersionID: before.VersionID})
+ if err != nil || after.ETag != before.ETag {
+ t.Errorf("refused pull changed the destination version: %v", err)
+ }
+ })
+ }
+}
+
+// A pull does not encrypt what it writes; encryption headers reported by the
+// source do not satisfy a destination policy that requires encryption.
+func TestBatchPullEncryptionCondition(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchAuthzPolicy(f.ctx, t, f.sys, "pull-encryption", `
+{"Effect":"Allow","Action":["s3:ReplicateObject"],"Resource":["arn:aws:s3:::batch-authz/*"]},
+{"Effect":"Allow","Action":["s3:PutObject"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"StringEquals":{"s3:x-amz-server-side-encryption":"aws:kms","s3:x-amz-server-side-encryption-aws-kms-key-id":"required-key"}}}`)
+ cred := iamAdminUser(f.ctx, t, f.sys, "pull-encryption-user", "pull-encryption")
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ job := batchPullJob(f.bucket)
+ version := mustGetUUID()
+ err = job.ReplicateFromSource(f.ctx, f.obj, batchLockRemote(t, "content"), ObjectInfo{
+ Name: "pull/encrypted", VersionID: version, ModTime: UTCNow(), ETag: "batchetag",
+ UserDefined: map[string]string{xhttp.AmzServerSideEncryption: "aws:kms", xhttp.AmzServerSideEncryptionKmsID: "required-key"},
+ }, false, sub)
+ if err == nil {
+ oi, gerr := f.obj.GetObjectInfo(f.ctx, f.bucket, "pull/encrypted", ObjectOptions{VersionID: version})
+ mustIAM(t, gerr)
+ if !crypto.S3KMS.IsEncrypted(oi.UserDefined) {
+ t.Fatal("a pull was allowed by an encryption condition it did not meet")
+ }
+ }
+}
+
+// A conditional Deny on an S3 Tables action covers the S3 data actions it
+// implies, under the tables resource form; the job refuses rather than skip it.
+func TestBatchJobTableConditionalDeny(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchAuthzPolicy(f.ctx, t, f.sys, "table-context", batchAuthzFullAccess+`,
+{"Effect":"Deny","Action":["s3tables:GetTableData"],"Resource":["arn:aws:s3tables:::bucket/batch-authz/table/table-id"],"Condition":{"Bool":{"aws:SecureTransport":"false"}}}`)
+ cred := iamAdminUser(f.ctx, t, f.sys, "table-context-user", "table-context")
+ if f.sys.isAllowedLocally(policy.Args{AccountName: cred.AccessKey, Action: policy.GetObjectAction, BucketName: f.bucket, ObjectName: "table-id/data", ConditionValues: map[string][]string{"SecureTransport": {"false"}}}) {
+ t.Fatal("control: with the request context the Deny applies")
+ }
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ if err := sub.authorize(f.ctx, batchAuthzRequest{action: policy.GetObjectAction, bucket: f.bucket, object: "table-id/data"}); !isBatchJobAuthzError(err) {
+ t.Fatalf("authorize returned %v, want a refusal", err)
+ }
+}
+
+type batchArchiveRevocation struct {
+ ObjectLayer
+ revoke func()
+}
+
+func (a *batchArchiveRevocation) GetObjectNInfo(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, opts ObjectOptions) (*GetObjectReader, error) {
+ if object == "archive/second" {
+ a.revoke()
+ }
+ return a.ObjectLayer.GetObjectNInfo(ctx, bucket, object, rs, h, opts)
+}
+
+// Each snowball entry is authorized for the submitter as it is then.
+func TestBatchPushSnowballEntryRevalidation(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ cred := iamAdminUser(f.ctx, t, f.sys, "archive-user", "")
+ sub, err := newBatchJobSubmitter(cred)
+ mustIAM(t, err)
+ pluginURL, err := xnet.ParseHTTPURL("https://authz.invalid/")
+ mustIAM(t, err)
+ previous := newGlobalAuthZPluginFn()
+ setGlobalAuthZPlugin(polplugin.New(polplugin.Args{
+ URL: pluginURL, CloseRespFn: func(r io.ReadCloser) { _ = r.Close() },
+ Transport: batchLockTransport(func(r *http.Request) (*http.Response, error) {
+ return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"result":true}`)), Request: r}, nil
+ }),
+ }))
+ t.Cleanup(func() { setGlobalAuthZPlugin(previous) })
+ entries := []ObjectInfo{
+ batchPutVersion(t, f.obj, f.bucket, "archive/first", nil),
+ batchPutVersion(t, f.obj, f.bucket, "archive/second", nil),
+ }
+ remote, err := minio.New("archive.invalid", &minio.Options{
+ Region: "us-east-1", Creds: credentials.NewStaticV4("batch-access", "batch-secret", ""),
+ Transport: batchLockTransport(func(r *http.Request) (*http.Response, error) {
+ if r.Body != nil {
+ _, _ = io.Copy(io.Discard, r.Body)
+ }
+ return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(strings.NewReader("")), Request: r}, nil
+ }),
+ })
+ mustIAM(t, err)
+ job := BatchJobReplicateV1{Source: BatchJobReplicateSource{Bucket: f.bucket}, Target: BatchJobReplicateTarget{Bucket: "remote"}}
+ job.Source.Snowball.InMemory = ptr(true)
+ job.Source.Snowball.Compress = ptr(false)
+ job.Source.Snowball.SkipErrs = ptr(false)
+ api := &batchArchiveRevocation{ObjectLayer: f.obj, revoke: func() { mustIAM(t, f.sys.DeleteUser(f.ctx, cred.AccessKey, false)) }}
+ if err = job.writeAsArchive(f.ctx, api, remote, entries, "", sub); !isBatchJobAuthzError(err) {
+ t.Fatalf("archive with a revoked submitter returned %v, want a refusal", err)
+ }
+}
+
+// A definition holds one job type; one with several is refused, so that what
+// is authorized is what runs.
+func TestBatchJobSeveralSections(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ older := f.putVersions(t, "data/a")
+ job := f.expireYAML("") + `keyrotate:
+ apiVersion: v1
+ bucket: other
+ encryption:
+ type: sse-s3
+`
+ rec, queued := f.submit(t, globalActiveCred, job)
+ if queued != nil {
+ f.run(t, queued)
+ }
+ if rec.Code != http.StatusBadRequest {
+ t.Errorf("submission status %d, want 400: %s", rec.Code, rec.Body.String())
+ }
+ if !f.versionExists(t, "data/a", older) {
+ t.Fatal("a definition with two job types ran")
+ }
+}
+
+// Prefixes are walked once each in one order: a cursor in a later prefix is
+// after every key of the earlier ones, and an overlapping prefix is not
+// walked twice.
+func TestBatchExpirePrefixes(t *testing.T) {
+ if got := batchCanonicalPrefixes([]string{"z/", "a/", "a/b/", "z/", ""}); len(got) != 1 || got[0] != "" {
+ t.Fatalf("canonical prefixes %q", got)
+ }
+ if got := batchCanonicalPrefixes([]string{"z/", "a/b/", "a/"}); strings.Join(got, ",") != "a/,z/" {
+ t.Fatalf("canonical prefixes %q", got)
+ }
+ f := newBatchAuthzFixture(t)
+ done := f.putVersions(t, "a/done")
+ todo := f.putVersions(t, "z/todo")
+ batchPutVersion(t, f.obj, f.bucket, "x/data", batchLockMeta("LEGALHOLD"))
+ batchPutVersion(t, f.obj, f.bucket, "x/data", nil)
+ rec, job := f.submit(t, globalActiveCred, fmt.Sprintf(`expire:
+ apiVersion: v1
+ bucket: %s
+ prefix: ["z/", "x/", "a/", "x/"]
+ rules:
+ - type: object
+ purge:
+ retainVersions: 1
+`, f.bucket))
+ if job == nil {
+ t.Fatalf("submission status %d: %s", rec.Code, rec.Body.String())
+ }
+ // A run that completed a/ and stopped.
+ ri := &batchJobInfo{JobID: job.ID, JobType: string(job.Type()), StartTime: job.Started, Bucket: f.bucket, Object: "a/done"}
+ mustIAM(t, ri.updateAfter(f.ctx, f.obj, 0, *job))
+ f.run(t, job)
+ if !f.versionExists(t, "a/done", done) {
+ t.Error("resume did again a key before the cursor")
+ }
+ if f.versionExists(t, "z/todo", todo) {
+ t.Error("resume skipped a later prefix")
+ }
+ if got := f.jobInfo(t, job).VersionsSkippedLocked; got != 1 {
+ t.Errorf("a locked version under two equal prefixes was counted %d times", got)
+ }
+}
+
+type batchPartialWalk struct {
+ ObjectLayer
+ version ObjectInfo
+ cancel context.CancelFunc
+ read chan struct{}
+}
+
+func (a *batchPartialWalk) Walk(_ context.Context, _ string, _ string, results chan<- itemOrErr[ObjectInfo], _ WalkOptions) error {
+ go func() {
+ defer close(results)
+ results <- itemOrErr[ObjectInfo]{Item: a.version}
+ if a.cancel != nil {
+ <-a.read
+ a.cancel()
+ return
+ }
+ results <- itemOrErr[ObjectInfo]{Err: errors.New("test: listing failed within a key")}
+ }()
+ return nil
+}
+
+func (a *batchPartialWalk) GetObjectInfo(context.Context, string, string, ObjectOptions) (ObjectInfo, error) {
+ if a.read != nil {
+ close(a.read)
+ }
+ return a.version, nil
+}
+
+// A rotation whose walk fails or is canceled within a key keeps that key
+// after its cursor and does not report completion.
+func TestBatchKeyRotationIncompleteWalk(t *testing.T) {
+ for _, canceled := range []bool{false, true} {
+ t.Run(fmt.Sprintf("canceled=%v", canceled), func(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ ctx, cancel := context.WithCancel(f.ctx)
+ defer cancel()
+ meta := batchLockMeta("LEGALHOLD")
+ meta[crypto.MetaSealedKeyS3] = "test-sealed-key"
+ api := &batchPartialWalk{ObjectLayer: f.obj, version: ObjectInfo{Bucket: f.bucket, Name: "rotation/key", VersionID: mustGetUUID(), UserDefined: meta}}
+ if canceled {
+ api.cancel, api.read = cancel, make(chan struct{})
+ }
+ job := BatchJobRequest{
+ ID: "keyrotate-partial" + getKeySeparator() + "0", Started: UTCNow(), Submitter: batchRootSubmitter(t),
+ KeyRotate: &BatchJobKeyRotateV1{Bucket: f.bucket, Encryption: BatchJobKeyRotateEncryption{Type: sses3}},
+ }
+ _ = job.KeyRotate.Start(ctx, api, job)
+ ri := f.jobInfo(t, &job)
+ if ri.Complete || ri.Object == api.version.Name || ri.VersionsSkippedLocked != 0 {
+ t.Fatalf("incomplete key committed: complete=%v cursor=%q skips=%d", ri.Complete, ri.Object, ri.VersionsSkippedLocked)
+ }
+ })
+ }
+}
diff --git a/cmd/batch-object-lock_test.go b/cmd/batch-object-lock_test.go
new file mode 100644
index 000000000..dc3096c0b
--- /dev/null
+++ b/cmd/batch-object-lock_test.go
@@ -0,0 +1,479 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+ "bytes"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/minio/madmin-go/v3"
+ "github.com/minio/minio-go/v7"
+ "github.com/minio/minio-go/v7/pkg/credentials"
+ "github.com/minio/minio/internal/amztime"
+ "github.com/minio/minio/internal/auth"
+ objectlock "github.com/minio/minio/internal/bucket/object/lock"
+ "github.com/minio/minio/internal/crypto"
+ xhttp "github.com/minio/minio/internal/http"
+ "github.com/minio/minio/internal/kms"
+ "github.com/pgsty/silo-pkg/v3/policy"
+)
+
+var batchLockKinds = []string{"COMPLIANCE", "GOVERNANCE", "LEGALHOLD"}
+
+func batchLockMeta(kind string) map[string]string {
+ m := map[string]string{}
+ switch kind {
+ case "COMPLIANCE", "GOVERNANCE":
+ m[strings.ToLower(xhttp.AmzObjectLockMode)] = kind
+ m[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(UTCNow().Add(24 * time.Hour))
+ case "LEGALHOLD":
+ m[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = "ON"
+ }
+ return m
+}
+
+func batchPutVersion(t *testing.T, obj ObjectLayer, bucket, key string, meta map[string]string) ObjectInfo {
+ t.Helper()
+ data := []byte("batch lock " + key)
+ oi, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""),
+ ObjectOptions{Versioned: true, UserDefined: meta})
+ if err != nil {
+ t.Fatal(err)
+ }
+ return oi
+}
+
+func batchVersionExists(t *testing.T, obj ObjectLayer, bucket, key, versionID string) bool {
+ t.Helper()
+ _, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: versionID})
+ switch {
+ case err == nil:
+ return true
+ case isErrVersionNotFound(err) || isErrObjectNotFound(err):
+ return false
+ }
+ t.Fatal(err)
+ return false
+}
+
+// batchJobStatusJSON returns the status response of a job as decoded JSON.
+func batchJobStatusJSON(t *testing.T, jobID string) map[string]any {
+ t.Helper()
+ cred := globalActiveCred
+ req, err := newTestSignedRequestV4(http.MethodGet, "http://localhost/minio/admin/v3/status-job?jobId="+jobID, 0, nil, cred.AccessKey, cred.SecretKey, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err = req.ParseForm(); err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ req = req.WithContext(newContext(req, rec, "BatchJobStatus"))
+ adminAPIHandlers{}.BatchJobStatus(rec, req)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+ }
+ var out struct{ LastMetric map[string]any }
+ if err = json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
+ t.Fatal(err)
+ }
+ return out.LastMetric
+}
+
+// Batch expiration keeps versions under retention or legal hold, on both
+// purge paths, and reports them; unlocked versions are still expired.
+func TestBatchExpireObjectLock(t *testing.T) {
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t, endpoints: []string{"PutObject"},
+ objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) {
+ defer r5Capacity(obj.(*erasureServerPools))()
+ enableBucketObjectLock(t, bucket)
+ f := &batchAuthzFixture{ctx: t.Context(), obj: obj, bucket: bucket}
+ batchTestJobPool(t.Context(), t, obj)
+
+ type version struct{ key, id string }
+ var locked, unlocked []version
+ for _, retain := range []int{0, 1} {
+ for _, kind := range batchLockKinds {
+ key := fmt.Sprintf("r%d/%s", retain, kind)
+ older := batchPutVersion(t, obj, bucket, key, batchLockMeta(kind))
+ batchPutVersion(t, obj, bucket, key, nil)
+ locked = append(locked, version{key, older.VersionID})
+ }
+ key := fmt.Sprintf("r%d/plain", retain)
+ older := batchPutVersion(t, obj, bucket, key, nil)
+ batchPutVersion(t, obj, bucket, key, nil)
+ unlocked = append(unlocked, version{key, older.VersionID})
+ }
+
+ var jobs []*BatchJobRequest
+ for _, retain := range []int{0, 1} {
+ job := fmt.Sprintf(`expire:
+ apiVersion: v1
+ bucket: %s
+ prefix: r%d/
+ rules:
+ - type: object
+ purge:
+ retainVersions: %d
+`, bucket, retain, retain)
+ rec, queued := f.submit(t, globalActiveCred, job)
+ if queued == nil {
+ t.Fatalf("%s: submission status %d: %s", instance, rec.Code, rec.Body.String())
+ }
+ // Running a job again, as a resumed job does, must not count
+ // the versions it left a second time.
+ f.run(t, queued)
+ f.run(t, queued)
+ jobs = append(jobs, queued)
+ }
+
+ for _, v := range locked {
+ if !batchVersionExists(t, obj, bucket, v.key, v.id) {
+ t.Errorf("%s: %s: locked version %s was expired", instance, v.key, v.id)
+ }
+ }
+ for _, v := range unlocked {
+ if batchVersionExists(t, obj, bucket, v.key, v.id) {
+ t.Errorf("%s: %s: control: unlocked version %s was not expired", instance, v.key, v.id)
+ }
+ }
+ for i, field := range []string{"keysRefused", "versionsSkippedLocked"} {
+ m := batchJobStatusJSON(t, jobs[i].ID)
+ if got, _ := m[field].(float64); got != float64(len(batchLockKinds)) {
+ t.Errorf("%s: retain %d: %s = %v, want %d", instance, i, field, m[field], len(batchLockKinds))
+ }
+ if m["complete"] != true || m["failed"] != false {
+ t.Errorf("%s: retain %d: complete=%v failed=%v", instance, i, m["complete"], m["failed"])
+ }
+ }
+ },
+ })
+}
+
+type batchLockTransport func(*http.Request) (*http.Response, error)
+
+func (f batchLockTransport) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
+
+// batchLockRemote is a source that serves replacement for every read.
+func batchLockRemote(t *testing.T, replacement string) *minio.Core {
+ t.Helper()
+ core, err := minio.NewCore("source.invalid", &minio.Options{
+ Region: "us-east-1", Creds: credentials.NewStaticV4("batch-access", "batch-secret", ""),
+ Transport: batchLockTransport(func(r *http.Request) (*http.Response, error) {
+ return &http.Response{StatusCode: http.StatusOK, Header: http.Header{
+ "Content-Length": []string{strconv.Itoa(len(replacement))},
+ "Last-Modified": []string{UTCNow().Format(http.TimeFormat)},
+ "Etag": []string{"\"batchetag\""},
+ }, Body: io.NopCloser(strings.NewReader(replacement)), ContentLength: int64(len(replacement)), Request: r}, nil
+ }),
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ return core
+}
+
+func batchRootSubmitter(t *testing.T) *batchJobSubmitter {
+ t.Helper()
+ sub, err := newBatchJobSubmitter(globalActiveCred)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return sub
+}
+
+// A pull leaves a destination version under Object Lock as it is, whether a
+// single PUT, a multipart completion or a delete marker would replace it.
+func TestBatchPullObjectLock(t *testing.T) {
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t, endpoints: []string{"PutObject"},
+ objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) {
+ defer r5Capacity(obj.(*erasureServerPools))()
+ enableBucketObjectLock(t, bucket)
+ const replacement = "replacement from remote"
+ core := batchLockRemote(t, replacement)
+ sub := batchRootSubmitter(t)
+ job := BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: "remote-bucket"},
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: bucket},
+ }
+ for _, kind := range batchLockKinds {
+ for _, path := range []string{"put", "multipart", "marker", "directory"} {
+ key := "pull/" + kind + "/" + path
+ if path == "directory" {
+ key += "/"
+ }
+ before := batchPutVersion(t, obj, bucket, key, batchLockMeta(kind))
+ src := ObjectInfo{Name: key, VersionID: before.VersionID, ModTime: UTCNow(), ETag: "batchetag", UserDefined: map[string]string{}}
+ switch path {
+ case "multipart":
+ src.ETag = "batchetag-1"
+ case "marker":
+ src.DeleteMarker = true
+ }
+ err := job.ReplicateFromSource(t.Context(), obj, core, src, false, sub)
+ if !isErrObjectLocked(err) {
+ t.Errorf("%s: %s: pull returned %v, want ObjectLocked", instance, key, err)
+ }
+ after, gerr := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: before.VersionID})
+ if gerr != nil {
+ t.Errorf("%s: %s: locked version lost: %v", instance, key, gerr)
+ continue
+ }
+ if !enforceRetentionForDeletion(t.Context(), after) || after.ETag != before.ETag {
+ t.Errorf("%s: %s: locked version was replaced", instance, key)
+ }
+ }
+ }
+
+ // Controls: a new version, and a new delete marker that hides a
+ // locked version without replacing it.
+ locked := batchPutVersion(t, obj, bucket, "pull/new", batchLockMeta("COMPLIANCE"))
+ newVersion := mustGetUUID()
+ if err := job.ReplicateFromSource(t.Context(), obj, core, ObjectInfo{
+ Name: "pull/new", VersionID: newVersion, ModTime: UTCNow(), ETag: "batchetag", UserDefined: map[string]string{},
+ }, false, sub); err != nil {
+ t.Fatalf("%s: new version: %v", instance, err)
+ }
+ if got := readCopyChecksumObject(t, obj, bucket, "pull/new", ObjectOptions{VersionID: newVersion}); string(got) != replacement {
+ t.Errorf("%s: new version content %q", instance, got)
+ }
+ marker := mustGetUUID()
+ if err := job.ReplicateFromSource(t.Context(), obj, core, ObjectInfo{
+ Name: "pull/new", VersionID: marker, ModTime: UTCNow(), DeleteMarker: true,
+ }, false, sub); err != nil {
+ t.Fatalf("%s: new marker: %v", instance, err)
+ }
+ if !batchVersionExists(t, obj, bucket, "pull/new", locked.VersionID) {
+ t.Errorf("%s: a new marker removed a locked version", instance)
+ }
+ },
+ })
+}
+
+// A pull applies retention as a PUT to the destination would: copied
+// retention is validated, and a version without one gets the bucket default.
+func TestBatchPullRetentionMetadata(t *testing.T) {
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t, endpoints: []string{"PutObject"},
+ objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) {
+ defer r5Capacity(obj.(*erasureServerPools))()
+ meta, err := globalBucketMetadataSys.Get(bucket)
+ if err != nil {
+ t.Fatal(err)
+ }
+ meta.ObjectLockConfigXML = []byte(`EnabledGOVERNANCE1`)
+ meta.VersioningConfigXML = enabledBucketVersioningConfig
+ if err = meta.parseAllConfigs(t.Context(), obj); err != nil {
+ t.Fatal(err)
+ }
+ globalBucketMetadataSys.Set(bucket, meta)
+ core := batchLockRemote(t, "retention content")
+ sub := batchRootSubmitter(t)
+ job := BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: "remote-bucket"},
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: bucket},
+ }
+ pull := func(key string, meta map[string]string) (ObjectInfo, error) {
+ vid := mustGetUUID()
+ err := job.ReplicateFromSource(t.Context(), obj, core, ObjectInfo{Name: key, VersionID: vid, ModTime: UTCNow().Add(-time.Hour), ETag: "batchetag", UserDefined: meta}, false, sub)
+ if err != nil {
+ return ObjectInfo{}, err
+ }
+ return obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: vid})
+ }
+ oi, err := pull("retention/default", map[string]string{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ ret := objectlock.GetObjectRetentionMeta(oi.UserDefined)
+ if ret.Mode != objectlock.RetGovernance || !ret.RetainUntilDate.After(UTCNow().Add(23*time.Hour)) {
+ t.Errorf("%s: default retention not applied with the destination's time: %+v", instance, ret)
+ }
+ until := UTCNow().Add(48 * time.Hour)
+ oi, err = pull("retention/copied", map[string]string{
+ xhttp.AmzObjectLockMode: "COMPLIANCE", xhttp.AmzObjectLockRetainUntilDate: amztime.ISO8601Format(until),
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if ret = objectlock.GetObjectRetentionMeta(oi.UserDefined); ret.Mode != objectlock.RetCompliance || !ret.RetainUntilDate.Equal(until.Truncate(time.Millisecond)) {
+ t.Errorf("%s: copied retention: %+v", instance, ret)
+ }
+ if _, err = pull("retention/invalid", map[string]string{xhttp.AmzObjectLockMode: "FOREVER", xhttp.AmzObjectLockRetainUntilDate: amztime.ISO8601Format(until)}); err == nil {
+ t.Errorf("%s: invalid copied retention was accepted", instance)
+ }
+ },
+ })
+}
+
+// Key rotation does not rewrite a version under Object Lock.
+func TestBatchKeyRotationObjectLock(t *testing.T) {
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t, endpoints: []string{"PutObject", "GetObject"},
+ objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) {
+ defer r5Capacity(obj.(*erasureServerPools))()
+ enableBucketObjectLock(t, bucket)
+ oldKMS := GlobalKMS
+ GlobalKMS = kms.NewStub("batch-lock-key")
+ defer func() { GlobalKMS = oldKMS }()
+ sub := batchRootSubmitter(t)
+ job := BatchJobKeyRotateV1{Bucket: bucket, Encryption: BatchJobKeyRotateEncryption{Type: sses3}}
+ for _, kind := range append([]string{"none"}, batchLockKinds...) {
+ key := "rotate/" + kind
+ headers := map[string]string{xhttp.AmzServerSideEncryption: xhttp.AmzEncryptionAES}
+ switch kind {
+ case "COMPLIANCE", "GOVERNANCE":
+ headers[xhttp.AmzObjectLockMode] = kind
+ headers[xhttp.AmzObjectLockRetainUntilDate] = UTCNow().Add(24 * time.Hour).Format(time.RFC3339)
+ case "LEGALHOLD":
+ headers[xhttp.AmzObjectLockLegalHold] = "ON"
+ }
+ putCopyChecksumSource(t, router, cred, bucket, key, []byte("rotate "+kind), headers)
+ before, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{})
+ if err != nil || !crypto.S3.IsEncrypted(before.UserDefined) {
+ t.Fatalf("%s: %s: encrypted control: %v", instance, key, err)
+ }
+ err = job.KeyRotate(t.Context(), obj, before, sub)
+ after, gerr := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: before.VersionID})
+ if gerr != nil {
+ t.Fatal(gerr)
+ }
+ if kind == "none" {
+ if err != nil || after.ModTime.Equal(before.ModTime) {
+ t.Errorf("%s: control: unlocked version not rotated: %v", instance, err)
+ }
+ continue
+ }
+ var locked ObjectLocked
+ if !errors.As(err, &locked) || !after.ModTime.Equal(before.ModTime) {
+ t.Errorf("%s: %s: rotation of a locked version returned %v; rewritten=%v", instance, key, err, !after.ModTime.Equal(before.ModTime))
+ }
+ }
+ },
+ })
+}
+
+// A push reads locked versions as any authorized read does, and a submitter
+// that may not read a version does not send it.
+func TestBatchPushReads(t *testing.T) {
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t, endpoints: []string{"PutObject"},
+ objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) {
+ defer r5Capacity(obj.(*erasureServerPools))()
+ enableBucketObjectLock(t, bucket)
+ var puts []string
+ target, err := minio.NewCore("target.invalid", &minio.Options{
+ Region: "us-east-1", Creds: credentials.NewStaticV4("batch-access", "batch-secret", ""),
+ Transport: batchLockTransport(func(r *http.Request) (*http.Response, error) {
+ if r.Body != nil {
+ io.Copy(io.Discard, r.Body)
+ r.Body.Close()
+ }
+ if r.Method == http.MethodPut {
+ puts = append(puts, r.URL.Path)
+ }
+ return &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Etag": []string{"\"pushetag\""}}, Body: io.NopCloser(strings.NewReader("")), Request: r}, nil
+ }),
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ job := BatchJobReplicateV1{
+ Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: bucket},
+ Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: "remote-bucket"},
+ }
+ src := batchPutVersion(t, obj, bucket, "push/locked", batchLockMeta("COMPLIANCE"))
+ if err := job.ReplicateToTarget(t.Context(), obj, target, src, false, batchRootSubmitter(t)); err != nil {
+ t.Fatalf("%s: authorized push of a locked version: %v", instance, err)
+ }
+ if len(puts) != 1 {
+ t.Fatalf("%s: authorized push sent %d writes", instance, len(puts))
+ }
+
+ // A user that may list but not read.
+ p, err := policy.ParseConfig(strings.NewReader(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:StartBatchJob"]},{"Effect":"Allow","Action":["s3:ListBucket","s3:ListBucketVersions"],"Resource":["arn:aws:s3:::` + bucket + `"]}]}`))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err = globalIAMSys.SetPolicy(t.Context(), "batch-list-only", *p); err != nil {
+ t.Fatal(err)
+ }
+ user := auth.Credentials{AccessKey: "batch-list-user", SecretKey: "batch-list-user-secret"}
+ if _, err = globalIAMSys.CreateUser(t.Context(), user.AccessKey, madmin.AddOrUpdateUserReq{SecretKey: user.SecretKey, Status: madmin.AccountEnabled}); err != nil {
+ t.Fatal(err)
+ }
+ if _, err = globalIAMSys.PolicyDBSet(t.Context(), user.AccessKey, "batch-list-only", regUser, false); err != nil {
+ t.Fatal(err)
+ }
+ sub, err := newBatchJobSubmitter(globalIAMSysUser(t, user.AccessKey))
+ if err != nil {
+ t.Fatal(err)
+ }
+ err = job.ReplicateToTarget(t.Context(), obj, target, src, false, sub)
+ if !isBatchJobAuthzError(err) || len(puts) != 1 {
+ t.Fatalf("%s: unauthorized push returned %v and sent %d writes", instance, err, len(puts)-1)
+ }
+ },
+ })
+}
+
+// globalIAMSysUser returns the stored credential of accessKey, as a request
+// authenticated with it has.
+func globalIAMSysUser(t *testing.T, accessKey string) auth.Credentials {
+ t.Helper()
+ u, ok := globalIAMSys.store.GetUser(accessKey)
+ if !ok {
+ t.Fatalf("no user %s", accessKey)
+ }
+ return u.Credentials
+}
+
+// The cursor of a walk done out of order moves only past keys whose items
+// are all done, and counts each item once.
+func TestBatchProgressCursor(t *testing.T) {
+ ri := &batchJobInfo{}
+ p := newBatchProgress(ri, "bucket")
+ items := []string{"a", "a", "b", "c", "c"}
+ seqs := make([]uint64, len(items))
+ for i := range items {
+ seqs[i] = p.begin()
+ }
+ skip := batchCounts{versionsSkipped: 1}
+ p.done(seqs[1], "a", skip)
+ p.done(seqs[3], "c", skip)
+ if ri.Object != "" || ri.VersionsSkippedLocked != 0 {
+ t.Fatalf("cursor moved before the first item: %q %d", ri.Object, ri.VersionsSkippedLocked)
+ }
+ p.done(seqs[0], "a", skip)
+ p.done(seqs[2], "b", skip)
+ // a is done; b is done but c may still have items.
+ if ri.Object != "b" && ri.Object != "a" {
+ t.Fatalf("cursor %q", ri.Object)
+ }
+ if ri.Object == "a" && ri.VersionsSkippedLocked != 2 {
+ t.Fatalf("counts %d at cursor a", ri.VersionsSkippedLocked)
+ }
+ // c's last item never finishes, as when a run stops: the cursor stays
+ // before c and c is not counted.
+ p.finish()
+ if ri.Object != "b" || ri.VersionsSkippedLocked != 3 {
+ t.Fatalf("interrupted: cursor %q counts %d, want b 3", ri.Object, ri.VersionsSkippedLocked)
+ }
+ p.done(seqs[4], "c", skip)
+ p.finish()
+ if ri.Object != "c" || ri.VersionsSkippedLocked != 5 {
+ t.Fatalf("finished: cursor %q counts %d, want c 5", ri.Object, ri.VersionsSkippedLocked)
+ }
+}
diff --git a/cmd/batch-submitter-authz_test.go b/cmd/batch-submitter-authz_test.go
new file mode 100644
index 000000000..64a16d97c
--- /dev/null
+++ b/cmd/batch-submitter-authz_test.go
@@ -0,0 +1,456 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+ "bytes"
+ "context"
+ "encoding/base64"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/minio/madmin-go/v3"
+ "github.com/minio/minio/internal/auth"
+ "github.com/pgsty/silo-pkg/v3/policy"
+ "go.yaml.in/yaml/v3"
+)
+
+// batchAuthzFixture runs batch jobs submitted through the admin API, one job
+// at a time and in the test's goroutine.
+type batchAuthzFixture struct {
+ ctx context.Context
+ sys *IAMSys
+ obj ObjectLayer
+ bucket string
+}
+
+func newBatchAuthzFixture(t *testing.T) *batchAuthzFixture {
+ t.Helper()
+ ctx, sys := iamAdminFixture(t)
+ obj := newObjectLayerFn()
+ batchTestJobPool(ctx, t, obj)
+ f := &batchAuthzFixture{ctx: ctx, sys: sys, obj: obj, bucket: "batch-authz"}
+ mustIAM(t, obj.MakeBucket(ctx, f.bucket, MakeBucketOptions{VersioningEnabled: true}))
+ _, err := globalBucketMetadataSys.Update(ctx, f.bucket, bucketVersioningConfig, enabledBucketVersioningConfig)
+ mustIAM(t, err)
+ return f
+}
+
+// batchTestJobPool queues submitted jobs without running them.
+func batchTestJobPool(ctx context.Context, t *testing.T, obj ObjectLayer) {
+ t.Helper()
+ prevPool := globalBatchJobPool
+ globalBatchJobPool = &BatchJobPool{
+ ctx: ctx,
+ objLayer: obj,
+ jobCh: make(chan *BatchJobRequest, 10),
+ jobCancelers: make(map[string]context.CancelFunc),
+ }
+ globalBatchJobsMetrics.Lock()
+ globalBatchJobsMetrics.metrics = make(map[string]*batchJobInfo)
+ globalBatchJobsMetrics.Unlock()
+ t.Cleanup(func() { globalBatchJobPool = prevPool })
+}
+
+// putVersions writes two versions of key and returns the older one.
+func (f *batchAuthzFixture) putVersions(t *testing.T, key string) string {
+ t.Helper()
+ var older string
+ for i := range 2 {
+ data := []byte(fmt.Sprintf("version %d", i))
+ oi, err := f.obj.PutObject(f.ctx, f.bucket, key, mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), ObjectOptions{Versioned: true})
+ mustIAM(t, err)
+ if i == 0 {
+ older = oi.VersionID
+ }
+ }
+ return older
+}
+
+func (f *batchAuthzFixture) versionExists(t *testing.T, key, versionID string) bool {
+ t.Helper()
+ _, err := f.obj.GetObjectInfo(f.ctx, f.bucket, key, ObjectOptions{VersionID: versionID})
+ if err == nil {
+ return true
+ }
+ if isErrVersionNotFound(err) || isErrObjectNotFound(err) {
+ return false
+ }
+ t.Fatal(err)
+ return false
+}
+
+func (f *batchAuthzFixture) expireYAML(prefix string) string {
+ return fmt.Sprintf(`expire:
+ apiVersion: v1
+ bucket: %s
+ prefix: %q
+ rules:
+ - type: object
+ purge:
+ retainVersions: 1
+`, f.bucket, prefix)
+}
+
+// submit sends a job definition to StartBatchJob as cred. An accepted job is
+// returned from the queue, not run.
+func (f *batchAuthzFixture) submit(t *testing.T, cred auth.Credentials, job string) (*httptest.ResponseRecorder, *BatchJobRequest) {
+ t.Helper()
+ headers := map[string]string{}
+ if cred.IsTemp() && !cred.IsServiceAccount() {
+ headers["X-Amz-Security-Token"] = cred.SessionToken
+ }
+ body := []byte(job)
+ req, err := newTestSignedRequestV4(http.MethodPost, "http://localhost/minio/admin/v3/start-job", int64(len(body)), bytes.NewReader(body), cred.AccessKey, cred.SecretKey, headers)
+ mustIAM(t, err)
+ mustIAM(t, req.ParseForm())
+ rec := httptest.NewRecorder()
+ req = req.WithContext(newContext(req, rec, "StartBatchJob"))
+ adminAPIHandlers{}.StartBatchJob(rec, req)
+ if rec.Code != http.StatusOK {
+ return rec, nil
+ }
+ select {
+ case queued := <-globalBatchJobPool.jobCh:
+ return rec, queued
+ default:
+ t.Fatal("accepted job was not queued")
+ return rec, nil
+ }
+}
+
+func (f *batchAuthzFixture) run(t *testing.T, job *BatchJobRequest) {
+ t.Helper()
+ ctx := job.ctx
+ if ctx == nil {
+ ctx = f.ctx
+ }
+ switch {
+ case job.Expire != nil:
+ if err := job.Expire.Start(ctx, f.obj, *job); err != nil {
+ t.Logf("start: %v", err)
+ }
+ default:
+ t.Fatalf("unexpected job type %s", job.Type())
+ }
+ ri := &batchJobInfo{JobID: job.ID, JobType: string(job.Type())}
+ if err := ri.load(f.ctx, f.obj, *job); err == nil {
+ t.Logf("job %s: %+v", job.ID, ri.clone())
+ }
+}
+
+func (f *batchAuthzFixture) jobInfo(t *testing.T, job *BatchJobRequest) *batchJobInfo {
+ t.Helper()
+ ri := &batchJobInfo{JobID: job.ID, JobType: string(job.Type())}
+ mustIAM(t, ri.load(f.ctx, f.obj, *job))
+ return ri
+}
+
+func batchAuthzPolicy(ctx context.Context, t *testing.T, sys *IAMSys, name, statements string) {
+ t.Helper()
+ p, err := policy.ParseConfig(strings.NewReader(`{"Version":"2012-10-17","Statement":[` + statements + `]}`))
+ mustIAM(t, err)
+ _, err = sys.SetPolicy(ctx, name, *p)
+ mustIAM(t, err)
+}
+
+const batchAuthzFullAccess = `{"Effect":"Allow","Action":["admin:StartBatchJob"]},
+{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::batch-authz","arn:aws:s3:::batch-authz/*"]}`
+
+// A credential that may start batch jobs but has no access to the bucket
+// cannot delete through a job.
+func TestBatchJobSubmitterWithoutDataAccess(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchAuthzPolicy(f.ctx, t, f.sys, "batch-only", `{"Effect":"Allow","Action":["admin:StartBatchJob","admin:ListBatchJobs"]}`)
+ cred := iamAdminUser(f.ctx, t, f.sys, "batch-only-user", "batch-only")
+ older := f.putVersions(t, "data/a")
+ rec, job := f.submit(t, cred, f.expireYAML(""))
+ if job != nil {
+ f.run(t, job)
+ }
+ if !f.versionExists(t, "data/a", older) {
+ t.Fatal("a job submitted without data access deleted a version")
+ }
+ if rec.Code != http.StatusForbidden {
+ t.Fatalf("submission status %d, want 403: %s", rec.Code, rec.Body.String())
+ }
+}
+
+// A job acts only while the credential that submitted it is the same valid
+// credential.
+func TestBatchJobSubmitterChangedBeforeStart(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ change func(t *testing.T, f *batchAuthzFixture, cred auth.Credentials)
+ }{
+ {"unchanged", nil},
+ {"deleted", func(t *testing.T, f *batchAuthzFixture, cred auth.Credentials) {
+ mustIAM(t, f.sys.DeleteUser(f.ctx, cred.AccessKey, false))
+ }},
+ {"disabled", func(t *testing.T, f *batchAuthzFixture, cred auth.Credentials) {
+ _, err := f.sys.SetUserStatus(f.ctx, cred.AccessKey, madmin.AccountDisabled)
+ mustIAM(t, err)
+ }},
+ {"secret-changed", func(t *testing.T, f *batchAuthzFixture, cred auth.Credentials) {
+ mustIAM(t, f.sys.SetUserSecretKey(f.ctx, cred.AccessKey, "batch-authz-changed-secret"))
+ }},
+ {"access-key-reused", func(t *testing.T, f *batchAuthzFixture, cred auth.Credentials) {
+ mustIAM(t, f.sys.DeleteUser(f.ctx, cred.AccessKey, false))
+ time.Sleep(10 * time.Millisecond)
+ iamAdminUser(f.ctx, t, f.sys, cred.AccessKey, "batch-full")
+ }},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchAuthzPolicy(f.ctx, t, f.sys, "batch-full", batchAuthzFullAccess)
+ cred := iamAdminUser(f.ctx, t, f.sys, "batch-user", "batch-full")
+ older := f.putVersions(t, "data/a")
+ rec, job := f.submit(t, cred, f.expireYAML(""))
+ if job == nil {
+ t.Fatalf("submission status %d: %s", rec.Code, rec.Body.String())
+ }
+ if tc.change != nil {
+ tc.change(t, f, cred)
+ }
+ f.run(t, job)
+ exists := f.versionExists(t, "data/a", older)
+ if tc.change == nil && exists {
+ t.Fatal("control: an authorized job did not expire the older version")
+ }
+ if tc.change != nil && !exists {
+ t.Fatal("a job ran as a credential that changed after submission")
+ }
+ })
+ }
+}
+
+// An expired temporary credential does not keep its job running.
+func TestBatchJobSubmitterExpired(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchAuthzPolicy(f.ctx, t, f.sys, "batch-full", batchAuthzFullAccess)
+ parent := iamAdminUser(f.ctx, t, f.sys, "batch-parent", "batch-full")
+ cred, err := auth.GetNewCredentialsWithMetadata(map[string]any{parentClaim: parent.AccessKey, "exp": UTCNow().Add(2 * time.Second).Unix()}, globalActiveCred.SecretKey)
+ mustIAM(t, err)
+ cred.ParentUser = parent.AccessKey
+ _, err = f.sys.SetTempUser(f.ctx, cred.AccessKey, cred, "")
+ mustIAM(t, err)
+ older := f.putVersions(t, "data/a")
+ rec, job := f.submit(t, cred, f.expireYAML(""))
+ if job == nil {
+ t.Fatalf("submission status %d: %s", rec.Code, rec.Body.String())
+ }
+ time.Sleep(time.Until(cred.Expiration) + time.Second)
+ f.run(t, job)
+ if !f.versionExists(t, "data/a", older) {
+ t.Fatal("a job ran as an expired credential")
+ }
+}
+
+// A session policy limits the job as it limits requests, also when an
+// authorization plugin decides the rest.
+func TestBatchJobSessionPolicyDeny(t *testing.T) {
+ session := `{"Version":"2012-10-17","Statement":[
+{"Effect":"Allow","Action":["admin:StartBatchJob"]},
+{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::batch-authz","arn:aws:s3:::batch-authz/*"]},
+{"Effect":"Deny","Action":["s3:DeleteObjectVersion"],"Resource":["arn:aws:s3:::batch-authz/*"]}]}`
+ for _, tc := range []struct {
+ name string
+ plugin bool
+ }{{"local", false}, {"plugin-allows", true}} {
+ t.Run(tc.name, func(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchAuthzPolicy(f.ctx, t, f.sys, "batch-full", batchAuthzFullAccess)
+ parent := iamAdminUser(f.ctx, t, f.sys, "batch-parent", "batch-full")
+ cred := iamAdminServiceAccount(f.ctx, t, f.sys, parent.AccessKey, "batch-restricted", session)
+ if tc.plugin {
+ iamAdminPlugin(t, true)
+ }
+ older := f.putVersions(t, "data/a")
+ rec, job := f.submit(t, cred, f.expireYAML(""))
+ if job != nil {
+ f.run(t, job)
+ } else {
+ t.Logf("submission status %d: %s", rec.Code, rec.Body.String())
+ }
+ if !f.versionExists(t, "data/a", older) {
+ t.Fatal("a job deleted a version its session policy denies")
+ }
+ })
+ }
+}
+
+// A temporary credential's session policy limits its job too.
+func TestBatchJobTemporarySessionPolicyDeny(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchAuthzPolicy(f.ctx, t, f.sys, "batch-full", batchAuthzFullAccess)
+ parent := iamAdminUser(f.ctx, t, f.sys, "batch-parent", "batch-full")
+ session := `{"Version":"2012-10-17","Statement":[
+{"Effect":"Allow","Action":["admin:StartBatchJob"]},
+{"Effect":"Allow","Action":["s3:ListBucket","s3:ListBucketVersions","s3:GetObject","s3:GetObjectVersion","s3:DeleteObject"],"Resource":["arn:aws:s3:::batch-authz","arn:aws:s3:::batch-authz/*"]}]}`
+ cred, err := auth.GetNewCredentialsWithMetadata(map[string]any{
+ parentClaim: parent.AccessKey, "exp": UTCNow().Add(time.Hour).Unix(),
+ policy.SessionPolicyName: base64.StdEncoding.EncodeToString([]byte(session)),
+ }, globalActiveCred.SecretKey)
+ mustIAM(t, err)
+ cred.ParentUser = parent.AccessKey
+ _, err = f.sys.SetTempUser(f.ctx, cred.AccessKey, cred, "")
+ mustIAM(t, err)
+ older := f.putVersions(t, "data/a")
+ rec, job := f.submit(t, cred, f.expireYAML(""))
+ if job != nil {
+ f.run(t, job)
+ } else {
+ t.Logf("submission status %d: %s", rec.Code, rec.Body.String())
+ }
+ if !f.versionExists(t, "data/a", older) {
+ t.Fatal("a job deleted a version its session policy does not allow")
+ }
+}
+
+// A Deny that depends on request context the job does not have refuses the
+// action instead of being skipped.
+func TestBatchJobAbsentContextDeny(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchAuthzPolicy(f.ctx, t, f.sys, "batch-context", batchAuthzFullAccess+`,
+{"Effect":"Deny","Action":["s3:DeleteObjectVersion"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"Bool":{"aws:SecureTransport":"false"}}}`)
+ cred := iamAdminUser(f.ctx, t, f.sys, "batch-user", "batch-context")
+ older := f.putVersions(t, "data/a")
+ rec, job := f.submit(t, cred, f.expireYAML(""))
+ if job != nil {
+ f.run(t, job)
+ } else {
+ t.Logf("submission status %d: %s", rec.Code, rec.Body.String())
+ }
+ if !f.versionExists(t, "data/a", older) {
+ t.Fatal("a job skipped a Deny that depends on request context")
+ }
+}
+
+// Each object is authorized: a Deny under an allowed prefix keeps its objects.
+func TestBatchJobPerObjectDeny(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ batchAuthzPolicy(f.ctx, t, f.sys, "batch-keep", batchAuthzFullAccess+`,
+{"Effect":"Deny","Action":["s3:DeleteObjectVersion"],"Resource":["arn:aws:s3:::batch-authz/data/keep*"]}`)
+ cred := iamAdminUser(f.ctx, t, f.sys, "batch-user", "batch-keep")
+ kept := f.putVersions(t, "data/keep")
+ rec, job := f.submit(t, cred, f.expireYAML("data/"))
+ if job == nil {
+ t.Fatalf("submission status %d: %s", rec.Code, rec.Body.String())
+ }
+ f.run(t, job)
+ if !f.versionExists(t, "data/keep", kept) {
+ t.Fatal("a job deleted a version that a per-object Deny protects")
+ }
+ if ri := f.jobInfo(t, job); !ri.Failed || ri.Complete {
+ t.Fatalf("refused job: complete=%v failed=%v", ri.Complete, ri.Failed)
+ }
+}
+
+// A job queued without its submitter, as an earlier release stored it, fails
+// and stays failed across another start.
+func TestBatchJobWithoutSubmitter(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ older := f.putVersions(t, "data/a")
+ job := &BatchJobRequest{}
+ mustIAM(t, yaml.Unmarshal([]byte(f.expireYAML("")), job))
+ job.ID = "expire-legacy" + getKeySeparator() + "0"
+ job.User = globalActiveCred.AccessKey
+ job.Started = time.Now()
+ mustIAM(t, job.save(f.ctx, f.obj))
+ for range 2 {
+ loaded := &BatchJobRequest{}
+ mustIAM(t, loaded.load(f.ctx, f.obj, getJobPath(*job)))
+ f.run(t, loaded)
+ if !f.versionExists(t, "data/a", older) {
+ t.Fatal("a job without a recorded submitter deleted a version")
+ }
+ if ri := f.jobInfo(t, job); !ri.Failed || ri.Complete {
+ t.Fatalf("job without submitter: complete=%v failed=%v", ri.Complete, ri.Failed)
+ }
+ }
+}
+
+// Root and an authorized user can still run jobs.
+func TestBatchJobAuthorizedControls(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")
+ }
+ older := f.putVersions(t, "data/a")
+ rec, job := f.submit(t, cred, f.expireYAML(""))
+ if job == nil {
+ t.Fatalf("submission status %d: %s", rec.Code, rec.Body.String())
+ }
+ f.run(t, job)
+ if f.versionExists(t, "data/a", older) {
+ t.Fatal("an authorized job did not expire the older version")
+ }
+ if ri := f.jobInfo(t, job); !ri.Complete || ri.Failed {
+ t.Fatalf("authorized job: complete=%v failed=%v", ri.Complete, ri.Failed)
+ }
+ })
+ }
+}
+
+// The describe output has no secrets of the job.
+func TestBatchJobDescribeRedactsSecrets(t *testing.T) {
+ f := newBatchAuthzFixture(t)
+ secrets := []string{"describe-source-secret", "describe-source-token", "describe-target-secret", "describe-target-token", "describe-notify-token"}
+ for _, job := range []BatchJobRequest{
+ {ID: "replicate-describe" + getKeySeparator() + "0", Replicate: &BatchJobReplicateV1{
+ APIVersion: batchReplJobAPIVersion,
+ Source: BatchJobReplicateSource{Bucket: "remote", Endpoint: "https://remote.invalid", Creds: BatchJobReplicateCredentials{
+ AccessKey: "describe-source-access", SecretKey: secrets[0], SessionToken: secrets[1],
+ }},
+ Target: BatchJobReplicateTarget{Bucket: f.bucket, Creds: BatchJobReplicateCredentials{
+ AccessKey: "describe-target-access", SecretKey: secrets[2], SessionToken: secrets[3],
+ }},
+ Flags: BatchJobReplicateFlags{Notify: BatchJobNotification{Endpoint: "https://notify.invalid", Token: secrets[4]}},
+ }},
+ {ID: "expire-describe" + getKeySeparator() + "0", Expire: &BatchJobExpire{
+ APIVersion: batchExpireAPIVersion, Bucket: f.bucket,
+ NotificationCfg: BatchJobNotification{Endpoint: "https://notify.invalid", Token: secrets[4]},
+ }},
+ {ID: "keyrotate-describe" + getKeySeparator() + "0", KeyRotate: &BatchJobKeyRotateV1{
+ APIVersion: batchKeyRotateAPIVersion, Bucket: f.bucket,
+ Flags: BatchJobKeyRotateFlags{Notify: BatchJobNotification{Endpoint: "https://notify.invalid", Token: secrets[4]}},
+ }},
+ } {
+ data, err := job.MarshalMsg(nil)
+ mustIAM(t, err)
+ mustIAM(t, saveConfig(f.ctx, f.obj, getJobPath(job), data))
+ cred := globalActiveCred
+ req, err := newTestSignedRequestV4(http.MethodGet, "http://localhost/minio/admin/v3/describe-job?jobId="+job.ID, 0, nil, cred.AccessKey, cred.SecretKey, nil)
+ mustIAM(t, err)
+ mustIAM(t, req.ParseForm())
+ rec := httptest.NewRecorder()
+ req = req.WithContext(newContext(req, rec, "DescribeBatchJob"))
+ adminAPIHandlers{}.DescribeBatchJob(rec, req)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s: describe status %d: %s", job.ID, rec.Code, rec.Body.String())
+ }
+ for _, s := range secrets {
+ if strings.Contains(rec.Body.String(), s) {
+ t.Errorf("%s: describe output contains %s", job.ID, s)
+ }
+ }
+ if job.Replicate != nil && !strings.Contains(rec.Body.String(), "describe-source-access") {
+ t.Errorf("%s: describe output lost the access key", job.ID)
+ }
+ // The stored job keeps what a run needs.
+ stored := &BatchJobRequest{}
+ mustIAM(t, stored.load(f.ctx, f.obj, getJobPath(job)))
+ if stored.Replicate != nil && stored.Replicate.Source.Creds.SecretKey != secrets[0] {
+ t.Errorf("%s: describe changed the stored job", job.ID)
+ }
+ }
+}