test(batch): cover submitter authorization and locked versions

Submission and start with changed, expired or restricted credentials,
session policies with and without an authorization plugin, request
context in policy conditions, per-object denies, jobs without a
recorded submitter and describe output; expire, pull, push and
rotation over versions under retention or legal hold, on one and
sixteen drives; cursor and counter persistence; and the cases found in
review: destination versions and tags of pulls, encryption values,
S3 Tables statements, copies in several pools, claims, prefixes,
listing errors and retry accounting.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Feng Ruohang <rh@vonng.com>

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(`<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Suspended</Status></VersioningConfiguration>`))
+	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, `<Error><Code>AccessDenied</Code><Message>test refusal</Message></Error>`
+		if r.URL.Query().Has("location") {
+			code, body = http.StatusOK, `<LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/">us-east-1</LocationConstraint>`
+		}
+		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")
+	}
+}

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, `<Error><Code>AccessDenied</Code><Message>test refusal</Message></Error>`
+		h := make(http.Header)
+		switch {
+		case r.URL.Query().Has("location"):
+			code, body = http.StatusOK, `<LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/">us-east-1</LocationConstraint>`
+		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)
+	}
+}

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")
+	}
+}

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(`<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Suspended</Status></VersioningConfiguration>`))
+	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)
+			}
+		})
+	}
+}

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)
+			}
+		})
+	}
+}

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(`<ObjectLockConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Mode>GOVERNANCE</Mode><Days>1</Days></DefaultRetention></Rule></ObjectLockConfiguration>`)
+			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)
+	}
+}

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)
+		}
+	}
+}