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 <chris.lu@gmail.com>
2026-08-20 06:07:05 +00:00 · 2026-05-21 00:39:42 -07:00
parent 7e4691f2dc
commit 9b9fdb5b76
3 changed files with 223 additions and 7 deletions
@@ -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")
@@ -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")
+}
@@ -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()