From 6b6e6d85474594b3108b1c60956159e9fa4aba9c Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 27 Jul 2026 14:06:04 -0700 Subject: [PATCH] s3: apply filer identity changes despite a static config file (#10392) * s3: apply filer identity changes despite a static config file A -config file with inline identities disabled the metadata-subscription reload entirely, leaving the best-effort filer->s3 push as the only way s3.configure changes could reach a running gateway. Reload on IAM events regardless: the merge keeps the file's identities protected, and a full credential-manager snapshot now also drops dynamic identities the store no longer has, so revocation works without a restart. * s3: log identity propagation failures as warnings * s3: retry failed IAM reloads and reconcile policies and groups An event-driven reload that fails now hands off to a coalescing retry loop, so a transient filer error cannot strand a revoked credential until the next IAM event. Full-state merges also drop dynamic policies the store no longer has, keeping the static file's, and treat the group snapshot as authoritative even when empty. * s3: serialize IAM configuration loads The SIGHUP file reload, subscription reloads, the retry loop, and the postgres poll run on different goroutines. Without an end-to-end lock a load holding an older store snapshot can commit after a newer one and revert it. Hold reloadMu from snapshot through commit in both load entry points; partial merges from pushed updates stay lock-free and self-heal through the next event-driven reload. * s3: keep static-file groups through full-state reconciliation Group names from the static config file are tracked like identities and policies, and a full snapshot that does not carry them keeps the current definition and its memberships instead of dropping them. * credential: include groups in postgres configuration snapshots Full-state reconciliation treats absent groups as deleted, so a snapshot that never carries them would erase every dynamic group. * s3: revoke static-file groups dropped from the config file A file reload is authoritative for the file's group set while keeping dynamic groups, mirroring how full snapshots are authoritative for dynamic groups while keeping the file's. * credential: fail filer snapshots on unreadable entries A skipped identity or policy file made the load report success with an incomplete snapshot, which reconciliation reads as deletion and the retry loop never sees. Unparseable content is still skipped: it is durable, matches boot behavior, and must not block reloads forever. * s3: ignore groups in static config files Groups are managed through the IAM API and the dynamic store; no deployment defines them in a bootstrap config file. Ignoring them with a warning removes the two-directional group merge: full snapshots are plainly authoritative and file reloads never touch groups. --- .../filer_etc/filer_etc_identity.go | 4 +- weed/credential/filer_etc/filer_etc_policy.go | 5 +- .../filer_etc/filer_etc_policy_test.go | 3 +- weed/credential/postgres/postgres_identity.go | 29 ++- weed/credential/propagating_store.go | 6 +- weed/s3api/auth_credentials.go | 151 +++++++++++---- .../auth_credentials_static_config_test.go | 172 +++++++++++++++++- weed/s3api/auth_credentials_subscribe.go | 10 +- 8 files changed, 325 insertions(+), 55 deletions(-) diff --git a/weed/credential/filer_etc/filer_etc_identity.go b/weed/credential/filer_etc/filer_etc_identity.go index f3b093381..518b299b5 100644 --- a/weed/credential/filer_etc/filer_etc_identity.go +++ b/weed/credential/filer_etc/filer_etc_identity.go @@ -100,8 +100,8 @@ func (store *FilerEtcStore) loadFromMultiFile(ctx context.Context, s3cfg *iam_pb } else { c, err := filer.ReadInsideFiler(ctx, client, dir, entry.Name) if err != nil { - glog.Warningf("Failed to read identity file %s: %v", entry.Name, err) - continue + // fail the snapshot: a skipped identity would read as deleted + return fmt.Errorf("failed to read identity file %s: %w", entry.Name, err) } content = c } diff --git a/weed/credential/filer_etc/filer_etc_policy.go b/weed/credential/filer_etc/filer_etc_policy.go index 2379e6c28..9bf2e98ff 100644 --- a/weed/credential/filer_etc/filer_etc_policy.go +++ b/weed/credential/filer_etc/filer_etc_policy.go @@ -3,6 +3,7 @@ package filer_etc import ( "context" "encoding/json" + "fmt" "strings" "github.com/seaweedfs/seaweedfs/weed/credential" @@ -187,8 +188,8 @@ func (store *FilerEtcStore) loadPoliciesFromMultiFile(ctx context.Context, polic } else { c, err := filer.ReadInsideFiler(ctx, client, dir, entry.Name) if err != nil { - glog.Warningf("Failed to read policy file %s: %v", entry.Name, err) - continue + // fail the snapshot: a skipped policy would read as deleted + return fmt.Errorf("failed to read policy file %s: %w", entry.Name, err) } content = c } diff --git a/weed/credential/filer_etc/filer_etc_policy_test.go b/weed/credential/filer_etc/filer_etc_policy_test.go index ebf340770..d64cb5fd5 100644 --- a/weed/credential/filer_etc/filer_etc_policy_test.go +++ b/weed/credential/filer_etc/filer_etc_policy_test.go @@ -308,8 +308,9 @@ func TestFilerEtcStoreLoadManagedPoliciesRespectsReadContext(t *testing.T) { } server.mu.Unlock() + // the canceled read must fail the snapshot, not silently omit the policy managedPolicies, err := store.LoadManagedPolicies(ctx) - require.NoError(t, err) + require.ErrorContains(t, err, "cancel-me.json") assert.Empty(t, managedPolicies) } diff --git a/weed/credential/postgres/postgres_identity.go b/weed/credential/postgres/postgres_identity.go index 40c4e9d96..4c5da2d62 100644 --- a/weed/credential/postgres/postgres_identity.go +++ b/weed/credential/postgres/postgres_identity.go @@ -84,7 +84,34 @@ func (store *PostgresStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3Ap return nil, fmt.Errorf("failed iterating user rows: %w", err) } - glog.V(0).Infof("credential postgres: LoadConfiguration loaded %d identities", len(config.Identities)) + // groups are part of the snapshot: full-state reconciliation treats their + // absence as deletion + groupRows, err := store.db.QueryContext(ctx, "SELECT name, members, policy_names, disabled FROM groups") + if err != nil { + return nil, fmt.Errorf("failed to query groups: %w", err) + } + defer groupRows.Close() + for groupRows.Next() { + var name string + var membersJSON, policyNamesJSON []byte + var disabled bool + if err := groupRows.Scan(&name, &membersJSON, &policyNamesJSON, &disabled); err != nil { + return nil, fmt.Errorf("failed to scan group row: %w", err) + } + group := &iam_pb.Group{Name: name, Disabled: disabled} + if err := json.Unmarshal(membersJSON, &group.Members); err != nil { + return nil, fmt.Errorf("failed to unmarshal members for group %s: %w", name, err) + } + if err := json.Unmarshal(policyNamesJSON, &group.PolicyNames); err != nil { + return nil, fmt.Errorf("failed to unmarshal policy_names for group %s: %w", name, err) + } + config.Groups = append(config.Groups, group) + } + if err := groupRows.Err(); err != nil { + return nil, fmt.Errorf("failed iterating group rows: %w", err) + } + + glog.V(0).Infof("credential postgres: LoadConfiguration loaded %d identities, %d groups", len(config.Identities), len(config.Groups)) return config, nil } diff --git a/weed/credential/propagating_store.go b/weed/credential/propagating_store.go index 050aff28e..f561d2007 100644 --- a/weed/credential/propagating_store.go +++ b/weed/credential/propagating_store.go @@ -62,7 +62,7 @@ func (s *PropagatingCredentialStore) propagateChange(ctx context.Context, fn fun FilerGroup: s.masterClient.FilerGroup, }) if err != nil { - glog.V(1).Infof("failed to list S3 servers: %v", err) + glog.Warningf("failed to list S3 servers: %v", err) return err } for _, node := range resp.ClusterNodes { @@ -72,7 +72,7 @@ func (s *PropagatingCredentialStore) propagateChange(ctx context.Context, fn fun return nil }) if err != nil { - glog.V(1).Infof("failed to list s3 servers via master client: %v", err) + glog.Warningf("failed to list s3 servers via master client: %v", err) return } glog.V(1).Infof("IAM: propagating change to %d S3 servers: %v", len(s3Servers), s3Servers) @@ -92,7 +92,7 @@ func (s *PropagatingCredentialStore) propagateChange(ctx context.Context, fn fun return fn(propagateCtx, client) }, server, false, s.grpcDialOption) if err != nil { - glog.V(1).Infof("failed to propagate change to s3 server %s: %v", server, err) + glog.Warningf("failed to propagate change to s3 server %s: %v", server, err) } }(server) } diff --git a/weed/s3api/auth_credentials.go b/weed/s3api/auth_credentials.go index 427036660..c593a4e36 100644 --- a/weed/s3api/auth_credentials.go +++ b/weed/s3api/auth_credentials.go @@ -90,6 +90,17 @@ type IdentityAccessManagement struct { // staticIdentityNames tracks identity names loaded from the static config file // These identities are immutable and cannot be updated by dynamic configuration staticIdentityNames map[string]bool + + // staticPolicyNames tracks policy names loaded from the static config file + // so full-state reconciliation does not drop them + staticPolicyNames map[string]bool + + // 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 sync.Mutex } type Identity struct { @@ -267,6 +278,8 @@ func NewIdentityAccessManagementWithStore(option *S3ApiServerOption, filerClient iam.credentialManager = credentialManager iam.stopChan = make(chan struct{}) + iam.reloadCh = make(chan struct{}, 1) + go iam.reloadRetryLoop() iam.grpcDialOption = option.GrpcDialOption // First, try to load configurations from file or filer @@ -350,6 +363,12 @@ func (iam *IdentityAccessManagement) markStaticIdentities(config *iam_pb.S3ApiCo for _, ident := range config.Identities { iam.staticIdentityNames[ident.Name] = true } + if iam.staticPolicyNames == nil { + iam.staticPolicyNames = make(map[string]bool) + } + for _, policy := range config.Policies { + iam.staticPolicyNames[policy.Name] = true + } for _, identity := range iam.identities { if iam.staticIdentityNames[identity.Name] { identity.IsStatic = true @@ -358,6 +377,40 @@ func (iam *IdentityAccessManagement) markStaticIdentities(config *iam_pb.S3ApiCo iam.useStaticConfig = len(iam.staticIdentityNames) > 0 } +var iamReloadRetryInterval = 5 * time.Second + +// scheduleReload queues a full configuration reload that retries until it +// succeeds. Signals coalesce, and the reload is state-based, so it is safe to +// call for every failed event. +func (iam *IdentityAccessManagement) scheduleReload() { + select { + case iam.reloadCh <- struct{}{}: + default: + } +} + +func (iam *IdentityAccessManagement) reloadRetryLoop() { + for { + select { + case <-iam.stopChan: + return + case <-iam.reloadCh: + } + for { + err := iam.LoadS3ApiConfigurationFromCredentialManager() + if err == nil || errors.Is(err, filer_pb.ErrNotFound) { + break + } + glog.Warningf("retrying IAM reload: %v", err) + select { + case <-iam.stopChan: + return + case <-time.After(iamReloadRetryInterval): + } + } + } +} + func (iam *IdentityAccessManagement) pollIamConfigChanges(interval time.Duration) { ticker := time.NewTicker(interval) defer ticker.Stop() @@ -492,6 +545,8 @@ func (iam *IdentityAccessManagement) doLoadS3ApiConfigurationFromFiler(option *S } func (iam *IdentityAccessManagement) loadS3ApiConfigurationFromFile(fileName string) error { + iam.reloadMu.Lock() + defer iam.reloadMu.Unlock() content, readErr := os.ReadFile(fileName) if readErr != nil { glog.Warningf("fail to read %s : %v", fileName, readErr) @@ -531,6 +586,11 @@ func (iam *IdentityAccessManagement) loadS3ApiConfigurationFromBytes(content []b return nil, fmt.Errorf("unmarshal error: %w", err) } + if fromStaticFile && len(s3ApiConfiguration.Groups) > 0 { + glog.Warningf("ignoring %d groups in static config file: groups are managed via the IAM API", len(s3ApiConfiguration.Groups)) + s3ApiConfiguration.Groups = nil + } + if err := filer.CheckDuplicateAccessKey(s3ApiConfiguration); err != nil { return nil, err } @@ -554,8 +614,8 @@ func (iam *IdentityAccessManagement) loadS3ApiConfigurationWithSource(config *ia iam.m.RUnlock() if hasStaticConfig { - // Merge mode: preserve static identities, add/update dynamic ones - return iam.MergeS3ApiConfiguration(config, fromStaticFile) + // Merge mode: a dynamic load is the full store state, so it also reconciles deletions + return iam.MergeS3ApiConfiguration(config, fromStaticFile, !fromStaticFile) } // Normal mode: completely replace configuration @@ -776,7 +836,9 @@ func (iam *IdentityAccessManagement) ReplaceS3ApiConfiguration(config *iam_pb.S3 // MergeS3ApiConfiguration adds/updates dynamic identities while preserving static // ones. A config-file reload (fromStaticFile) may also overwrite its static identities. -func (iam *IdentityAccessManagement) MergeS3ApiConfiguration(config *iam_pb.S3ApiConfiguration, fromStaticFile bool) error { +// isFullState marks config as the complete store snapshot, so dynamic identities +// absent from it are removed; partial merges (a single pushed identity) pass false. +func (iam *IdentityAccessManagement) MergeS3ApiConfiguration(config *iam_pb.S3ApiConfiguration, fromStaticFile bool, isFullState bool) error { // Start with current configuration (which includes static identities) iam.m.RLock() identities := make([]*Identity, len(iam.identities)) @@ -806,6 +868,10 @@ func (iam *IdentityAccessManagement) MergeS3ApiConfiguration(config *iam_pb.S3Ap for k, v := range iam.staticIdentityNames { staticNames[k] = v } + staticPolicies := make(map[string]bool) + for k, v := range iam.staticPolicyNames { + staticPolicies[k] = v + } iam.m.RUnlock() // Process accounts from dynamic config (can add new accounts) @@ -926,6 +992,31 @@ func (iam *IdentityAccessManagement) MergeS3ApiConfiguration(config *iam_pb.S3Ap nameToIdentity[t.Name] = t } + // full snapshot: drop dynamic identities the store no longer has + if isFullState { + present := make(map[string]bool, len(config.Identities)) + for _, ident := range config.Identities { + present[ident.Name] = true + } + kept := identities[:0] + for _, existing := range identities { + if staticNames[existing.Name] || present[existing.Name] { + kept = append(kept, existing) + continue + } + delete(nameToIdentity, existing.Name) + for _, cred := range existing.Credentials { + if accessKeyIdent[cred.AccessKey] == existing { + delete(accessKeyIdent, cred.AccessKey) + } + } + if identityAnonymous == existing { + identityAnonymous = nil + } + } + identities = kept + } + // Process service accounts from dynamic config for _, sa := range config.ServiceAccounts { if sa.Credential == nil { @@ -979,38 +1070,21 @@ 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) } - // If the anonymous identity was carried over from the previous state but is - // no longer present in the credential-manager snapshot, clear it so that - // deleted anonymous users do not persist across merges. - if identityAnonymous != nil && !identityAnonymous.IsStatic { - stillPresent := false - for _, ident := range config.Identities { - if ident.Name == s3_constants.AccountAnonymousId { - stillPresent = true - break - } - } - if !stillPresent { - // Remove from identities slice and maps - for i, ident := range identities { - if ident == identityAnonymous { - identities = append(identities[:i], identities[i+1:]...) - break - } - } - delete(nameToIdentity, identityAnonymous.Name) - for _, cred := range identityAnonymous.Credentials { - if accessKeyIdent[cred.AccessKey] == identityAnonymous { - delete(accessKeyIdent, cred.AccessKey) - } - } - identityAnonymous = nil - } - } - 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 @@ -1022,9 +1096,12 @@ func (iam *IdentityAccessManagement) MergeS3ApiConfiguration(config *iam_pb.S3Ap iam.accessKeyIdent = accessKeyIdent iam.policies = policies - // Process groups: only replace if config.Groups is non-nil (full config reload). - // Partial updates (e.g., UpsertIdentity) pass nil Groups and should preserve existing state. - if config.Groups != nil { + // Groups: a full snapshot is authoritative even when empty (last group + // deleted); partial updates pass nil Groups and preserve existing state. + // Groups: a full snapshot is authoritative, even when empty (last group + // deleted). Partial updates pass nil Groups and preserve existing state; + // static config files never carry groups (stripped at load). + if isFullState || config.Groups != nil { mergedGroups := make(map[string]*iam_pb.Group) mergedUserGroups := make(map[string][]string) for _, g := range config.Groups { @@ -1115,7 +1192,7 @@ func (iam *IdentityAccessManagement) UpsertIdentity(ident *iam_pb.Identity) erro glog.V(1).Infof("IAM: upsert identity %s", ident.Name) return iam.MergeS3ApiConfiguration(&iam_pb.S3ApiConfiguration{ Identities: []*iam_pb.Identity{ident}, - }, false) + }, false, false) } // isEnabled reports whether S3 auth should be enforced for this server. @@ -2078,6 +2155,8 @@ func actionScopedToBucket(action, bucket string) bool { // LoadS3ApiConfigurationFromCredentialManager loads configuration using the credential manager func (iam *IdentityAccessManagement) LoadS3ApiConfigurationFromCredentialManager() error { + iam.reloadMu.Lock() + defer iam.reloadMu.Unlock() glog.V(1).Infof("Loading S3 API configuration from credential manager") s3ApiConfiguration, err := iam.credentialManager.LoadConfiguration(context.Background()) diff --git a/weed/s3api/auth_credentials_static_config_test.go b/weed/s3api/auth_credentials_static_config_test.go index 7f4e8d71a..70c791a80 100644 --- a/weed/s3api/auth_credentials_static_config_test.go +++ b/weed/s3api/auth_credentials_static_config_test.go @@ -2,10 +2,13 @@ package s3api import ( "context" + "fmt" "os" "path/filepath" "testing" + "time" + "github.com/seaweedfs/seaweedfs/weed/credential" _ "github.com/seaweedfs/seaweedfs/weed/credential/memory" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -40,9 +43,9 @@ func TestIamConfigWithoutIdentitiesIsNotStatic(t *testing.T) { } } -// A -config identity file marks its identities static, protecting them and keeping the -// established behavior of not live-reloading those from the filer. -func TestConfigWithIdentitiesIsStatic(t *testing.T) { +// A -config identity file protects its identities but must not block live +// delivery of filer-managed identities to a running gateway. +func TestConfigWithIdentitiesStillLiveReloadsDynamic(t *testing.T) { s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{}) path := writeTempIamConfig(t, `{"identities":[{"name":"static-admin","credentials":[{"accessKey":"AKIAITEST","secretKey":"c2VjcmV0"}],"actions":["Admin"]}]}`) @@ -61,15 +64,60 @@ func TestConfigWithIdentitiesIsStatic(t *testing.T) { t.Fatalf("expected static-admin to be marked static") } - // A static identity file does not live-reload dynamic identities from the filer. if err := s3a.iam.credentialManager.CreateUser(context.Background(), &iam_pb.Identity{Name: "alice"}); err != nil { t.Fatalf("failed to create alice: %v", err) } if err := s3a.onIamConfigChange(filer.IamConfigDirectory+"/identities", nil, &filer_pb.Entry{Name: "alice.json"}); err != nil { t.Fatalf("onIamConfigChange returned error: %v", err) } + if !hasIdentity(s3a.iam, "alice") { + t.Fatalf("expected alice to live-reload despite the static identity file") + } + if !hasIdentity(s3a.iam, "static-admin") { + t.Fatalf("static-admin must survive the dynamic reload") + } + + // deletion on the filer must revoke on the running gateway too + if err := s3a.iam.credentialManager.DeleteUser(context.Background(), "alice"); err != nil { + t.Fatalf("failed to delete alice: %v", err) + } + if err := s3a.onIamConfigChange(filer.IamConfigDirectory+"/identities", &filer_pb.Entry{Name: "alice.json"}, nil); err != nil { + t.Fatalf("onIamConfigChange returned error: %v", err) + } if hasIdentity(s3a.iam, "alice") { - t.Fatalf("did not expect alice to load while running off a static identity file") + t.Fatalf("expected alice to be removed after deletion on the filer") + } + if !hasIdentity(s3a.iam, "static-admin") { + t.Fatalf("static-admin must survive the deletion reload") + } +} + +// A single pushed identity (PutIdentity) is a partial merge and must not wipe +// other dynamic identities or a dynamic anonymous identity. +func TestUpsertIdentityKeepsOtherDynamicIdentities(t *testing.T) { + s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{}) + + path := writeTempIamConfig(t, `{"identities":[{"name":"static-admin","credentials":[{"accessKey":"AKIAITEST","secretKey":"c2VjcmV0"}],"actions":["Admin"]}]}`) + if err := s3a.iam.loadS3ApiConfigurationFromFile(path); err != nil { + t.Fatalf("failed to load identity config: %v", err) + } + + for _, name := range []string{"alice", "anonymous"} { + if err := s3a.iam.credentialManager.CreateUser(context.Background(), &iam_pb.Identity{Name: name}); err != nil { + t.Fatalf("failed to create %s: %v", name, err) + } + } + if err := s3a.iam.LoadS3ApiConfigurationFromCredentialManager(); err != nil { + t.Fatalf("failed to load from credential manager: %v", err) + } + + if err := s3a.iam.UpsertIdentity(&iam_pb.Identity{Name: "bob", Actions: []string{"Read"}}); err != nil { + t.Fatalf("failed to upsert bob: %v", err) + } + for _, name := range []string{"static-admin", "alice", "anonymous", "bob"} { + if !hasIdentity(s3a.iam, name) { + t.Fatalf("expected %s to survive a partial upsert", name) + } } } @@ -167,6 +215,120 @@ func TestReloadStaticConfigUpdatesServiceAccountSecret(t *testing.T) { } } +// A full snapshot reconciles policy and group deletions, static-file policies +// survive, and groups in a static config file are ignored: the dynamic store +// is the only source of groups. +func TestFullStateMergeReconcilesPoliciesAndGroups(t *testing.T) { + s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{}) + + path := writeTempIamConfig(t, `{"identities":[{"name":"static-admin","credentials":[{"accessKey":"AKIAITEST","secretKey":"c2VjcmV0"}],"actions":["Admin"]}],"policies":[{"name":"file-policy","content":"{}"}],"groups":[{"name":"file-group","members":["static-admin"]}]}`) + if err := s3a.iam.loadS3ApiConfigurationFromFile(path); err != nil { + t.Fatalf("failed to load identity config: %v", err) + } + if hasGroup(s3a.iam, "file-group") { + t.Fatalf("groups in a static config file must be ignored") + } + + full := &iam_pb.S3ApiConfiguration{ + Policies: []*iam_pb.Policy{{Name: "dynamic-policy", Content: "{}"}}, + Groups: []*iam_pb.Group{{Name: "g1"}}, + } + if err := s3a.iam.MergeS3ApiConfiguration(full, false, true); err != nil { + t.Fatalf("full merge failed: %v", err) + } + if !hasPolicy(s3a.iam, "file-policy") || !hasPolicy(s3a.iam, "dynamic-policy") || !hasGroup(s3a.iam, "g1") { + t.Fatalf("expected file-policy, dynamic-policy and g1 after full merge") + } + + // partial merge preserves groups and policies + if err := s3a.iam.UpsertIdentity(&iam_pb.Identity{Name: "bob"}); err != nil { + t.Fatalf("upsert failed: %v", err) + } + if !hasPolicy(s3a.iam, "dynamic-policy") || !hasGroup(s3a.iam, "g1") { + t.Fatalf("partial merge must not drop dynamic-policy or g1") + } + + // a static-file reload leaves dynamic groups alone + if err := s3a.iam.loadS3ApiConfigurationFromFile(path); err != nil { + t.Fatalf("failed to reload config: %v", err) + } + if !hasGroup(s3a.iam, "g1") { + t.Fatalf("dynamic g1 must survive a static-file reload") + } + + // empty full snapshot: dynamic policy and last group deleted, file policy stays + if err := s3a.iam.MergeS3ApiConfiguration(&iam_pb.S3ApiConfiguration{}, false, true); err != nil { + t.Fatalf("empty full merge failed: %v", err) + } + if hasPolicy(s3a.iam, "dynamic-policy") { + t.Fatalf("expected dynamic-policy to be removed by empty full snapshot") + } + if hasGroup(s3a.iam, "g1") { + t.Fatalf("expected g1 to be removed by empty full snapshot") + } + if !hasPolicy(s3a.iam, "file-policy") { + t.Fatalf("file-policy must survive full-state reconciliation") + } +} + +// flakyStore fails LoadConfiguration a fixed number of times. +type flakyStore struct { + credential.CredentialStore + failures int +} + +func (f *flakyStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3ApiConfiguration, error) { + if f.failures > 0 { + f.failures-- + return nil, fmt.Errorf("transient store failure") + } + return f.CredentialStore.LoadConfiguration(ctx) +} + +// A failed event-driven reload must keep retrying until the store recovers. +func TestFailedReloadRetriesUntilSuccess(t *testing.T) { + s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{}) + + prev := iamReloadRetryInterval + iamReloadRetryInterval = 10 * time.Millisecond + t.Cleanup(func() { iamReloadRetryInterval = prev }) + + s3a.iam.reloadCh = make(chan struct{}, 1) + go s3a.iam.reloadRetryLoop() + t.Cleanup(s3a.iam.Shutdown) + + if err := s3a.iam.credentialManager.CreateUser(context.Background(), &iam_pb.Identity{Name: "alice"}); err != nil { + t.Fatalf("failed to create alice: %v", err) + } + s3a.iam.credentialManager.Store = &flakyStore{CredentialStore: s3a.iam.credentialManager.Store, failures: 2} + + // the event-driven reload fails and hands off to the retry loop + if err := s3a.onIamConfigChange(filer.IamConfigDirectory+"/identities", nil, &filer_pb.Entry{Name: "alice.json"}); err == nil { + t.Fatalf("expected the event-driven reload to fail") + } + deadline := time.Now().Add(5 * time.Second) + for !hasIdentity(s3a.iam, "alice") { + if time.Now().After(deadline) { + t.Fatalf("expected alice to load once the store recovered") + } + time.Sleep(10 * time.Millisecond) + } +} + +func hasPolicy(iam *IdentityAccessManagement, name string) bool { + iam.m.RLock() + defer iam.m.RUnlock() + _, ok := iam.policies[name] + return ok +} + +func hasGroup(iam *IdentityAccessManagement, name string) bool { + iam.m.RLock() + defer iam.m.RUnlock() + _, ok := iam.groups[name] + return ok +} + func isStaticName(iam *IdentityAccessManagement, name string) bool { iam.m.RLock() defer iam.m.RUnlock() diff --git a/weed/s3api/auth_credentials_subscribe.go b/weed/s3api/auth_credentials_subscribe.go index 38657a3ed..2f6714839 100644 --- a/weed/s3api/auth_credentials_subscribe.go +++ b/weed/s3api/auth_credentials_subscribe.go @@ -76,12 +76,10 @@ func (s3a *S3ApiServer) subscribeMetaEvents(clientName string, lastTsNs int64, p }) } -// onIamConfigChange handles IAM config file changes (create, update, delete) +// onIamConfigChange handles IAM config file changes (create, update, delete). +// It reloads even with a static -config file: the merge protects the file's +// identities, and the filer->s3 push alone is best-effort. func (s3a *S3ApiServer) onIamConfigChange(dir string, oldEntry *filer_pb.Entry, newEntry *filer_pb.Entry) error { - if s3a.iam != nil && s3a.iam.IsStaticConfig() { - glog.V(1).Infof("Skipping IAM config update for static configuration") - return nil - } if s3a.iam == nil { return nil } @@ -89,7 +87,9 @@ func (s3a *S3ApiServer) onIamConfigChange(dir string, oldEntry *filer_pb.Entry, reloadIamConfig := func(reason string) error { glog.V(1).Infof("IAM change detected in %s, reloading configuration", reason) if err := s3a.iam.LoadS3ApiConfigurationFromCredentialManager(); err != nil { + // the event stream moves on; retry state-based until a reload succeeds glog.Errorf("failed to reload IAM configuration after change in %s: %v", reason, err) + s3a.iam.scheduleReload() return err } return nil