From 9b9fdb5b761e73e0b60e992f0de51575fda6b15a Mon Sep 17 00:00:00 2001 From: Mmx233 <36563672+Mmx233@users.noreply.github.com> Date: Thu, 21 May 2026 15:39:42 +0800 Subject: [PATCH] fix(s3): sync IAM policies to advanced IAM Manager policy engine (#9577) * fix(s3): sync IAM policies to advanced IAM Manager policy engine * test(s3): add unit tests for PutPolicy/DeletePolicy IAM Manager sync * fix(s3): flush loaded policies in SetIAMIntegration, drop extra reload Sync the policies already loaded from the credential store into the IAM Manager's engine from SetIAMIntegration itself, instead of re-running a full LoadS3ApiConfigurationFromCredentialManager after setup. This covers both startup orderings without a second filer round-trip or racing the async loader goroutine: if the load won, the policies are in memory to push; if SetIAMIntegration won, the load's own sync runs afterward. Move the runtime PutPolicy/DeletePolicy sync out of the iam.m write lock so the per-request auth RLock path isn't blocked by the policy recompile. * fix(s3): serialize IAM manager policy resync to avoid stale snapshots SyncRuntimePolicies replaces the manager's full policy set, so applying a policy view captured before a later mutation can resurrect a deleted policy or drop a new one. Funnel every path (PutPolicy, DeletePolicy, SetIAMIntegration, and the credential-manager load) through a single resyncIAMManagerPolicies that serializes on a dedicated mutex and reads iam.policies fresh at apply time, so the live map always wins regardless of interleaving. The load now installs the config into iam.policies before resyncing, closing the window where the manager held policies the map didn't yet have. --------- Co-authored-by: Chris Lu --- weed/s3api/auth_credentials.go | 65 ++++++- .../auth_credentials_policy_sync_test.go | 158 ++++++++++++++++++ weed/s3api/s3api_server.go | 7 + 3 files changed, 223 insertions(+), 7 deletions(-) create mode 100644 weed/s3api/auth_credentials_policy_sync_test.go diff --git a/weed/s3api/auth_credentials.go b/weed/s3api/auth_credentials.go index 0c09cd8db..7d714a765 100644 --- a/weed/s3api/auth_credentials.go +++ b/weed/s3api/auth_credentials.go @@ -68,6 +68,11 @@ type IdentityAccessManagement struct { // IAM Integration for advanced features iamIntegration IAMIntegration + // Serializes resyncs of the policy set into the advanced IAM manager so that + // concurrent policy updates can't apply stale views out of order. See + // resyncIAMManagerPolicies. + iamManagerSyncMu sync.Mutex + // Bucket policy engine for evaluating bucket policies policyEngine *BucketPolicyEngine @@ -1859,6 +1864,28 @@ func (iam *IdentityAccessManagement) syncRuntimePoliciesToIAMManager(ctx context return manager.SyncRuntimePolicies(ctx, policies) } +// resyncIAMManagerPolicies pushes the current policy set into the advanced IAM +// manager's engine. SyncRuntimePolicies treats its argument as the full desired +// state, so callers must not pass snapshots captured earlier: two updates that +// mutate iam.policies in order could otherwise apply their snapshots in the +// opposite order and resurrect a deleted policy or drop a new one. Instead this +// serializes on iamManagerSyncMu and reads iam.policies fresh, so whatever the +// map holds when the last caller runs becomes the manager's state. Call it after +// the iam.policies mutation has been committed (lock released). +func (iam *IdentityAccessManagement) resyncIAMManagerPolicies() { + if iam == nil { + return + } + iam.iamManagerSyncMu.Lock() + defer iam.iamManagerSyncMu.Unlock() + iam.m.RLock() + policies := iam.collectPoliciesLocked() + iam.m.RUnlock() + if err := iam.syncRuntimePoliciesToIAMManager(context.Background(), policies); err != nil { + glog.Warningf("Failed to sync runtime policies to IAM Manager: %v", err) + } +} + // PruneBucketFromConfiguration removes any identity actions scoped to the given // bucket (e.g. "Read:bucket", "Write:bucket/prefix") from the persisted S3 IAM // configuration. Wildcarded resources and global actions are preserved because @@ -1953,15 +1980,16 @@ func (iam *IdentityAccessManagement) LoadS3ApiConfigurationFromCredentialManager glog.Errorf("Failed to hydrate runtime IAM policies: %v", err) return err } - if err := iam.syncRuntimePoliciesToIAMManager(context.Background(), s3ApiConfiguration.Policies); err != nil { - glog.Errorf("Failed to sync runtime IAM policies to advanced IAM manager: %v", err) - return err - } + // Install the loaded config into iam.policies first, then resync the advanced + // IAM manager from that committed map. Syncing before the load would leave a + // window where the manager holds policies the map doesn't, which a concurrent + // resync (e.g. SetIAMIntegration or a runtime PutPolicy) would then clobber. if err := iam.loadS3ApiConfiguration(s3ApiConfiguration); err != nil { glog.Errorf("Failed to load S3 API configuration: %v", err) return err } + iam.resyncIAMManagerPolicies() glog.V(1).Infof("Successfully loaded S3 API configuration from credential manager") return nil @@ -2010,8 +2038,14 @@ func (iam *IdentityAccessManagement) initializeKMSFromJSON(configContent []byte) // isAuthEnabled themselves via EnableAuthEnforcement / updateAuthenticationState. func (iam *IdentityAccessManagement) SetIAMIntegration(integration *S3IAMIntegration) { iam.m.Lock() - defer iam.m.Unlock() iam.iamIntegration = integration + iam.m.Unlock() + // Config loaded before the integration was attached skipped the policy sync + // (syncRuntimePoliciesToIAMManager no-ops while iamIntegration is nil), so + // flush the policies already in memory into the manager's engine now. This + // covers either startup ordering: if the load won the race the policies are + // here to push; if SetIAMIntegration won, the load's own resync runs next. + iam.resyncIAMManagerPolicies() } // EnableAuthEnforcement turns on the auth-required mode unconditionally. Use @@ -2351,7 +2385,6 @@ func (iam *IdentityAccessManagement) authorizeWithIAM(r *http.Request, identity // PutPolicy adds or updates a policy func (iam *IdentityAccessManagement) PutPolicy(name string, content string) error { iam.m.Lock() - defer iam.m.Unlock() if iam.policies == nil { iam.policies = make(map[string]*iam_pb.Policy) } @@ -2362,6 +2395,12 @@ func (iam *IdentityAccessManagement) PutPolicy(name string, content string) erro if err := iam.iamPolicyEngine.SetBucketPolicy(name, content); err != nil { glog.Warningf("IAM policy %q is stored but could not be compiled for cache: %v", name, err) } + iam.m.Unlock() + // Also sync to the advanced IAM Manager's policy engine so that the + // authorizeWithIAM path (used when identity has policy_names) sees the update. + // Done after releasing iam.m so the per-request auth RLock path isn't blocked + // by the policy recompile. + iam.resyncIAMManagerPolicies() return nil } @@ -2378,14 +2417,26 @@ func (iam *IdentityAccessManagement) GetPolicy(name string) (*iam_pb.Policy, err // DeletePolicy removes a policy func (iam *IdentityAccessManagement) DeletePolicy(name string) error { iam.m.Lock() - defer iam.m.Unlock() delete(iam.policies, name) if iam.iamPolicyEngine != nil { _ = iam.iamPolicyEngine.DeleteBucketPolicy(name) } + iam.m.Unlock() + // Also sync the removal to the advanced IAM Manager's policy engine. + iam.resyncIAMManagerPolicies() return nil } +// collectPoliciesLocked returns all policies as a slice for SyncRuntimePolicies. +// Caller must hold iam.m (read or write). +func (iam *IdentityAccessManagement) collectPoliciesLocked() []*iam_pb.Policy { + policies := make([]*iam_pb.Policy, 0, len(iam.policies)) + for _, p := range iam.policies { + policies = append(policies, p) + } + return policies +} + func (iam *IdentityAccessManagement) PutGroup(group *iam_pb.Group) error { if group == nil { return fmt.Errorf("put group failed: nil group") diff --git a/weed/s3api/auth_credentials_policy_sync_test.go b/weed/s3api/auth_credentials_policy_sync_test.go new file mode 100644 index 000000000..a5bdeeca4 --- /dev/null +++ b/weed/s3api/auth_credentials_policy_sync_test.go @@ -0,0 +1,158 @@ +package s3api + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/iam/integration" + "github.com/seaweedfs/seaweedfs/weed/iam/policy" + "github.com/seaweedfs/seaweedfs/weed/iam/sts" + "github.com/seaweedfs/seaweedfs/weed/pb/iam_pb" + "github.com/stretchr/testify/require" +) + +func newTestIAMManager(t *testing.T) *integration.IAMManager { + t.Helper() + mgr := integration.NewIAMManager() + require.NoError(t, mgr.Initialize(&integration.IAMConfig{ + STS: &sts.STSConfig{ + TokenDuration: sts.FlexibleDuration{Duration: time.Hour}, + MaxSessionLength: sts.FlexibleDuration{Duration: 12 * time.Hour}, + Issuer: "test", + SigningKey: []byte("test-signing-key-32-characters!!"), + AccountId: "111122223333", + }, + Policy: &policy.PolicyEngineConfig{DefaultEffect: "Allow", StoreType: "memory"}, + Roles: &integration.RoleStoreConfig{StoreType: "memory"}, + }, func() string { return "localhost:8888" })) + return mgr +} + +func TestPutPolicy_SyncsToIAMManager(t *testing.T) { + mgr := newTestIAMManager(t) + iam := &IdentityAccessManagement{} + iam.SetIAMIntegration(NewS3IAMIntegration(mgr, "")) + + policyDoc, _ := json.Marshal(map[string]interface{}{ + "Version": "2012-10-17", + "Statement": []map[string]interface{}{ + {"Effect": "Allow", "Action": "s3:*", "Resource": "arn:aws:s3:::backup/*"}, + }, + }) + + require.NoError(t, iam.PutPolicy("BackupAll", string(policyDoc))) + + allowed, err := mgr.IsActionAllowed(context.Background(), &integration.ActionRequest{ + Principal: "arn:aws:iam::111122223333:user/test", + Action: "s3:PutObject", + Resource: "arn:aws:s3:::backup/file.txt", + PolicyNames: []string{"BackupAll"}, + }) + require.NoError(t, err) + require.True(t, allowed, "PutPolicy should sync to IAM Manager policy engine") +} + +func TestDeletePolicy_SyncsToIAMManager(t *testing.T) { + mgr := newTestIAMManager(t) + iam := &IdentityAccessManagement{} + iam.SetIAMIntegration(NewS3IAMIntegration(mgr, "")) + + policyDoc, _ := json.Marshal(map[string]interface{}{ + "Version": "2012-10-17", + "Statement": []map[string]interface{}{ + {"Effect": "Allow", "Action": "s3:*", "Resource": "arn:aws:s3:::backup/*"}, + }, + }) + + require.NoError(t, iam.PutPolicy("BackupAll", string(policyDoc))) + require.NoError(t, iam.DeletePolicy("BackupAll")) + + allowed, err := mgr.IsActionAllowed(context.Background(), &integration.ActionRequest{ + Principal: "arn:aws:iam::111122223333:user/test", + Action: "s3:PutObject", + Resource: "arn:aws:s3:::backup/file.txt", + PolicyNames: []string{"BackupAll"}, + }) + require.NoError(t, err) + require.False(t, allowed, "DeletePolicy should remove from IAM Manager policy engine") +} + +// TestSetIAMIntegration_FlushesLoadedPolicies reproduces the startup race: a +// policy is loaded into the IAM cache before the integration is attached (so the +// sync was skipped), then SetIAMIntegration must flush it into the manager's +// engine. Without the flush, policy_names identities get AccessDenied until an +// external IAM change triggers a reload. +func TestSetIAMIntegration_FlushesLoadedPolicies(t *testing.T) { + iam := &IdentityAccessManagement{} + + policyDoc, _ := json.Marshal(map[string]interface{}{ + "Version": "2012-10-17", + "Statement": []map[string]interface{}{ + {"Effect": "Allow", "Action": "s3:*", "Resource": "arn:aws:s3:::backup/*"}, + }, + }) + + // Integration not attached yet, so this only lands in the legacy engine. + require.NoError(t, iam.PutPolicy("BackupAll", string(policyDoc))) + + mgr := newTestIAMManager(t) + iam.SetIAMIntegration(NewS3IAMIntegration(mgr, "")) + + allowed, err := mgr.IsActionAllowed(context.Background(), &integration.ActionRequest{ + Principal: "arn:aws:iam::111122223333:user/test", + Action: "s3:PutObject", + Resource: "arn:aws:s3:::backup/file.txt", + PolicyNames: []string{"BackupAll"}, + }) + require.NoError(t, err) + require.True(t, allowed, "SetIAMIntegration should flush already-loaded policies into the IAM Manager") +} + +// TestResyncIAMManager_ReflectsCurrentPolicies pins the property that prevents +// the stale-snapshot race: SyncRuntimePolicies replaces the full desired set, so +// the resync must derive that set from the live iam.policies map at apply time, +// not from a view captured earlier. Here the map is mutated out from under an +// already-synced policy, and the next resync must converge the manager onto the +// new map rather than reinstating the old state. +func TestResyncIAMManager_ReflectsCurrentPolicies(t *testing.T) { + mgr := newTestIAMManager(t) + iam := &IdentityAccessManagement{} + iam.SetIAMIntegration(NewS3IAMIntegration(mgr, "")) + + doc := func(name string) string { + b, _ := json.Marshal(map[string]interface{}{ + "Version": "2012-10-17", + "Statement": []map[string]interface{}{ + {"Effect": "Allow", "Action": "s3:*", "Resource": "arn:aws:s3:::" + name + "/*"}, + }, + }) + return string(b) + } + allows := func(name string) bool { + allowed, err := mgr.IsActionAllowed(context.Background(), &integration.ActionRequest{ + Principal: "arn:aws:iam::111122223333:user/test", + Action: "s3:PutObject", + Resource: "arn:aws:s3:::" + name + "/file.txt", + PolicyNames: []string{name}, + }) + require.NoError(t, err) + return allowed + } + + require.NoError(t, iam.PutPolicy("alpha", doc("alpha"))) + require.True(t, allows("alpha")) + + // Swap the map contents without going through PutPolicy/DeletePolicy, then + // resync. A snapshot-based sync would push a pre-swap view; the fresh-read + // resync must mirror the current map: alpha gone, beta present. + iam.m.Lock() + delete(iam.policies, "alpha") + iam.policies["beta"] = &iam_pb.Policy{Name: "beta", Content: doc("beta")} + iam.m.Unlock() + iam.resyncIAMManagerPolicies() + + require.False(t, allows("alpha"), "resync should drop a policy no longer in the map") + require.True(t, allows("beta"), "resync should add a policy newly in the map") +} diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index 4aa109983..b3e6d9d25 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -323,6 +323,13 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl // IAM config file. Without one, EnableIam is the implicit mini default // and we must keep the "no credentials = allow all" startup behavior so // `docker run seaweedfs` works out of the box (fixes #9557). + // + // SetIAMIntegration also flushes the policies already loaded from the + // credential store into this manager's engine. The earlier config + // loads (the synchronous one in NewIdentityAccessManagementWithStore + // and the async goroutine) may have run before iamIntegration was set, + // in which case their syncRuntimePoliciesToIAMManager call was a no-op + // and identities relying on policy_names would get AccessDenied. iam.SetIAMIntegration(s3iam) if option.IamConfig != "" { iam.EnableAuthEnforcement()