diff --git a/cmd/config-migrate.go b/cmd/config-migrate.go index aa74a13e1..c90070c17 100644 --- a/cmd/config-migrate.go +++ b/cmd/config-migrate.go @@ -91,7 +91,7 @@ func readConfigWithoutMigrate(ctx context.Context, objAPI ObjectLayer) (config.C return nil, err } - data, err = decryptData(data, configFile) + data, err = decryptData(ctx, data, configFile) if err != nil { return nil, err } diff --git a/cmd/config.go b/cmd/config.go index 7311d628f..5933d41c7 100644 --- a/cmd/config.go +++ b/cmd/config.go @@ -68,7 +68,7 @@ func listServerConfigHistory(ctx context.Context, objAPI ObjectLayer, withData b continue } - data, err = decryptData(data, obj.Name) + data, err = decryptData(ctx, data, obj.Name) if err != nil { // ignore history file that cannot be loaded. continue @@ -110,7 +110,7 @@ func readServerConfigHistory(ctx context.Context, objAPI ObjectLayer, uuidKV str return nil, err } - return decryptData(data, historyFile) + return decryptData(ctx, data, historyFile) } func saveServerConfigHistory(ctx context.Context, objAPI ObjectLayer, kv []byte) error { @@ -162,7 +162,7 @@ func readServerConfig(ctx context.Context, objAPI ObjectLayer, data []byte) (con return nil, err } - data, err = decryptData(data, configFile) + data, err = decryptData(ctx, data, configFile) if err != nil { lookupConfigs(srvCfg, objAPI) return nil, err diff --git a/cmd/iam-etcd-store.go b/cmd/iam-etcd-store.go index 0307fcac4..fe8a71904 100644 --- a/cmd/iam-etcd-store.go +++ b/cmd/iam-etcd-store.go @@ -123,8 +123,8 @@ func (ies *IAMEtcdStore) saveIAMConfig(ctx context.Context, item any, itemPath s return nil } -func (ies *IAMEtcdStore) decodeIAMConfig(item any, data []byte, path string) error { - data, err := decryptData(data, path) +func (ies *IAMEtcdStore) decodeIAMConfig(ctx context.Context, item any, data []byte, path string) error { + data, err := decryptData(ctx, data, path) if err != nil { return err } @@ -145,7 +145,7 @@ func (ies *IAMEtcdStore) loadIAMConfigBytes(ctx context.Context, path string) ([ if err != nil { return nil, err } - data, err = decryptData(data, path) + data, err = decryptData(ctx, data, path) if err == nil { ies.index.observe(path, data) } @@ -187,7 +187,7 @@ func (ies *IAMEtcdStore) loadPolicyDoc(ctx context.Context, policy string, m map } func (ies *IAMEtcdStore) getPolicyDocKV(ctx context.Context, kvs *mvccpb.KeyValue, m map[string]PolicyDoc) error { - data, err := decryptData(kvs.Value, string(kvs.Key)) + data, err := decryptData(ctx, kvs.Value, string(kvs.Key)) if err != nil { if err == errConfigNotFound { return errNoSuchPolicy @@ -211,27 +211,31 @@ func (ies *IAMEtcdStore) getPolicyDocKV(ctx context.Context, kvs *mvccpb.KeyValu } func (ies *IAMEtcdStore) loadPolicyDocs(ctx context.Context, m map[string]PolicyDoc) error { - ctx, cancel := context.WithTimeout(ctx, defaultContextTimeout) + // Bound the etcd read without shortening the caller's decryption budget. + cctx, cancel := context.WithTimeout(ctx, defaultContextTimeout) defer cancel() // Retrieve all keys and values to avoid too many calls to etcd in case of // a large number of policies - r, err := ies.client.Get(ctx, iamConfigPoliciesPrefix, etcd.WithPrefix()) + r, err := ies.client.Get(cctx, iamConfigPoliciesPrefix, etcd.WithPrefix()) if err != nil { return err } // Parse all values to construct the policies data model. for _, kvs := range r.Kvs { + if err := ctx.Err(); err != nil { + return err + } if err = ies.getPolicyDocKV(ctx, kvs, m); err != nil && !errors.Is(err, errNoSuchPolicy) { return err } } - return nil + return ctx.Err() } func (ies *IAMEtcdStore) getUserKV(ctx context.Context, userkv *mvccpb.KeyValue, userType IAMUserType, m map[string]UserIdentity, basePrefix string) error { var u UserIdentity - err := ies.decodeIAMConfig(&u, userkv.Value, string(userkv.Key)) + err := ies.decodeIAMConfig(ctx, &u, userkv.Value, string(userkv.Key)) if err != nil { if err == errConfigNotFound { return errNoSuchUser @@ -392,9 +396,9 @@ func (ies *IAMEtcdStore) loadMappedPolicy(ctx context.Context, name string, user return nil } -func (ies *IAMEtcdStore) getMappedPolicy(kv *mvccpb.KeyValue, m *xsync.MapOf[string, MappedPolicy], basePrefix string) error { +func (ies *IAMEtcdStore) getMappedPolicy(ctx context.Context, kv *mvccpb.KeyValue, m *xsync.MapOf[string, MappedPolicy], basePrefix string) error { var p MappedPolicy - err := ies.decodeIAMConfig(&p, kv.Value, string(kv.Key)) + err := ies.decodeIAMConfig(ctx, &p, kv.Value, string(kv.Key)) if err != nil { if err == errConfigNotFound { return errNoSuchPolicy @@ -434,7 +438,7 @@ func (ies *IAMEtcdStore) loadMappedPolicies(ctx context.Context, userType IAMUse // Parse all policies mapping to create the proper data model for _, kv := range r.Kvs { - if err = ies.getMappedPolicy(kv, m, basePrefix); err != nil && !errors.Is(err, errNoSuchPolicy) { + if err = ies.getMappedPolicy(ctx, kv, m, basePrefix); err != nil && !errors.Is(err, errNoSuchPolicy) { return err } } diff --git a/cmd/iam-object-store.go b/cmd/iam-object-store.go index c5ede946e..be2969388 100644 --- a/cmd/iam-object-store.go +++ b/cmd/iam-object-store.go @@ -108,7 +108,10 @@ func (iamOS *IAMObjectStore) saveIAMConfig(ctx context.Context, item any, objPat return nil } -func decryptData(data []byte, objPath string) ([]byte, error) { +func decryptData(ctx context.Context, data []byte, objPath string) ([]byte, error) { + if err := ctx.Err(); err != nil { + return nil, err + } if utf8.Valid(data) { return data, nil } @@ -118,13 +121,17 @@ func decryptData(data []byte, objPath string) ([]byte, error) { return pdata, nil } if GlobalKMS != nil { - pdata, err = config.DecryptBytes(GlobalKMS, data, kms.Context{ + pdata, err = config.DecryptBytes(ctx, GlobalKMS, data, kms.Context{ minioMetaBucket: path.Join(minioMetaBucket, objPath), }) if err == nil { return pdata, nil } - pdata, err = config.DecryptBytes(GlobalKMS, data, kms.Context{ + // Cancellation must not trigger the legacy associated-data fallback. + if err := ctx.Err(); err != nil { + return nil, err + } + pdata, err = config.DecryptBytes(ctx, GlobalKMS, data, kms.Context{ minioMetaBucket: objPath, }) if err == nil { @@ -164,7 +171,7 @@ func (iamOS *IAMObjectStore) loadIAMConfigBytesWithMetadata(ctx context.Context, } return nil, meta, err } - data, err = decryptData(data, objPath) + data, err = decryptData(ctx, data, objPath) if err != nil { return nil, meta, err } @@ -276,12 +283,15 @@ func (iamOS *IAMObjectStore) loadPolicyDocs(ctx context.Context, m map[string]Po return item.Err } + if err := ctx.Err(); err != nil { + return err + } policyName := path.Dir(item.Item) if err := iamOS.loadPolicyDoc(ctx, policyName, m); err != nil && !errors.Is(err, errNoSuchPolicy) { return err } } - return nil + return ctx.Err() } func (iamOS *IAMObjectStore) loadSecretKey(ctx context.Context, user string, userType IAMUserType) (string, error) { diff --git a/cmd/iam-policy-budget_test.go b/cmd/iam-policy-budget_test.go new file mode 100644 index 000000000..3952238b9 --- /dev/null +++ b/cmd/iam-policy-budget_test.go @@ -0,0 +1,181 @@ +// Copyright (c) 2026 Feng Ruohang +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "sync/atomic" + "testing" + "time" + + "github.com/minio/minio/internal/config" + "github.com/minio/minio/internal/kms" + "github.com/puzpuzpuz/xsync/v3" + etcd "go.etcd.io/etcd/client/v3" +) + +type iamPolicyBudgetGetObserver struct { + etcd.KV + elapsed time.Duration + budget time.Duration + block bool +} + +func (k *iamPolicyBudgetGetObserver) Get(ctx context.Context, key string, opts ...etcd.OpOption) (*etcd.GetResponse, error) { + start := time.Now() + if d, ok := ctx.Deadline(); ok { + k.budget = time.Until(d) + } + defer func() { k.elapsed = time.Since(start) }() + if k.block { + <-ctx.Done() + return nil, ctx.Err() + } + return k.KV.Get(ctx, key, opts...) +} + +func iamPolicyBudgetCall(ctx context.Context, store *IAMStoreSys, operation string) (int, error) { + switch operation { + case "policies": + m, err := store.ListPolicies(ctx, "") + return len(m), err + case "docs": + m, err := store.ListPolicyDocs(ctx, "") + return len(m), err + default: + m := xsync.NewMapOf[string, MappedPolicy]() + err := store.IAMStorageAPI.(*IAMEtcdStore).loadMappedPolicies(ctx, regUser, false, m) + return m.Size(), err + } +} + +func prepareIAMPolicyBudgetFixture(t *testing.T, operation string) (*IAMStoreSys, *iamPolicyBudgetGetObserver, *atomic.Int32) { + t.Helper() + ctx, sys, _ := prepareIAMRevisionFixture(t, "etcd") + backend := sys.store.IAMStorageAPI.(*IAMEtcdStore) + builtin, err := kms.NewBuiltin("cancellation-key", bytes.Repeat([]byte{43}, 32)) + mustIAM(t, err) + var doc any + path := getPolicyDocPath("budget-slow") + if operation == "mappings" { + path = getMappedPolicyPath("budget-slow", regUser, false) + doc = newMappedPolicy("readonly") + } else { + p, err := sys.store.GetPolicy("readonly") + mustIAM(t, err) + doc = newPolicyDoc(p) + } + plain, err := json.Marshal(doc) + mustIAM(t, err) + encrypted, err := config.EncryptBytes(builtin, plain, kms.Context{minioMetaBucket: minioMetaBucket + "/" + path}) + mustIAM(t, err) + _, err = backend.client.Put(ctx, path, string(encrypted)) + mustIAM(t, err) + calls := &atomic.Int32{} + priorKMS := GlobalKMS + GlobalKMS = connectCancellationKES(t, cancellationDecryptHandler(t, builtin, func(r *http.Request) { + calls.Add(1) + select { + case <-time.After(450 * time.Millisecond): + case <-r.Context().Done(): + } + })) + t.Cleanup(func() { GlobalKMS = priorKMS }) + priorTimeout := defaultContextTimeout + defaultContextTimeout = 200 * time.Millisecond + t.Cleanup(func() { defaultContextTimeout = priorTimeout }) + observer := &iamPolicyBudgetGetObserver{KV: backend.client.KV} + backend.client.KV = observer + t.Cleanup(func() { backend.client.KV = observer.KV }) + return sys.store, observer, calls +} + +// A fast etcd Get followed by slow KES decryption must retain the caller's +// budget, even when decryption lasts longer than the etcd read timeout. +func TestIAMPolicyEtcdDecodeBudget(t *testing.T) { + for _, operation := range []string{"policies", "docs", "mappings"} { + t.Run(operation, func(t *testing.T) { + store, observer, calls := prepareIAMPolicyBudgetFixture(t, operation) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + start := time.Now() + n, err := iamPolicyBudgetCall(ctx, store, operation) + t.Logf("operation=%s total_ms=%.3f get_ms=%.3f get_budget_ms=%.3f decrypts=%d count=%d error=%v caller_error=%v", operation, float64(time.Since(start).Microseconds())/1000, float64(observer.elapsed.Microseconds())/1000, float64(observer.budget.Microseconds())/1000, calls.Load(), n, err, ctx.Err()) + if err != nil { + t.Fatalf("valid caller budget was shortened after etcd Get completed: %v", err) + } + if observer.elapsed >= 200*time.Millisecond || observer.budget <= 0 || observer.budget > 210*time.Millisecond { + t.Fatalf("etcd read budget invalid: %+v", observer) + } + if n == 0 || calls.Load() != 1 || ctx.Err() != nil { + t.Fatal("successful result was incomplete") + } + }) + } +} + +// The correction must retain bounded etcd RPCs and caller cancellation. A +// failed list cannot publish its private map or advance the cache timestamp. +func TestIAMPolicyEtcdBudgetGuards(t *testing.T) { + for _, operation := range []string{"policies", "docs", "mappings"} { + for _, mode := range []string{"read_timeout", "caller_deadline", "caller_cancel"} { + t.Run(fmt.Sprintf("%s/%s", operation, mode), func(t *testing.T) { + store, observer, calls := prepareIAMPolicyBudgetFixture(t, operation) + cache := store.rlock() + before, count := cache.updatedAt, len(cache.iamPolicyDocsMap) + store.runlock() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + want := context.DeadlineExceeded + if mode == "read_timeout" { + observer.block = true + } + if mode == "caller_deadline" { + cancel() + ctx, cancel = context.WithTimeout(context.Background(), 80*time.Millisecond) + defer cancel() + } + var timer *time.Timer + if mode == "caller_cancel" { + timer = time.AfterFunc(80*time.Millisecond, cancel) + defer timer.Stop() + want = context.Canceled + } + start := time.Now() + _, err := iamPolicyBudgetCall(ctx, store, operation) + t.Logf("operation=%s mode=%s elapsed_ms=%.3f decrypts=%d error=%v", operation, mode, float64(time.Since(start).Microseconds())/1000, calls.Load(), err) + if !errors.Is(err, want) { + t.Fatalf("expected %v, got %v", want, err) + } + if mode == "read_timeout" && (calls.Load() != 0 || ctx.Err() != nil) { + t.Fatal("read timeout did not remain local to the etcd RPC") + } + cache = store.rlock() + unchanged := cache.updatedAt.Equal(before) && len(cache.iamPolicyDocsMap) == count + store.runlock() + if !unchanged { + t.Fatal("failed list published a cache replacement") + } + }) + } + } +} diff --git a/cmd/iam-policy-refresh_test.go b/cmd/iam-policy-refresh_test.go new file mode 100644 index 000000000..ddf72298d --- /dev/null +++ b/cmd/iam-policy-refresh_test.go @@ -0,0 +1,182 @@ +// Copyright (c) 2026 Feng Ruohang +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net/http" + "sync" + "testing" + "time" + + "github.com/minio/madmin-go/v3" + "github.com/minio/minio/internal/auth" + "github.com/pgsty/silo-pkg/v3/policy" +) + +type iamPolicyRefreshContext struct{} + +// Keep IAMStorageAPI as *IAMObjectStore so LoadIAMCache exercises the actual +// 32-worker object loader. Capture bytes and release the physical read lock +// before pausing; concurrent writes can commit against real object storage. +type iamPolicyRefreshReadGate struct { + ObjectLayer + policyPath, userPath string + listEntered, listRelease, refreshEntered, refreshRelease chan struct{} + listOnce, refreshOnce sync.Once +} + +func (g *iamPolicyRefreshReadGate) GetObjectNInfo(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, opts ObjectOptions) (*GetObjectReader, error) { + r, err := g.ObjectLayer.GetObjectNInfo(ctx, bucket, object, rs, h, opts) + if err != nil { + return nil, err + } + mode, _ := ctx.Value(iamPolicyRefreshContext{}).(string) + var release <-chan struct{} + if bucket == minioMetaBucket && mode == "list" && object == g.policyPath { + release = g.listRelease + } + if bucket == minioMetaBucket && mode == "refresh" && object == g.userPath { + release = g.refreshRelease + } + if release == nil { + return r, nil + } + data, err := io.ReadAll(r) + info := r.ObjInfo + r.Close() + if err != nil { + return nil, err + } + if mode == "list" { + g.listOnce.Do(func() { close(g.listEntered) }) + } else { + g.refreshOnce.Do(func() { close(g.refreshEntered) }) + } + select { + case <-release: + return NewGetObjectReaderFromReader(bytes.NewReader(data), info, opts) + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +func TestIAMPolicyReloadActualObjectRefresh(t *testing.T) { + for _, docs := range []bool{false, true} { + for _, canceled := range []bool{false, true} { + t.Run(fmt.Sprintf("docs=%t/cancel=%t", docs, canceled), func(t *testing.T) { + ctx, sys, obj := prepareIAMRevisionFixture(t) + store := sys.store + rw, err := store.GetPolicy("readwrite") + mustIAM(t, err) + ro, err := store.GetPolicy("readonly") + mustIAM(t, err) + for _, name := range []string{"review-edit", "review-delete"} { + _, err = store.SetPolicy(ctx, name, rw) + mustIAM(t, err) + } + const parent = "review-parent" + _, err = sys.CreateUser(ctx, parent, madmin.AddOrUpdateUserReq{SecretKey: "synthetic-review-password", Status: madmin.AccountEnabled}) + mustIAM(t, err) + secret, err := getTokenSigningKey() + mustIAM(t, err) + issue := func(typ string) auth.Credentials { + c, err := auth.GetNewCredentialsWithMetadata(map[string]any{"exp": UTCNow().Add(time.Hour).Unix(), parentClaim: parent, tokenRevokeTypeClaim: typ}, secret) + mustIAM(t, err) + c.ParentUser = parent + _, err = store.SetTempUser(ctx, c.AccessKey, c, "") + mustIAM(t, err) + return c + } + revoked, retained := issue("selected"), issue("other") + gate := &iamPolicyRefreshReadGate{ObjectLayer: obj, policyPath: getPolicyDocPath("review-edit"), userPath: getUserIdentityPath(parent, regUser), listEntered: make(chan struct{}), listRelease: make(chan struct{}), refreshEntered: make(chan struct{}), refreshRelease: make(chan struct{})} + store.IAMStorageAPI.(*IAMObjectStore).objAPI = gate + freeList := sync.OnceFunc(func() { close(gate.listRelease) }) + freeRefresh := sync.OnceFunc(func() { close(gate.refreshRelease) }) + var wg sync.WaitGroup + t.Cleanup(func() { freeList(); freeRefresh(); wg.Wait() }) + start := func(fn func() error) <-chan error { + done := make(chan error, 1) + wg.Go(func() { done <- fn() }) + return done + } + refreshDone := start(func() error { + return store.LoadIAMCache(context.WithValue(ctx, iamPolicyRefreshContext{}, "refresh"), false) + }) + awaitIAMPolicyReload(t, gate.refreshEntered) + listCtx, cancel := context.WithCancel(context.WithValue(ctx, iamPolicyRefreshContext{}, "list")) + defer cancel() + listDone := start(func() error { return iamPolicyReloadCall(listCtx, store, docs) }) + awaitIAMPolicyReload(t, gate.listEntered) + read := start(func() error { + if _, ok := store.GetUser(parent); !ok { + return errors.New("cached user lost") + } + _, err := store.GetPolicy("review-edit") + return err + }) + select { + case err := <-read: + mustIAM(t, err) + case <-time.After(time.Second): + t.Fatal("actual object reload blocked cached reads") + } + writes := []<-chan error{ + start(func() error { _, err := store.SetPolicy(ctx, "review-edit", ro); return err }), + start(func() error { return store.DeletePolicy(ctx, "review-delete", false) }), + start(func() error { _, err := store.RevokeTokens(ctx, parent, "selected"); return err }), + } + if canceled { + cancel() + } else { + freeList() + } + err = <-listDone + if canceled { + if !errors.Is(err, context.Canceled) { + t.Fatalf("cancel error: %v", err) + } + } else { + mustIAM(t, err) + } + for _, done := range writes { + mustIAM(t, <-done) + } + freeRefresh() + mustIAM(t, <-refreshDone) + p, err := store.GetPolicy("review-edit") + mustIAM(t, err) + if p.IsAllowed(policy.Args{Action: policy.PutObjectAction, BucketName: "bucket", ObjectName: "object"}) || !p.IsAllowed(policy.Args{Action: policy.GetObjectAction, BucketName: "bucket", ObjectName: "object"}) { + t.Fatal("actual object refresh restored stale policy") + } + if _, err := store.GetPolicy("review-delete"); !errors.Is(err, errNoSuchPolicy) { + t.Fatal("deleted policy restored") + } + if _, ok := store.GetUser(revoked.AccessKey); ok { + t.Fatal("revoked STS token restored") + } + if _, ok := store.GetUser(retained.AccessKey); !ok { + t.Fatal("unrelated token lost") + } + }) + } + } +} diff --git a/cmd/iam-policy-reload_test.go b/cmd/iam-policy-reload_test.go new file mode 100644 index 000000000..596b1d377 --- /dev/null +++ b/cmd/iam-policy-reload_test.go @@ -0,0 +1,373 @@ +// Copyright (c) 2026 Feng Ruohang +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/minio/madmin-go/v3" + "github.com/minio/minio/internal/auth" + "github.com/pgsty/silo-pkg/v3/policy" +) + +type iamPolicyReloadStore struct { + IAMStorageAPI + policies func(context.Context, map[string]PolicyDoc) error + users func(context.Context, IAMUserType, map[string]UserIdentity) error +} + +func (s *iamPolicyReloadStore) loadPolicyDocs(ctx context.Context, m map[string]PolicyDoc) error { + return s.policies(ctx, m) +} + +func (s *iamPolicyReloadStore) loadUsers(ctx context.Context, typ IAMUserType, m map[string]UserIdentity) error { + if s.users != nil { + return s.users(ctx, typ, m) + } + return s.IAMStorageAPI.loadUsers(ctx, typ, m) +} + +func iamPolicyReloadCall(ctx context.Context, store *IAMStoreSys, docs bool) error { + if docs { + _, err := store.ListPolicyDocs(ctx, "") + return err + } + _, err := store.ListPolicies(ctx, "") + return err +} + +func awaitIAMPolicyReload(t *testing.T, ch <-chan struct{}) { + t.Helper() + select { + case <-ch: + case <-time.After(5 * time.Second): + t.Fatal("IAM operation did not reach the controlled barrier") + } +} + +// Both list entry points must allow cached readers while preserving the writer +// ordering used by policy edits, peer notifications and credential revocation. +// A refresh that captured the old state must not publish after those writes. +func TestIAMPolicyReloadConcurrentChanges(t *testing.T) { + for _, backend := range []string{"object", "etcd"} { + for _, docs := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/docs=%t", backend, docs), func(t *testing.T) { + ctx, sys, obj := prepareIAMRevisionFixture(t, backend) + store := sys.store + original := store.IAMStorageAPI + rw, err := store.GetPolicy("readwrite") + mustIAM(t, err) + ro, err := store.GetPolicy("readonly") + mustIAM(t, err) + for _, name := range []string{"reload-edit", "reload-delete", "reload-peer"} { + _, err := store.SetPolicy(ctx, name, rw) + mustIAM(t, err) + } + const parent = "reload-parent" + _, err = sys.CreateUser(ctx, parent, madmin.AddOrUpdateUserReq{SecretKey: "synthetic-reload-password", Status: madmin.AccountEnabled}) + mustIAM(t, err) + secret, err := getTokenSigningKey() + mustIAM(t, err) + issue := func(typ string) auth.Credentials { + c, err := auth.GetNewCredentialsWithMetadata(map[string]any{"exp": UTCNow().Add(time.Hour).Unix(), parentClaim: parent, tokenRevokeTypeClaim: typ}, secret) + mustIAM(t, err) + c.ParentUser = parent + _, err = store.SetTempUser(ctx, c.AccessKey, c, "") + mustIAM(t, err) + return c + } + revoked, retained := issue("selected"), issue("other") + peer := &IAMStoreSys{IAMStorageAPI: newIAMObjectStore(obj, MinIOUsersSysType)} + if backend == "etcd" { + peer.IAMStorageAPI = newIAMEtcdStore(original.(*IAMEtcdStore).client, MinIOUsersSysType) + } + mustIAM(t, peer.LoadIAMCache(ctx, true)) + + type refreshKey struct{} + loading, releaseLoad := make(chan struct{}), make(chan struct{}) + refreshing, releaseRefresh := make(chan struct{}), make(chan struct{}) + unblockLoad := sync.OnceFunc(func() { close(releaseLoad) }) + unblockRefresh := sync.OnceFunc(func() { close(releaseRefresh) }) + var wg sync.WaitGroup + t.Cleanup(func() { unblockLoad(); unblockRefresh(); wg.Wait() }) + store.IAMStorageAPI = &iamPolicyReloadStore{ + IAMStorageAPI: original, + policies: func(ctx context.Context, m map[string]PolicyDoc) error { + if err := original.loadPolicyDocs(ctx, m); err != nil { + return err + } + if ctx.Value(refreshKey{}) == nil { + close(loading) + select { + case <-releaseLoad: + case <-ctx.Done(): + return ctx.Err() + } + } + return nil + }, + users: func(ctx context.Context, typ IAMUserType, m map[string]UserIdentity) error { + if err := original.loadUsers(ctx, typ, m); err != nil { + return err + } + if typ == stsUser && ctx.Value(refreshKey{}) != nil { + close(refreshing) + <-releaseRefresh + } + return nil + }, + } + start := func(fn func() error) <-chan error { + out := make(chan error, 1) + wg.Go(func() { out <- fn() }) + return out + } + refreshDone := start(func() error { return store.LoadIAMCache(context.WithValue(ctx, refreshKey{}, true), false) }) + awaitIAMPolicyReload(t, refreshing) + listDone := start(func() error { return iamPolicyReloadCall(ctx, store, docs) }) + awaitIAMPolicyReload(t, loading) + reader := start(func() error { + if _, ok := store.GetUser(parent); !ok { + return errors.New("committed user disappeared during policy loading") + } + _, err := store.GetPolicy("reload-edit") + return err + }) + select { + case err := <-reader: + mustIAM(t, err) + case <-time.After(time.Second): + t.Fatal("policy loading blocked cached authentication") + } + + // The peer commits independently; its notification waits alongside + // local changes while the older list snapshot remains in flight. + _, err = peer.SetPolicy(ctx, "reload-peer", ro) + mustIAM(t, err) + writersStarted := make(chan struct{}, 4) + startWrite := func(fn func() error) <-chan error { + return start(func() error { writersStarted <- struct{}{}; return fn() }) + } + writes := []<-chan error{ + startWrite(func() error { _, err := store.SetPolicy(ctx, "reload-edit", ro); return err }), + startWrite(func() error { return store.DeletePolicy(ctx, "reload-delete", false) }), + startWrite(func() error { _, err := store.RevokeTokens(ctx, parent, "selected"); return err }), + startWrite(func() error { return store.PolicyNotificationHandler(ctx, "reload-peer") }), + } + for range writes { + <-writersStarted + } + select { + case err := <-writes[0]: + t.Fatalf("policy write overtook the pending reload: %v", err) + case <-time.After(25 * time.Millisecond): + } + for _, done := range writes[1:] { + select { + case err := <-done: + t.Fatalf("IAM mutation overtook the pending reload: %v", err) + default: + } + } + unblockLoad() + mustIAM(t, <-listDone) + for _, done := range writes { + mustIAM(t, <-done) + } + unblockRefresh() + mustIAM(t, <-refreshDone) + for _, name := range []string{"reload-edit", "reload-peer"} { + p, err := store.GetPolicy(name) + mustIAM(t, err) + if p.IsAllowed(policy.Args{Action: policy.PutObjectAction, BucketName: "bucket", ObjectName: "object"}) { + t.Fatalf("stale list/refresh restored the old grant: %s", name) + } + if !p.IsAllowed(policy.Args{Action: policy.GetObjectAction, BucketName: "bucket", ObjectName: "object"}) { + t.Fatalf("reload lost the committed read grant: %s", name) + } + } + if _, err := store.GetPolicy("reload-delete"); !errors.Is(err, errNoSuchPolicy) { + t.Fatalf("deleted policy returned: %v", err) + } + if _, ok := store.GetUser(revoked.AccessKey); ok { + t.Fatal("stale list/refresh restored a revoked token") + } + for _, key := range []string{retained.AccessKey, parent} { + if _, ok := store.GetUser(key); !ok { + t.Fatal("reload removed an unrelated credential") + } + } + }) + } + } +} + +func TestIAMPolicyReloadFailurePreservesCache(t *testing.T) { + for _, docs := range []bool{false, true} { + for _, failure := range []string{"storage", "cancel_partial", "caller_deadline_partial"} { + t.Run(fmt.Sprintf("docs=%t/%s", docs, failure), func(t *testing.T) { + ctx, sys, _ := prepareIAMRevisionFixture(t) + var cancel context.CancelFunc + if failure == "caller_deadline_partial" { + ctx, cancel = context.WithTimeout(ctx, 100*time.Millisecond) + } else { + ctx, cancel = context.WithCancel(ctx) + } + defer cancel() + store := sys.store + cache := store.rlock() + updated := cache.updatedAt + store.runlock() + injected := errors.New("policy read failed") + entered := make(chan struct{}) + original := store.IAMStorageAPI + store.IAMStorageAPI = &iamPolicyReloadStore{IAMStorageAPI: original, policies: func(ctx context.Context, m map[string]PolicyDoc) error { + m["partial-result"] = PolicyDoc{} + close(entered) + if failure == "storage" { + return injected + } + <-ctx.Done() + // A canceled enumeration can close without yielding an error. + return nil + }} + done := make(chan error, 1) + go func() { done <- iamPolicyReloadCall(ctx, store, docs) }() + awaitIAMPolicyReload(t, entered) + want := injected + switch failure { + case "cancel_partial": + cancel() + want = context.Canceled + case "caller_deadline_partial": + want = context.DeadlineExceeded + } + select { + case err := <-done: + if !errors.Is(err, want) { + t.Fatalf("got %v, want %v", err, want) + } + case <-time.After(time.Second): + cancel() + <-done + t.Fatal("reload did not honor caller cancellation") + } + cache = store.rlock() + _, partial := cache.iamPolicyDocsMap["partial-result"] + unchanged := cache.updatedAt.Equal(updated) + store.runlock() + if partial || !unchanged { + t.Fatal("failed/canceled reload published a partial policy map") + } + store.IAMStorageAPI = original + // Cancellation must release both locks so the next operation works. + mustIAM(t, iamPolicyReloadCall(context.Background(), store, docs)) + if _, err := store.GetPolicy("readwrite"); err != nil { + t.Fatal("reload lost default policies") + } + }) + } + } +} + +type iamPolicyRelockObserver struct { + IAMStorageAPI + calls atomic.Int32 + reacquiring chan struct{} +} + +func (s *iamPolicyRelockObserver) lock() *iamCache { + if s.calls.Add(1) == 2 { + close(s.reacquiring) + } + return s.IAMStorageAPI.lock() +} + +func TestIAMPolicyReloadCallerBudgetPreserved(t *testing.T) { + for _, docs := range []bool{false, true} { + t.Run(fmt.Sprintf("docs=%t", docs), func(t *testing.T) { + _, sys, _ := prepareIAMRevisionFixture(t) + original := sys.store.IAMStorageAPI + prior := defaultContextTimeout + defaultContextTimeout = 50 * time.Millisecond + t.Cleanup(func() { defaultContextTimeout = prior }) + sys.store.IAMStorageAPI = &iamPolicyReloadStore{IAMStorageAPI: original, policies: func(ctx context.Context, m map[string]PolicyDoc) error { + select { + case <-time.After(120 * time.Millisecond): + return original.loadPolicyDocs(ctx, m) + case <-ctx.Done(): + return ctx.Err() + } + }} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + err := iamPolicyReloadCall(ctx, sys.store, docs) + if err != nil { + t.Fatalf("reload cut off a still-valid caller budget: err=%v callerErr=%v", err, ctx.Err()) + } + }) + } +} + +func TestIAMPolicyReloadCancelWhileReacquiringCache(t *testing.T) { + for _, docs := range []bool{false, true} { + t.Run(fmt.Sprintf("docs=%t", docs), func(t *testing.T) { + _, sys, _ := prepareIAMRevisionFixture(t) + store := sys.store + original := store.IAMStorageAPI + observer := &iamPolicyRelockObserver{IAMStorageAPI: original, reacquiring: make(chan struct{})} + entered, release := make(chan struct{}), make(chan struct{}) + store.IAMStorageAPI = &iamPolicyReloadStore{IAMStorageAPI: observer, policies: func(ctx context.Context, m map[string]PolicyDoc) error { + m["partial-reload"] = PolicyDoc{} + close(entered) + <-release + return nil + }} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- iamPolicyReloadCall(ctx, store, docs) }() + awaitIAMPolicyReload(t, entered) + cache := store.rlock() + before := cache.updatedAt + count := len(cache.iamPolicyDocsMap) + close(release) + awaitIAMPolicyReload(t, observer.reacquiring) + // This reader forces the reload to wait before its commit step. + // The request is canceled after I/O completed, before publishing. + cancel() + store.runlock() + err := <-done + if !errors.Is(err, context.Canceled) { + t.Fatalf("cancel before publish lost: %v", err) + } + cache = store.rlock() + _, partial := cache.iamPolicyDocsMap["partial-reload"] + unchanged := cache.updatedAt.Equal(before) && len(cache.iamPolicyDocsMap) == count + store.runlock() + if partial || !unchanged { + t.Fatal("canceled complete load still published") + } + }) + } +} diff --git a/cmd/iam-store.go b/cmd/iam-store.go index 05ec9e6d9..89f78284a 100644 --- a/cmd/iam-store.go +++ b/cmd/iam-store.go @@ -1581,10 +1581,22 @@ func (store *IAMStoreSys) ListPolicies(ctx context.Context, bucketName string) ( defer store.unlock() m := map[string]PolicyDoc{} - err := store.loadPolicyDocs(ctx, m) + // Keep cache writers serialized, but let authentication use the committed + // cache while the replacement is loaded into this private map. + err := func() error { + store.IAMStorageAPI.unlock() + defer store.IAMStorageAPI.lock() + if err := store.loadPolicyDocs(ctx, m); err != nil { + return err + } + return ctx.Err() + }() if err != nil { return nil, err } + if err := ctx.Err(); err != nil { + return nil, err + } // Sets default canned policies setDefaultCannedPolicies(m) @@ -1609,10 +1621,20 @@ func (store *IAMStoreSys) ListPolicyDocs(ctx context.Context, bucketName string) defer store.unlock() m := map[string]PolicyDoc{} - err := store.loadPolicyDocs(ctx, m) + err := func() error { + store.IAMStorageAPI.unlock() + defer store.IAMStorageAPI.lock() + if err := store.loadPolicyDocs(ctx, m); err != nil { + return err + } + return ctx.Err() + }() if err != nil { return nil, err } + if err := ctx.Err(); err != nil { + return nil, err + } // Sets default canned policies setDefaultCannedPolicies(m) diff --git a/cmd/kms-cancellation_test.go b/cmd/kms-cancellation_test.go new file mode 100644 index 000000000..08f5318e6 --- /dev/null +++ b/cmd/kms-cancellation_test.go @@ -0,0 +1,256 @@ +// Copyright (c) 2026 Feng Ruohang +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "bytes" + "context" + "encoding/json" + "encoding/pem" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/minio/kms-go/kes" + "github.com/minio/minio/internal/config" + "github.com/minio/minio/internal/crypto" + "github.com/minio/minio/internal/kms" +) + +// Use the configured KES client over TLS, rather than a mocked KMS interface: +// cancellation must survive every layer, including the HTTP request context. +func connectCancellationKES(t *testing.T, handler http.Handler) *kms.KMS { + t.Helper() + for _, key := range []string{kms.EnvKMSEndpoint, kms.EnvKMSSecretKey, kms.EnvKMSSecretKeyFile} { + t.Setenv(key, "") + mustIAM(t, os.Unsetenv(key)) + } + server := httptest.NewTLSServer(handler) + t.Cleanup(server.Close) + ca := filepath.Join(t.TempDir(), "ca.pem") + mustIAM(t, os.WriteFile(ca, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw}), 0o600)) + key, err := kes.GenerateAPIKey(nil) + mustIAM(t, err) + t.Setenv(kms.EnvKESEndpoint, server.URL) + t.Setenv(kms.EnvKESAPIKey, key.String()) + t.Setenv(kms.EnvKESDefaultKey, "cancellation-key") + t.Setenv(kms.EnvKESServerCA, ca) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + k, err := kms.Connect(ctx, nil) + mustIAM(t, err) + return k +} + +func cancellationDecryptHandler(t *testing.T, k *kms.KMS, before func(*http.Request)) http.Handler { + t.Helper() + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // KMS Connect periodically describes the default key to keep it cached. + // This maintenance request is independent of the decrypt under test. + if r.Method == http.MethodGet && r.URL.Path == "/v1/key/describe/cancellation-key" { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]string{"name": "cancellation-key"}) + return + } + if r.URL.Path != "/v1/key/decrypt/cancellation-key" { + t.Errorf("unexpected KES path: %s", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + var request struct { + Ciphertext []byte `json:"ciphertext"` + Context []byte `json:"context"` + } + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + t.Error(err) + return + } + // Drain the HTTP/1 body so net/http can observe a peer disconnect while + // the handler is blocked in the simulated KES operation. + _, _ = io.Copy(io.Discard, r.Body) + _ = r.Body.Close() + before(r) + if r.Context().Err() != nil { + return + } + var aad kms.Context + if err := json.Unmarshal(request.Context, &aad); err != nil { + t.Error(err) + return + } + plain, err := k.Decrypt(r.Context(), &kms.DecryptRequest{Name: "cancellation-key", Ciphertext: request.Ciphertext, AssociatedData: aad}) + w.Header().Set("Content-Type", "application/json") + if err != nil { + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]string{"message": kes.ErrDecrypt.Error()}) + return + } + _ = json.NewEncoder(w).Encode(map[string][]byte{"plaintext": plain}) + }) +} + +func TestKMSRequestCancellation(t *testing.T) { + for _, operation := range []string{"iam", "sse-list"} { + for _, expiry := range []string{"cancel", "deadline"} { + t.Run(operation+"/"+expiry, func(t *testing.T) { + builtin, err := kms.NewBuiltin("cancellation-key", bytes.Repeat([]byte{42}, 32)) + mustIAM(t, err) + entered, canceled, release := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var calls atomic.Int32 + var enteredOnce, canceledOnce sync.Once + unblock := sync.OnceFunc(func() { close(release) }) + client := connectCancellationKES(t, cancellationDecryptHandler(t, builtin, func(r *http.Request) { + calls.Add(1) + enteredOnce.Do(func() { close(entered) }) + select { + case <-r.Context().Done(): + canceledOnce.Do(func() { close(canceled) }) + case <-release: + } + })) + t.Cleanup(unblock) + prior := GlobalKMS + GlobalKMS = client + t.Cleanup(func() { GlobalKMS = prior }) + var run func(context.Context) error + if operation == "iam" { + const name = "config/iam/policies/cancel/policy.json" + plaintext := []byte(`{"Version":1}`) + ciphertext, err := config.EncryptBytes(builtin, plaintext, kms.Context{minioMetaBucket: minioMetaBucket + "/" + name}) + mustIAM(t, err) + run = func(ctx context.Context) error { + got, err := decryptData(ctx, ciphertext, name) + if err == nil && !bytes.Equal(got, plaintext) { + return errors.New("config plaintext changed") + } + return err + } + } else { + var metadata []map[string]string + var buckets, names []string + var expected []crypto.ObjectKey + for i := range 3 { + name := fmt.Sprintf("object-%d", i) + dek, err := builtin.GenerateKey(context.Background(), &kms.GenerateKeyRequest{AssociatedData: kms.Context{"bucket": "bucket/" + name}}) + mustIAM(t, err) + key := crypto.GenerateKey(dek.Plaintext, nil) + sealed := key.Seal(dek.Plaintext, crypto.GenerateIV(nil), crypto.S3.String(), "bucket", name) + metadata = append(metadata, crypto.S3.CreateMetadata(nil, dek.KeyID, dek.Ciphertext, sealed)) + buckets, names = append(buckets, "bucket"), append(names, name) + expected = append(expected, key) + } + run = func(ctx context.Context) error { + keys, err := crypto.S3.UnsealObjectKeys(ctx, client, metadata, buckets, names) + if err != nil { + return err + } + if len(keys) != len(expected) { + return errors.New("partial key list returned as success") + } + for i := range keys { + if keys[i] != expected[i] { + return errors.New("object key or ordering changed") + } + } + return nil + } + } + ctx, cancel := context.WithCancel(context.Background()) + if expiry == "deadline" { + cancel() + ctx, cancel = context.WithTimeout(context.Background(), 300*time.Millisecond) + } + defer cancel() + done := make(chan error, 1) + go func() { done <- run(ctx) }() + select { + case <-entered: + case err := <-done: + t.Fatalf("decrypt returned before reaching KES: %v (type=%v)", err, client.Type) + case <-time.After(5 * time.Second): + t.Fatal("decrypt did not reach KES") + } + want := context.DeadlineExceeded + if expiry == "cancel" { + cancel() + want = context.Canceled + } + select { + case err := <-done: + if !errors.Is(err, want) { + t.Fatalf("got %v, want %v", err, want) + } + // The current KES SDK may sleep twice for up to 999ms after a + // deadline error. Include that existing backoff plus the 300ms + // request deadline; cancellation must still stop new decrypts. + case <-time.After(3 * time.Second): + unblock() + <-done + t.Fatal("request exceeded the cancellation and KES retry budget") + } + awaitIAMPolicyReload(t, canceled) + if n := calls.Load(); n != 1 { + t.Fatalf("cancellation launched a fallback or more object decryptions: %d calls", n) + } + unblock() + mustIAM(t, run(context.Background())) + before := calls.Load() + cctx, ccancel := context.WithCancel(context.Background()) + ccancel() + if err := run(cctx); !errors.Is(err, context.Canceled) || calls.Load() != before { + t.Fatal("already-canceled request still issued decryption work") + } + }) + } + } +} + +func TestIAMDecryptAssociatedDataCompatibility(t *testing.T) { + builtin, err := kms.NewBuiltin("cancellation-key", bytes.Repeat([]byte{42}, 32)) + mustIAM(t, err) + var calls atomic.Int32 + client := connectCancellationKES(t, cancellationDecryptHandler(t, builtin, func(*http.Request) { calls.Add(1) })) + prior := GlobalKMS + GlobalKMS = client + t.Cleanup(func() { GlobalKMS = prior }) + const name = "config/iam/policies/compat/policy.json" + plaintext := []byte(`{"Version":1}`) + for _, legacy := range []bool{false, true} { + t.Run(fmt.Sprint(legacy), func(t *testing.T) { + aad, wantCalls := minioMetaBucket+"/"+name, int32(1) + if legacy { + aad, wantCalls = name, 2 + } + ciphertext, err := config.EncryptBytes(builtin, plaintext, kms.Context{minioMetaBucket: aad}) + mustIAM(t, err) + before := calls.Load() + got, err := decryptData(context.Background(), ciphertext, name) + mustIAM(t, err) + if !bytes.Equal(got, plaintext) || calls.Load()-before != wantCalls { + t.Fatal("current/legacy encrypted IAM records no longer decrypt correctly") + } + }) + } +} diff --git a/internal/config/crypto.go b/internal/config/crypto.go index 757b1db4e..5ce567484 100644 --- a/internal/config/crypto.go +++ b/internal/config/crypto.go @@ -48,8 +48,8 @@ func EncryptBytes(k *kms.KMS, plaintext []byte, context kms.Context) ([]byte, er // DecryptBytes decrypts the ciphertext using a key managed by the KMS. // The same context that have been used during encryption must be // provided. -func DecryptBytes(k *kms.KMS, ciphertext []byte, context kms.Context) ([]byte, error) { - plaintext, err := Decrypt(k, bytes.NewReader(ciphertext), context) +func DecryptBytes(ctx context.Context, k *kms.KMS, ciphertext []byte, associatedData kms.Context) ([]byte, error) { + plaintext, err := Decrypt(ctx, k, bytes.NewReader(ciphertext), associatedData) if err != nil { return nil, err } @@ -115,7 +115,10 @@ func Encrypt(k *kms.KMS, plaintext io.Reader, ctx kms.Context) (io.Reader, error // Decrypt decrypts the ciphertext using a key managed by the KMS. // The same context that have been used during encryption must be // provided. -func Decrypt(k *kms.KMS, ciphertext io.Reader, associatedData kms.Context) (io.Reader, error) { +func Decrypt(ctx context.Context, k *kms.KMS, ciphertext io.Reader, associatedData kms.Context) (io.Reader, error) { + if err := ctx.Err(); err != nil { + return nil, err + } const ( MaxMetadataSize = 1 << 20 // max. size of the metadata Version = 1 @@ -145,7 +148,7 @@ func Decrypt(k *kms.KMS, ciphertext io.Reader, associatedData kms.Context) (io.R return nil, err } - key, err := k.Decrypt(context.TODO(), &kms.DecryptRequest{ + key, err := k.Decrypt(ctx, &kms.DecryptRequest{ Name: metadata.KeyID, Ciphertext: metadata.KMSKey, AssociatedData: associatedData, diff --git a/internal/config/crypto_test.go b/internal/config/crypto_test.go index 224d230bd..98b63ed33 100644 --- a/internal/config/crypto_test.go +++ b/internal/config/crypto_test.go @@ -19,6 +19,7 @@ package config import ( "bytes" + "context" "encoding/hex" "io" "testing" @@ -68,7 +69,7 @@ func TestEncryptDecrypt(t *testing.T) { t.Fatalf("Test %d: failed to encrypt stream: %v", i, err) } - plaintext, err := Decrypt(KMS, bytes.NewReader(data), test.Context) + plaintext, err := Decrypt(context.Background(), KMS, bytes.NewReader(data), test.Context) if err != nil { t.Fatalf("Test %d: failed to decrypt stream: %v", i, err) } diff --git a/internal/crypto/sse-s3.go b/internal/crypto/sse-s3.go index ce34d5a4f..17abc76c5 100644 --- a/internal/crypto/sse-s3.go +++ b/internal/crypto/sse-s3.go @@ -72,6 +72,15 @@ func (sses3) IsEncrypted(metadata map[string]string) bool { // from the metadata using KMS and returns the decrypted object // key. func (s3 sses3) UnsealObjectKey(k *kms.KMS, metadata map[string]string, bucket, object string) (key ObjectKey, err error) { + return s3.unsealObjectKey(context.Background(), k, metadata, bucket, object) +} + +// unsealObjectKey carries the listing request's cancellation through the KMS. +// Keep the existing single-object API for callers without a request context. +func (s3 sses3) unsealObjectKey(ctx context.Context, k *kms.KMS, metadata map[string]string, bucket, object string) (key ObjectKey, err error) { + if err := ctx.Err(); err != nil { + return key, err + } if k == nil { return key, Errorf("KMS not configured") } @@ -79,7 +88,7 @@ func (s3 sses3) UnsealObjectKey(k *kms.KMS, metadata map[string]string, bucket, if err != nil { return key, err } - unsealKey, err := k.Decrypt(context.TODO(), &kms.DecryptRequest{ + unsealKey, err := k.Decrypt(ctx, &kms.DecryptRequest{ Name: keyID, Ciphertext: kmsKey, AssociatedData: kms.Context{bucket: path.Join(bucket, object)}, @@ -106,12 +115,15 @@ func (s3 sses3) UnsealObjectKeys(ctx context.Context, k *kms.KMS, metadata []map } keys := make([]ObjectKey, 0, len(metadata)) for i := range metadata { - key, err := s3.UnsealObjectKey(k, metadata[i], buckets[i], objects[i]) + key, err := s3.unsealObjectKey(ctx, k, metadata[i], buckets[i], objects[i]) if err != nil { return nil, err } keys = append(keys, key) } + if err := ctx.Err(); err != nil { + return nil, err + } return keys, nil } diff --git a/internal/kms/kes.go b/internal/kms/kes.go index 8bd387c27..27ca3429f 100644 --- a/internal/kms/kes.go +++ b/internal/kms/kes.go @@ -235,8 +235,11 @@ func (c *kesConn) Decrypt(ctx context.Context, req *DecryptRequest) ([]byte, err return nil, err } - plaintext, err := c.client.Decrypt(context.Background(), req.Name, req.Ciphertext, aad) + plaintext, err := c.client.Decrypt(ctx, req.Name, req.Ciphertext, aad) if err != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } if errors.Is(err, kes.ErrKeyNotFound) { return nil, ErrKeyNotFound }