diff --git a/weed/s3api/auth_credentials.go b/weed/s3api/auth_credentials.go index 11e575806e..f1f520c8bb 100644 --- a/weed/s3api/auth_credentials.go +++ b/weed/s3api/auth_credentials.go @@ -113,8 +113,9 @@ type IdentityAccessManagement struct { // reloadCh coalesces failed-reload retries handled by reloadRetryLoop reloadCh chan struct{} - // reloadMu serializes configuration loads end-to-end (store snapshot - // through commit) so an older snapshot cannot overwrite a newer one + // reloadMu serializes configuration loads and pushes end-to-end (store + // snapshot through commit) so an older snapshot cannot overwrite a newer + // one reloadMu sync.Mutex } @@ -1027,10 +1028,6 @@ func (iam *IdentityAccessManagement) MergeS3ApiConfiguration(config *iam_pb.S3Ap for k, v := range iam.nameToIdentity { nameToIdentity[k] = v } - policies := make(map[string]*iam_pb.Policy) - for k, v := range iam.policies { - policies[k] = v - } accounts := make(map[string]*Account) for k, v := range iam.accounts { accounts[k] = v @@ -1246,22 +1243,6 @@ func (iam *IdentityAccessManagement) MergeS3ApiConfiguration(config *iam_pb.S3Ap glog.V(3).Infof("Loaded service account %s for dynamic parent %s (expiration: %d)", sa.Id, sa.ParentUser, sa.Expiration) } - for _, policy := range config.Policies { - policies[policy.Name] = policy - } - // full snapshot: drop dynamic policies the store no longer has - if isFullState { - presentPolicies := make(map[string]bool, len(config.Policies)) - for _, policy := range config.Policies { - presentPolicies[policy.Name] = true - } - for name := range policies { - if !staticPolicies[name] && !presentPolicies[name] { - delete(policies, name) - } - } - } - iam.m.Lock() // atomically switch iam.identities = identities @@ -1270,7 +1251,28 @@ func (iam *IdentityAccessManagement) MergeS3ApiConfiguration(config *iam_pb.S3Ap iam.emailAccount = emailAccount iam.nameToIdentity = nameToIdentity iam.accessKeyIdent = accessKeyIdent - iam.policies = policies + + // Apply policy changes to the live map rather than swapping in a snapshot + // taken before this merge prepared: a policy committed in the meantime (a + // pushed inline document or a PutPolicy update) must not be discarded. + if iam.policies == nil { + iam.policies = make(map[string]*iam_pb.Policy) + } + for _, policy := range config.Policies { + iam.policies[policy.Name] = policy + } + // full snapshot: drop dynamic policies the store no longer has + if isFullState { + presentPolicies := make(map[string]bool, len(config.Policies)) + for _, policy := range config.Policies { + presentPolicies[policy.Name] = true + } + for name := range iam.policies { + if !staticPolicies[name] && !presentPolicies[name] { + delete(iam.policies, name) + } + } + } // Groups: a full snapshot is authoritative even when empty (last group // deleted); partial updates pass nil Groups and preserve existing state. @@ -1366,9 +1368,51 @@ func (iam *IdentityAccessManagement) UpsertIdentity(ident *iam_pb.Identity) erro return fmt.Errorf("upsert identity failed: empty identity name") } glog.V(1).Infof("IAM: upsert identity %s", ident.Name) - return iam.MergeS3ApiConfiguration(&iam_pb.S3ApiConfiguration{ + // Serialize the store reads below with the merge that publishes them: + // reloadMu already gives full reloads the same snapshot-through-commit + // ordering, so a document read here cannot overwrite a newer copy that a + // concurrent push or reload committed in the meantime. + iam.reloadMu.Lock() + defer iam.reloadMu.Unlock() + config := &iam_pb.S3ApiConfiguration{ Identities: []*iam_pb.Identity{ident}, - }, false, false) + } + // The wire proto lacks runtime inline policy names; merging it raw would + // drop them until the next reload and authorize by flattened Actions. + // Reattach the names and their documents, scoped to this identity: a name + // without its document would deny requests the policy allows until the + // next reload. + if iam.credentialManager != nil { + if store, ok := iam.credentialManager.GetStore().(credential.InlinePolicyStore); ok { + policyNames, err := store.ListUserInlinePolicies(context.Background(), ident.Name) + if err != nil { + return err + } + for _, policyName := range policyNames { + runtimeName := inlinePolicyRuntimeName(ident.Name, policyName) + ident.PolicyNames = appendUniquePolicyName(ident.PolicyNames, runtimeName) + doc, err := store.GetUserInlinePolicy(context.Background(), ident.Name, policyName) + if err != nil { + return err + } + if doc == nil { + continue + } + content, err := json.Marshal(doc) + if err != nil { + return fmt.Errorf("failed to marshal inline policy %q for user %q: %w", policyName, ident.Name, err) + } + config.Policies = append(config.Policies, &iam_pb.Policy{Name: runtimeName, Content: string(content)}) + } + } + } + if err := iam.MergeS3ApiConfiguration(config, false, false); err != nil { + return err + } + if len(config.Policies) > 0 { + iam.resyncIAMManagerPolicies() + } + return nil } // isEnabled reports whether S3 auth should be enforced for this server. @@ -3190,8 +3234,42 @@ func (iam *IdentityAccessManagement) PutGroup(group *iam_pb.Group) error { } glog.V(1).Infof("IAM: put group %s", group.Name) + // Serialize the store reads below with their installation under iam.m: + // the same reloadMu ordering that identity pushes and full reloads use, + // so a document read here cannot overwrite a newer committed copy. + iam.reloadMu.Lock() + defer iam.reloadMu.Unlock() + + // As for identity pushes, the wire proto lacks the group's runtime + // inline policy names; reattach them with their documents, scoped to + // this group. + var inlinePolicies []*iam_pb.Policy + if iam.credentialManager != nil { + if store, ok := iam.credentialManager.GetStore().(credential.GroupInlinePolicyStore); ok { + policyNames, err := store.ListGroupInlinePolicies(context.Background(), group.Name) + if err != nil { + return err + } + for _, policyName := range policyNames { + runtimeName := inlineGroupPolicyRuntimeName(group.Name, policyName) + group.PolicyNames = appendUniquePolicyName(group.PolicyNames, runtimeName) + doc, err := store.GetGroupInlinePolicy(context.Background(), group.Name, policyName) + if err != nil { + return err + } + if doc == nil { + continue + } + content, err := json.Marshal(doc) + if err != nil { + return fmt.Errorf("failed to marshal inline policy %q for group %q: %w", policyName, group.Name, err) + } + inlinePolicies = append(inlinePolicies, &iam_pb.Policy{Name: runtimeName, Content: string(content)}) + } + } + } + iam.m.Lock() - defer iam.m.Unlock() // Remove old reverse index entries for this group if old, ok := iam.groups[group.Name]; ok && !old.Disabled { @@ -3209,6 +3287,20 @@ func (iam *IdentityAccessManagement) PutGroup(group *iam_pb.Group) error { } } + if len(inlinePolicies) > 0 { + if iam.policies == nil { + iam.policies = make(map[string]*iam_pb.Policy) + } + for _, policy := range inlinePolicies { + iam.policies[policy.Name] = policy + } + iam.rebuildIAMPolicyEngineLocked() + } + iam.m.Unlock() + + if len(inlinePolicies) > 0 { + iam.resyncIAMManagerPolicies() + } return nil } diff --git a/weed/s3api/auth_credentials_upsert_test.go b/weed/s3api/auth_credentials_upsert_test.go new file mode 100644 index 0000000000..bc272f330a --- /dev/null +++ b/weed/s3api/auth_credentials_upsert_test.go @@ -0,0 +1,369 @@ +package s3api + +import ( + "context" + "net/http" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/credential" + "github.com/seaweedfs/seaweedfs/weed/pb/iam_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine" + s3_constants "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// PutIdentity pushes carry the persisted identity proto, which records the +// inline policy flattened into Actions but not its runtime policy name. The +// upsert must keep the identity on the attached-policy route; otherwise its +// coarse "Write:bucket" action authorizes deletes the policy never granted. +// The push must also install the policy document itself: the inline policy +// here is never loaded by a full reload, so only the push can supply it. +func TestUpsertIdentityKeepsInlinePolicyAuthorization(t *testing.T) { + s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{{ + Name: "writer", + Credentials: []*iam_pb.Credential{{AccessKey: "AKW", SecretKey: "secret"}}, + Actions: []string{"Read:bucket", "Write:bucket", "List:bucket"}, + }}) + ctx := context.Background() + + var doc policy_engine.PolicyDocument + require.NoError(t, doc.UnmarshalJSON([]byte(`{ + "Version":"2012-10-17", + "Statement":[{ + "Effect":"Allow", + "Action":["s3:GetObject","s3:PutObject","s3:ListBucket"], + "Resource":["arn:aws:s3:::bucket","arn:aws:s3:::bucket/*"] + }] + }`))) + require.NoError(t, s3a.iam.credentialManager.PutUserInlinePolicy(ctx, "writer", "writer", doc)) + + pushed, err := s3a.iam.credentialManager.GetUser(ctx, "writer") + require.NoError(t, err) + require.NotContains(t, pushed.PolicyNames, inlinePolicyRuntimeName("writer", "writer"), + "persisted protos do not carry runtime inline policy names") + require.NoError(t, s3a.iam.UpsertIdentity(pushed)) + + ident := s3a.iam.lookupByIdentityName("writer") + require.NotNil(t, ident) + require.Contains(t, ident.PolicyNames, inlinePolicyRuntimeName("writer", "writer")) + + _, err = s3a.iam.GetPolicy(inlinePolicyRuntimeName("writer", "writer")) + require.NoError(t, err, "the push must install the inline policy document, not just its name") + + req, err := http.NewRequest(http.MethodGet, "http://s3/bucket/obj", nil) + require.NoError(t, err) + assert.Equal(t, s3err.ErrNone, + s3a.iam.VerifyActionPermission(req, ident, s3_constants.ACTION_READ, "bucket", "obj"), + "an allowed request must succeed through the pushed inline policy") + + req, err = http.NewRequest(http.MethodDelete, "http://s3/bucket/obj", nil) + require.NoError(t, err) + assert.Equal(t, s3err.ErrAccessDenied, + s3a.iam.VerifyActionPermission(req, ident, s3_constants.ACTION_WRITE, "bucket", "obj"), + "an identity pushed without its inline policy must not fall back to legacy Actions") +} + +// A pushed identity must also refresh a cached inline policy document that has +// since been replaced in the store; otherwise the stale allow keeps +// authorizing requests the new policy revoked. +func TestUpsertIdentityRefreshesInlinePolicyDocument(t *testing.T) { + s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{{ + Name: "writer", + Credentials: []*iam_pb.Credential{{AccessKey: "AKW", SecretKey: "secret"}}, + Actions: []string{"Read:bucket", "Write:bucket", "List:bucket"}, + }}) + ctx := context.Background() + + var doc policy_engine.PolicyDocument + require.NoError(t, doc.UnmarshalJSON([]byte(`{ + "Version":"2012-10-17", + "Statement":[{"Effect":"Allow","Action":"s3:GetObject","Resource":"*"}] + }`))) + require.NoError(t, s3a.iam.credentialManager.PutUserInlinePolicy(ctx, "writer", "writer", doc)) + require.NoError(t, s3a.iam.LoadS3ApiConfigurationFromCredentialManager()) + + ident := s3a.iam.lookupByIdentityName("writer") + require.NotNil(t, ident) + req, err := http.NewRequest(http.MethodGet, "http://s3/bucket/obj", nil) + require.NoError(t, err) + require.Equal(t, s3err.ErrNone, + s3a.iam.VerifyActionPermission(req, ident, s3_constants.ACTION_READ, "bucket", "obj")) + + require.NoError(t, doc.UnmarshalJSON([]byte(`{ + "Version":"2012-10-17", + "Statement":[{"Effect":"Allow","Action":"s3:ListBucket","Resource":"*"}] + }`))) + require.NoError(t, s3a.iam.credentialManager.PutUserInlinePolicy(ctx, "writer", "writer", doc)) + + pushed, err := s3a.iam.credentialManager.GetUser(ctx, "writer") + require.NoError(t, err) + require.NoError(t, s3a.iam.UpsertIdentity(pushed)) + + ident = s3a.iam.lookupByIdentityName("writer") + require.NotNil(t, ident) + assert.Equal(t, s3err.ErrAccessDenied, + s3a.iam.VerifyActionPermission(req, ident, s3_constants.ACTION_READ, "bucket", "obj"), + "a push must replace the stale cached document, not keep the old allow") +} + +// The same applies to PutGroup pushes: the persisted group proto lacks the +// group's runtime inline policy names, and the inline document is never +// loaded by a full reload here, so only the push can supply it. +func TestPutGroupKeepsInlinePolicyAuthorization(t *testing.T) { + s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{{Name: "alice"}}) + ctx := context.Background() + + require.NoError(t, s3a.iam.credentialManager.CreateGroup(ctx, + &iam_pb.Group{Name: "devs", Members: []string{"alice"}})) + var doc policy_engine.PolicyDocument + require.NoError(t, doc.UnmarshalJSON([]byte(`{ + "Version":"2012-10-17", + "Statement":[{"Effect":"Allow","Action":"s3:GetObject","Resource":"*"}] + }`))) + require.NoError(t, s3a.iam.credentialManager.PutGroupInlinePolicy(ctx, "devs", "p", doc)) + + pushed, err := s3a.iam.credentialManager.GetGroup(ctx, "devs") + require.NoError(t, err) + require.NoError(t, s3a.iam.PutGroup(pushed)) + + s3a.iam.m.RLock() + require.Contains(t, s3a.iam.groups["devs"].PolicyNames, inlineGroupPolicyRuntimeName("devs", "p")) + s3a.iam.m.RUnlock() + + _, err = s3a.iam.GetPolicy(inlineGroupPolicyRuntimeName("devs", "p")) + require.NoError(t, err, "the push must install the group's inline policy document") + + ident := s3a.iam.lookupByIdentityName("alice") + require.NotNil(t, ident) + req, err := http.NewRequest(http.MethodGet, "http://s3/bucket/obj", nil) + require.NoError(t, err) + assert.Equal(t, s3err.ErrNone, + s3a.iam.VerifyActionPermission(req, ident, s3_constants.ACTION_READ, "bucket", "obj"), + "a pushed group must keep its inline policy grant active") +} + +// pausingInlineStore pauses only the first user and first group inline policy +// Get, after capturing the document, so a test can interleave a newer write +// between a push's store read and the install of what it read. Later calls +// proceed immediately. +type pausingInlineStore struct { + credential.CredentialStore + + userGets int32 + userPaused chan struct{} + userRelease chan struct{} + groupGets int32 + groupPaused chan struct{} + groupRelease chan struct{} +} + +func (s *pausingInlineStore) userPolicies() credential.InlinePolicyStore { + return s.CredentialStore.(credential.InlinePolicyStore) +} + +func (s *pausingInlineStore) groupPolicies() credential.GroupInlinePolicyStore { + return s.CredentialStore.(credential.GroupInlinePolicyStore) +} + +func (s *pausingInlineStore) PutUserInlinePolicy(ctx context.Context, userName, policyName string, doc policy_engine.PolicyDocument) error { + return s.userPolicies().PutUserInlinePolicy(ctx, userName, policyName, doc) +} + +func (s *pausingInlineStore) GetUserInlinePolicy(ctx context.Context, userName, policyName string) (*policy_engine.PolicyDocument, error) { + doc, err := s.userPolicies().GetUserInlinePolicy(ctx, userName, policyName) + if err != nil { + return nil, err + } + if atomic.AddInt32(&s.userGets, 1) == 1 { + close(s.userPaused) + <-s.userRelease + } + return doc, nil +} + +func (s *pausingInlineStore) DeleteUserInlinePolicy(ctx context.Context, userName, policyName string) error { + return s.userPolicies().DeleteUserInlinePolicy(ctx, userName, policyName) +} + +func (s *pausingInlineStore) ListUserInlinePolicies(ctx context.Context, userName string) ([]string, error) { + return s.userPolicies().ListUserInlinePolicies(ctx, userName) +} + +func (s *pausingInlineStore) PutGroupInlinePolicy(ctx context.Context, groupName, policyName string, doc policy_engine.PolicyDocument) error { + return s.groupPolicies().PutGroupInlinePolicy(ctx, groupName, policyName, doc) +} + +func (s *pausingInlineStore) GetGroupInlinePolicy(ctx context.Context, groupName, policyName string) (*policy_engine.PolicyDocument, error) { + doc, err := s.groupPolicies().GetGroupInlinePolicy(ctx, groupName, policyName) + if err != nil { + return nil, err + } + if atomic.AddInt32(&s.groupGets, 1) == 1 { + close(s.groupPaused) + <-s.groupRelease + } + return doc, nil +} + +func (s *pausingInlineStore) DeleteGroupInlinePolicy(ctx context.Context, groupName, policyName string) error { + return s.groupPolicies().DeleteGroupInlinePolicy(ctx, groupName, policyName) +} + +func (s *pausingInlineStore) ListGroupInlinePolicies(ctx context.Context, groupName string) ([]string, error) { + return s.groupPolicies().ListGroupInlinePolicies(ctx, groupName) +} + +// A push paused after reading a permissive inline policy must not overwrite a +// tighter document a later push already installed: once the pause lifts, the +// stale read may still land, but the newer push must be the state that wins. +func TestUpsertIdentityRaceKeepsNewerInlinePolicy(t *testing.T) { + s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{{ + Name: "writer", + Credentials: []*iam_pb.Credential{{AccessKey: "AKW", SecretKey: "secret"}}, + }}) + ctx := context.Background() + + pauser := &pausingInlineStore{ + CredentialStore: s3a.iam.credentialManager.GetStore(), + userPaused: make(chan struct{}), + userRelease: make(chan struct{}), + } + s3a.iam.credentialManager.Store = pauser + + var docV1 policy_engine.PolicyDocument + require.NoError(t, docV1.UnmarshalJSON([]byte( + `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:GetObject","Resource":"*"}]}`))) + require.NoError(t, s3a.iam.credentialManager.PutUserInlinePolicy(ctx, "writer", "p", docV1)) + + pushed, err := s3a.iam.credentialManager.GetUser(ctx, "writer") + require.NoError(t, err) + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + assert.NoError(t, s3a.iam.UpsertIdentity(pushed)) + }() + + select { + case <-pauser.userPaused: + case <-time.After(5 * time.Second): + t.Fatal("first push did not reach the inline policy read") + } + + // The administrator tightens the policy; its push must be the surviving one. + var docV2 policy_engine.PolicyDocument + require.NoError(t, docV2.UnmarshalJSON([]byte( + `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:ListBucket","Resource":"*"}]}`))) + require.NoError(t, s3a.iam.credentialManager.PutUserInlinePolicy(ctx, "writer", "p", docV2)) + + wg.Add(1) + go func() { + defer wg.Done() + assert.NoError(t, s3a.iam.UpsertIdentity(pushed)) + }() + + // Give the second push a moment to install the newer document; with the + // ordering fix it instead waits for the first push and this poll expires. + deadline := time.Now().Add(500 * time.Millisecond) + for time.Now().Before(deadline) { + if p, err := s3a.iam.GetPolicy(inlinePolicyRuntimeName("writer", "p")); err == nil && strings.Contains(p.Content, "s3:ListBucket") { + break + } + time.Sleep(time.Millisecond) + } + close(pauser.userRelease) + wg.Wait() + + installed, err := s3a.iam.GetPolicy(inlinePolicyRuntimeName("writer", "p")) + require.NoError(t, err) + assert.JSONEq(t, + `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:ListBucket","Resource":"*"}]}`, + installed.Content, "the newer pushed document must win over the stale read") + + ident := s3a.iam.lookupByIdentityName("writer") + require.NotNil(t, ident) + req, err := http.NewRequest(http.MethodGet, "http://s3/bucket/obj", nil) + require.NoError(t, err) + assert.Equal(t, s3err.ErrAccessDenied, + s3a.iam.VerifyActionPermission(req, ident, s3_constants.ACTION_READ, "bucket", "obj"), + "a stale push must not restore a grant the newer policy revoked") +} + +// Same ordering guarantee for group pushes: a paused PutGroup must not +// overwrite a tighter inline document installed by a later push. +func TestPutGroupRaceKeepsNewerInlinePolicy(t *testing.T) { + s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{{Name: "alice"}}) + ctx := context.Background() + + pauser := &pausingInlineStore{ + CredentialStore: s3a.iam.credentialManager.GetStore(), + groupPaused: make(chan struct{}), + groupRelease: make(chan struct{}), + } + s3a.iam.credentialManager.Store = pauser + + require.NoError(t, s3a.iam.credentialManager.CreateGroup(ctx, + &iam_pb.Group{Name: "devs", Members: []string{"alice"}})) + var docV1 policy_engine.PolicyDocument + require.NoError(t, docV1.UnmarshalJSON([]byte( + `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:GetObject","Resource":"*"}]}`))) + require.NoError(t, s3a.iam.credentialManager.PutGroupInlinePolicy(ctx, "devs", "p", docV1)) + + pushed, err := s3a.iam.credentialManager.GetGroup(ctx, "devs") + require.NoError(t, err) + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + assert.NoError(t, s3a.iam.PutGroup(pushed)) + }() + + select { + case <-pauser.groupPaused: + case <-time.After(5 * time.Second): + t.Fatal("first group push did not reach the inline policy read") + } + + var docV2 policy_engine.PolicyDocument + require.NoError(t, docV2.UnmarshalJSON([]byte( + `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:ListBucket","Resource":"*"}]}`))) + require.NoError(t, s3a.iam.credentialManager.PutGroupInlinePolicy(ctx, "devs", "p", docV2)) + + wg.Add(1) + go func() { + defer wg.Done() + assert.NoError(t, s3a.iam.PutGroup(pushed)) + }() + + deadline := time.Now().Add(500 * time.Millisecond) + for time.Now().Before(deadline) { + if p, err := s3a.iam.GetPolicy(inlineGroupPolicyRuntimeName("devs", "p")); err == nil && strings.Contains(p.Content, "s3:ListBucket") { + break + } + time.Sleep(time.Millisecond) + } + close(pauser.groupRelease) + wg.Wait() + + installed, err := s3a.iam.GetPolicy(inlineGroupPolicyRuntimeName("devs", "p")) + require.NoError(t, err) + assert.JSONEq(t, + `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:ListBucket","Resource":"*"}]}`, + installed.Content, "the newer pushed group document must win over the stale read") + + ident := s3a.iam.lookupByIdentityName("alice") + require.NotNil(t, ident) + req, err := http.NewRequest(http.MethodGet, "http://s3/bucket/obj", nil) + require.NoError(t, err) + assert.Equal(t, s3err.ErrAccessDenied, + s3a.iam.VerifyActionPermission(req, ident, s3_constants.ACTION_READ, "bucket", "obj"), + "a stale group push must not restore a grant the newer policy revoked") +}