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]