From ae54718d8465b9e455c74527716d8e1390d2cfcf Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Thu, 5 Nov 2026 04:14:06 +0800 Subject: [PATCH] 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 Signed-off-by: Feng Ruohang --- cmd/batch-authz-review2_test.go | 356 ++++++++++++++++++++++ cmd/batch-authz-review3_test.go | 140 +++++++++ cmd/batch-authz-review4_test.go | 157 ++++++++++ cmd/batch-authz-review5_test.go | 142 +++++++++ cmd/batch-authz-review_test.go | 287 ++++++++++++++++++ cmd/batch-object-lock_test.go | 479 ++++++++++++++++++++++++++++++ cmd/batch-submitter-authz_test.go | 456 ++++++++++++++++++++++++++++ 7 files changed, 2017 insertions(+) create mode 100644 cmd/batch-authz-review2_test.go create mode 100644 cmd/batch-authz-review3_test.go create mode 100644 cmd/batch-authz-review4_test.go create mode 100644 cmd/batch-authz-review5_test.go create mode 100644 cmd/batch-authz-review_test.go create mode 100644 cmd/batch-object-lock_test.go create mode 100644 cmd/batch-submitter-authz_test.go 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) + } + } +}