From 9f9398a1a3162c8294c92eb3573a03f65aefc9ae Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 7 Nov 2026 16:01:42 +0800 Subject: [PATCH] fix(batch): keep cached authorization available during batch policy loads Batch authorization evaluates the submitter's mapped policies from one IAM cache revision. It took the cache write lock for the whole evaluation and performed the lazy loads it needed under that lock: the mapping of a user whose policies come from a group is never cached, so every authorization of such a submitter (and of an LDAP or STS parent user) read storage twice with the write lock held, and an evicted policy document was read under it too. While a read lasted, every reader of the cache waited: request authentication (GetUser, CheckKey) and PolicyDBGet on the whole node. An expire job authorizes each version several times, so a large job with a group-mapped submitter serialized node authentication behind IAM storage latency, and a mapping load had no deadline at all. batchPolicySnapshot now holds the store's writer mutex and the cache read lock, and releases the read lock around each storage read, as MergePolicies releases the cache lock around a document load. The writer mutex still excludes cache writers, so the mapping and the documents of one evaluation still come from one revision; the read lock is enough because the batch path writes nothing to the cache except lazily loaded mappings, which the request path stores under the read lock as well, and taking the write lock back after each read would have made readers queue behind it. Each read is bounded by the job's context and the same 5s MergePolicies grants a document load. Concurrent reads of the group fallback release the lock once and take it back after the last. policyDBGet and policyDBGetGroups take the loads they need through the iamMappingLoader interface, which *IAMStoreSys satisfies unchanged. Unreadable policies still end the attempt as unavailable, and the error classification is unchanged. TestBatchAuthorizeSlowLoadKeepsCachedAuthorizationAvailable holds the lazy load of a group-mapped user's mapping, and of an evicted document, and requires that GetUser and IsAllowed of a cached user return within a second and that a load ends with the job's context. Before this change both cases timed out on the blocked cache. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Feng Ruohang --- cmd/batch-authz-snapshot_test.go | 159 +++++++++++++++++++++++++++++++ cmd/batch-authz.go | 100 ++++++++++++++++--- cmd/iam-store.go | 15 ++- 3 files changed, 259 insertions(+), 15 deletions(-) create mode 100644 cmd/batch-authz-snapshot_test.go diff --git a/cmd/batch-authz-snapshot_test.go b/cmd/batch-authz-snapshot_test.go new file mode 100644 index 000000000..15b3e3d9d --- /dev/null +++ b/cmd/batch-authz-snapshot_test.go @@ -0,0 +1,159 @@ +// Copyright (c) 2026 PGSTY +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/pgsty/silo-pkg/v3/policy" + "github.com/puzpuzpuz/xsync/v3" +) + +// batchBlockedIAMStore holds the lazy load of one user's policy mapping, or +// of one policy document, until released or until the load's context ends. +type batchBlockedIAMStore struct { + IAMStorageAPI + mapping string // access key whose mapping load blocks + doc string // policy whose document load blocks + started chan struct{} + release chan struct{} + once sync.Once +} + +func (s *batchBlockedIAMStore) block(ctx context.Context) error { + s.once.Do(func() { close(s.started) }) + select { + case <-s.release: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (s *batchBlockedIAMStore) loadMappedPolicy(ctx context.Context, name string, userType IAMUserType, isGroup bool, m *xsync.MapOf[string, MappedPolicy]) error { + if name == s.mapping && !isGroup { + if err := s.block(ctx); err != nil { + return err + } + } + return s.IAMStorageAPI.loadMappedPolicy(ctx, name, userType, isGroup, m) +} + +func (s *batchBlockedIAMStore) loadPolicyDoc(ctx context.Context, name string, out map[string]PolicyDoc) error { + if name == s.doc { + if err := s.block(ctx); err != nil { + return err + } + } + return s.IAMStorageAPI.loadPolicyDoc(ctx, name, out) +} + +// The batch path and the request path share one IAM cache. A lazy load that +// batch authorization needs, of a mapping that is never cached (a user whose +// policies come from a group) or of an evicted policy document, must not +// hold request authentication and cached authorization while storage is +// slow, and must end at the deadline of the job's context. +func TestBatchAuthorizeSlowLoadKeepsCachedAuthorizationAvailable(t *testing.T) { + for _, tc := range []struct { + name string + // submitter returns the submitter whose authorization lazily loads, + // and arranges for blocked to hold that load. + submitter func(t *testing.T, f *batchAuthzFixture, blocked *batchBlockedIAMStore) *batchJobSubmitter + }{ + {"mapping", func(t *testing.T, f *batchAuthzFixture, blocked *batchBlockedIAMStore) *batchJobSubmitter { + // No mapping of the user itself is stored, so every lookup of + // its policies reads storage before it falls back to the group. + cred := iamAdminUser(f.ctx, t, f.sys, "snapshot-group-user", "") + _, err := f.sys.AddUsersToGroup(f.ctx, "snapshot-group", []string{cred.AccessKey}) + mustIAM(t, err) + _, err = f.sys.PolicyDBSet(f.ctx, "snapshot-group", "snapshot-full", regUser, true) + mustIAM(t, err) + blocked.mapping = cred.AccessKey + sub, err := newBatchJobSubmitter(cred) + mustIAM(t, err) + return sub + }}, + {"document", func(t *testing.T, f *batchAuthzFixture, blocked *batchBlockedIAMStore) *batchJobSubmitter { + batchAuthzPolicy(f.ctx, t, f.sys, "snapshot-evicted", batchAuthzFullAccess) + cred := iamAdminUser(f.ctx, t, f.sys, "snapshot-doc-user", "snapshot-evicted") + cache := f.sys.store.lock() + delete(cache.iamPolicyDocsMap, "snapshot-evicted") + f.sys.store.unlock() + blocked.doc = "snapshot-evicted" + sub, err := newBatchJobSubmitter(cred) + mustIAM(t, err) + return sub + }}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newBatchAuthzFixture(t) + batchAuthzPolicy(f.ctx, t, f.sys, "snapshot-full", batchAuthzFullAccess) + cached := iamAdminUser(f.ctx, t, f.sys, "snapshot-cached-user", "snapshot-full") + store := f.sys.store + original := store.IAMStorageAPI + blocked := &batchBlockedIAMStore{IAMStorageAPI: original, started: make(chan struct{}), release: make(chan struct{})} + sub := tc.submitter(t, f, blocked) + req := batchAuthzRequest{action: policy.GetObjectAction, bucket: f.bucket, object: "key", versionID: mustGetUUID()} + if err := sub.authorize(f.ctx, req); err != nil { + t.Fatalf("control: %v", err) + } + store.IAMStorageAPI = blocked + var wg sync.WaitGroup + var once sync.Once + unblock := func() { once.Do(func() { close(blocked.release) }) } + t.Cleanup(func() { unblock(); wg.Wait(); store.IAMStorageAPI = original }) + + slowDone := make(chan error, 1) + wg.Go(func() { slowDone <- sub.authorize(f.ctx, req) }) + select { + case <-blocked.started: + case <-time.After(5 * time.Second): + t.Fatal("the lazy load did not start") + } + cachedDone := make(chan bool, 2) + wg.Go(func() { + _, ok := f.sys.GetUser(f.ctx, cached.AccessKey) + cachedDone <- ok + }) + wg.Go(func() { + cachedDone <- f.sys.IsAllowed(policy.Args{AccountName: cached.AccessKey, Action: policy.GetObjectAction, BucketName: f.bucket, ObjectName: "key"}) + }) + for range 2 { + select { + case ok := <-cachedDone: + if !ok { + t.Fatal("cached authentication or authorization was rejected") + } + case <-time.After(time.Second): + t.Fatal("a slow IAM load during batch authorization blocked cached authentication") + } + } + unblock() + if err := <-slowDone; err != nil { + t.Fatalf("authorization after the load completed: %v", err) + } + + // A load that does not return ends with the job's context, as an + // unavailable dependency, not with the storage layer's patience. + blocked.release = make(chan struct{}) + once = sync.Once{} + unblock = func() { once.Do(func() { close(blocked.release) }) } + ctx, cancel := context.WithTimeout(f.ctx, 200*time.Millisecond) + defer cancel() + deadlineDone := make(chan error, 1) + wg.Go(func() { deadlineDone <- sub.authorize(ctx, req) }) + select { + case err := <-deadlineDone: + if !isBatchJobUnavailable(err) { + t.Fatalf("a load that ended with the job's context: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("a lazy load outlived the job's context") + } + }) + } +} diff --git a/cmd/batch-authz.go b/cmd/batch-authz.go index e5e2ccc30..f22a7474b 100644 --- a/cmd/batch-authz.go +++ b/cmd/batch-authz.go @@ -15,6 +15,7 @@ import ( "fmt" "strconv" "strings" + "sync" "time" "github.com/minio/minio-go/v7/pkg/tags" @@ -22,6 +23,7 @@ import ( "github.com/minio/minio/internal/auth" "github.com/pgsty/silo-pkg/v3/policy" "github.com/pgsty/silo-pkg/v3/policy/condition" + "github.com/puzpuzpuz/xsync/v3" ) // A batch job acts on objects long after the request that submitted it has @@ -467,12 +469,12 @@ func (p batchJobPrincipal) authorizeArgs(args policy.Args, req *batchAuthzReques ctx = GlobalContext } plugin := newGlobalAuthZPluginFn() - var cache *iamCache + var snap *batchPolicySnapshot if plugin == nil && !args.IsOwner { - cache = globalIAMSys.store.lock() - defer globalIAMSys.store.unlock() + snap = globalIAMSys.store.batchSnapshot(ctx) + defer snap.release() } - policies, err := globalIAMSys.batchLocalPolicies(ctx, p, args, plugin == nil, cache) + policies, err := globalIAMSys.batchLocalPolicies(p, args, plugin == nil, snap) if err != nil { if errors.Is(err, errNoSuchPolicy) || errors.Is(err, errAuthentication) || errors.Is(err, errNoSuchUser) || errors.Is(err, errNoSuchGroup) { return denied @@ -555,24 +557,99 @@ func (p batchJobPrincipal) sessionPolicyDecision(args policy.Args) (bool, bool) return false, false } +// batchIAMLoadTimeout bounds one storage read of a lazy IAM load during +// batch authorization, as MergePolicies bounds a policy document load. +const batchIAMLoadTimeout = 5 * time.Second + +// batchPolicySnapshot reads the mappings and policy documents of a submitter +// at one cache revision while cached authentication and authorization +// proceed. +// +// It holds the store's writer mutex, so no writer changes the cache between +// its reads, and the cache read lock, which it releases around each storage +// read: a lazy mapping load or a policy document load. Request +// authentication then neither waits for storage nor queues behind a writer +// waiting for the cache lock, which taking the write lock back after each +// read would cause. It writes nothing to the cache but the lazily loaded +// mappings, which the request path stores under the read lock as well. +type batchPolicySnapshot struct { + *IAMStoreSys + ctx context.Context + cache *iamCache + mu sync.Mutex + reads int // storage reads in progress; the read lock is released while any +} + +// batchSnapshot takes the snapshot; release it with release. +func (store *IAMStoreSys) batchSnapshot(ctx context.Context) *batchPolicySnapshot { + store.writeMu.Lock() + return &batchPolicySnapshot{IAMStoreSys: store, ctx: ctx, cache: store.IAMStorageAPI.rlock()} +} + +func (s *batchPolicySnapshot) release() { + s.IAMStorageAPI.runlock() + s.writeMu.Unlock() +} + +// read runs one storage read without the cache lock, bounded by the job's +// context and batchIAMLoadTimeout. Concurrent reads (the group fallback of +// policyDBGetGroups) release the lock before the first and take it back +// after the last. +func (s *batchPolicySnapshot) read(fn func(ctx context.Context) error) error { + s.mu.Lock() + if s.reads == 0 { + s.IAMStorageAPI.runlock() + } + s.reads++ + s.mu.Unlock() + defer func() { + s.mu.Lock() + defer s.mu.Unlock() + s.reads-- + if s.reads == 0 { + s.IAMStorageAPI.rlock() + } + }() + ctx, cancel := context.WithTimeout(s.ctx, batchIAMLoadTimeout) + defer cancel() + return fn(ctx) +} + +func (s *batchPolicySnapshot) loadMappedPolicy(_ context.Context, name string, userType IAMUserType, isGroup bool, m *xsync.MapOf[string, MappedPolicy]) error { + return s.read(func(ctx context.Context) error { + return s.IAMStoreSys.loadMappedPolicy(ctx, name, userType, isGroup, m) + }) +} + +// loadGroup would write the cache under the read lock. The snapshot resolves +// the policies of users, which never loads a group. +func (s *batchPolicySnapshot) loadGroup(context.Context, string, map[string]GroupInfo) error { + return errors.New("batch authorization does not load groups") +} + +func (s *batchPolicySnapshot) loadPolicyDoc(name string, out map[string]PolicyDoc) error { + return s.read(func(ctx context.Context) error { return s.IAMStoreSys.loadPolicyDoc(ctx, name, out) }) +} + // batchLocalPolicies returns every local policy that can take part in the // decision on args: the mapped policies that IsAllowed would merge when // withMapped is set (no authorization plugin), and the session policy. -func (sys *IAMSys) batchLocalPolicies(ctx context.Context, p batchJobPrincipal, args policy.Args, withMapped bool, cache *iamCache) ([]policy.Policy, error) { +func (sys *IAMSys) batchLocalPolicies(p batchJobPrincipal, args policy.Args, withMapped bool, snap *batchPolicySnapshot) ([]policy.Policy, error) { var out []policy.Policy if withMapped && args.IsOwner { // IsAllowed does not evaluate any policy for the owner. return nil, nil } if withMapped { - // The caller holds the IAM writer lock through evaluation. Lazy mapping - // loads and selected documents therefore belong to one cache revision. + // The caller holds snap through evaluation. Lazy mapping loads and + // selected documents therefore belong to one cache revision. + cache := snap.cache mappedPolicies := func(name string) ([]string, error) { - names, _, err := cache.policyDBGet(sys.store, name, false, false) + names, _, err := cache.policyDBGet(snap, name, false, false) if err != nil { return nil, err } - groups, err := cache.policyDBGetGroups(sys.store, len(names) > 0, args.Groups...) + groups, err := cache.policyDBGetGroups(snap, len(names) > 0, args.Groups...) return append(names, groups...), err } var names []string @@ -611,10 +688,7 @@ func (sys *IAMSys) batchLocalPolicies(ctx context.Context, p batchJobPrincipal, doc, ok := cache.iamPolicyDocsMap[name] if !ok { loaded := make(map[string]PolicyDoc) - loadCtx, cancel := context.WithTimeout(ctx, 5*time.Second) - err := sys.store.loadPolicyDoc(loadCtx, name, loaded) - cancel() - if err != nil { + if err := snap.loadPolicyDoc(name, loaded); err != nil { return nil, err } doc, ok = loaded[name] diff --git a/cmd/iam-store.go b/cmd/iam-store.go index b45534993..a647d24a7 100644 --- a/cmd/iam-store.go +++ b/cmd/iam-store.go @@ -372,7 +372,18 @@ func (c *iamCache) removeGroupFromMembershipsMap(group string) { } } -func (c *iamCache) policyDBGetGroups(store *IAMStoreSys, userPolicyPresent bool, groups ...string) ([]string, error) { +// iamMappingLoader is what policyDBGet and policyDBGetGroups need of the +// store: the identity backend in use and the loads of a mapping or a group +// the cache does not hold. *IAMStoreSys loads under the lock its caller +// holds; a caller that must not hold the cache lock across a storage read +// wraps it (see batchPolicySnapshot). +type iamMappingLoader interface { + getUsersSysType() UsersSysType + loadGroup(ctx context.Context, group string, m map[string]GroupInfo) error + loadMappedPolicy(ctx context.Context, name string, userType IAMUserType, isGroup bool, m *xsync.MapOf[string, MappedPolicy]) error +} + +func (c *iamCache) policyDBGetGroups(store iamMappingLoader, userPolicyPresent bool, groups ...string) ([]string, error) { var policies []string for _, group := range groups { if store.getUsersSysType() == MinIOUsersSysType { @@ -444,7 +455,7 @@ func (c *iamCache) policyDBGetGroups(store *IAMStoreSys, userPolicyPresent bool, // information in IAM (i.e sys.iam*Map) - this info is stored only in the STS // generated credentials. Thus we skip looking up group memberships, user map, // and group map and check the appropriate policy maps directly. -func (c *iamCache) policyDBGet(store *IAMStoreSys, name string, isGroup bool, policyPresent bool) ([]string, time.Time, error) { +func (c *iamCache) policyDBGet(store iamMappingLoader, name string, isGroup bool, policyPresent bool) ([]string, time.Time, error) { if isGroup { if store.getUsersSysType() == MinIOUsersSysType { g, ok := c.iamGroupsMap[name]