diff --git a/weed/credential/filer_etc/filer_etc_group.go b/weed/credential/filer_etc/filer_etc_group.go index 36a5d4fb4..4d78cd5fe 100644 --- a/weed/credential/filer_etc/filer_etc_group.go +++ b/weed/credential/filer_etc/filer_etc_group.go @@ -30,6 +30,9 @@ func (store *FilerEtcStore) loadGroupsFromMultiFile(ctx context.Context, s3cfg * if entry.IsDirectory { continue } + if !strings.HasSuffix(entry.Name, ".json") { + continue + } var content []byte if len(entry.Content) > 0 { @@ -42,24 +45,28 @@ func (store *FilerEtcStore) loadGroupsFromMultiFile(ctx context.Context, s3cfg * content = c } - if len(content) > 0 { - g := &iam_pb.Group{} - if err := json.Unmarshal(content, g); err != nil { - return fmt.Errorf("failed to unmarshal group %s: %w", entry.Name, err) - } - // Merge: overwrite existing group with same name or append - found := false - for i, existing := range s3cfg.Groups { - if existing.Name == g.Name { - s3cfg.Groups[i] = g - found = true - break - } - } - if !found { - s3cfg.Groups = append(s3cfg.Groups, g) + if len(content) == 0 { + return fmt.Errorf("group file %s is empty", entry.Name) + } + g := &iam_pb.Group{} + if err := json.Unmarshal(content, g); err != nil { + return fmt.Errorf("failed to unmarshal group %s: %w", entry.Name, err) + } + if g.Name == "" { + return fmt.Errorf("group file %s has empty name", entry.Name) + } + // Merge: overwrite existing group with same name or append + found := false + for i, existing := range s3cfg.Groups { + if existing.Name == g.Name { + s3cfg.Groups[i] = g + found = true + break } } + if !found { + s3cfg.Groups = append(s3cfg.Groups, g) + } } return nil }) diff --git a/weed/credential/filer_etc/filer_etc_identity.go b/weed/credential/filer_etc/filer_etc_identity.go index b2a83ec5c..cdf59ca71 100644 --- a/weed/credential/filer_etc/filer_etc_identity.go +++ b/weed/credential/filer_etc/filer_etc_identity.go @@ -92,6 +92,9 @@ func (store *FilerEtcStore) loadFromMultiFile(ctx context.Context, s3cfg *iam_pb if entry.IsDirectory { continue } + if !strings.HasSuffix(entry.Name, ".json") { + continue + } hasIdentities = true var content []byte @@ -100,26 +103,28 @@ func (store *FilerEtcStore) loadFromMultiFile(ctx context.Context, s3cfg *iam_pb } else { c, err := filer.ReadInsideFiler(ctx, client, dir, entry.Name) if err != nil { - // 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 } - if len(content) > 0 { - identity := &iam_pb.Identity{} - if err := json.Unmarshal(content, identity); err != nil { - glog.Warningf("Failed to unmarshal identity %s: %v", entry.Name, err) - continue - } + if len(content) == 0 { + return fmt.Errorf("identity file %s is empty", entry.Name) + } + identity := &iam_pb.Identity{} + if err := json.Unmarshal(content, identity); err != nil { + return fmt.Errorf("failed to unmarshal identity %s: %w", entry.Name, err) + } + if identity.Name == "" { + return fmt.Errorf("identity file %s has empty name", entry.Name) + } - // Merge logic: Overwrite existing or Append - idx := findIdentity(identity.Name) - if idx != -1 { - s3cfg.Identities[idx] = identity - } else { - s3cfg.Identities = append(s3cfg.Identities, identity) - } + // Merge logic: Overwrite existing or Append + idx := findIdentity(identity.Name) + if idx != -1 { + s3cfg.Identities[idx] = identity + } else { + s3cfg.Identities = append(s3cfg.Identities, identity) } } return nil diff --git a/weed/credential/filer_etc/filer_etc_identity_test.go b/weed/credential/filer_etc/filer_etc_identity_test.go new file mode 100644 index 000000000..27d182c23 --- /dev/null +++ b/weed/credential/filer_etc/filer_etc_identity_test.go @@ -0,0 +1,144 @@ +package filer_etc + +import ( + "context" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func putEntry(t *testing.T, server *policyTestFilerServer, dir, name string, content []byte) { + t.Helper() + _, err := server.CreateEntry(context.Background(), &filer_pb.CreateEntryRequest{ + Directory: dir, + Entry: &filer_pb.Entry{Name: name, Content: content}, + }) + require.NoError(t, err) +} + +// A mid-rewrite identity file (empty content, as seen while a secrets tool +// truncates and rewrites the file) must fail the snapshot instead of being +// silently dropped. Silently dropping it would make a full reload install an +// incomplete identity set and deny unrelated clients mid-reload. +func TestLoadConfigurationFailsOnEmptyIdentityFile(t *testing.T) { + ctx := context.Background() + store, server := newPolicyTestStoreWithServer(t) + + identDir := filer.IamConfigDirectory + "/" + IamIdentitiesDirectory + putEntry(t, server, identDir, "alice.json", []byte(`{"name":"alice","credentials":[{"accessKey":"AK","secretKey":"SK"}]}`)) + putEntry(t, server, identDir, "bob.json", []byte{}) // mid-rewrite: empty + + _, err := store.LoadConfiguration(ctx) + require.Error(t, err) + assert.Contains(t, err.Error(), "bob.json") +} + +// A malformed identity file (partial JSON from a mid-rewrite) must also fail +// the snapshot rather than being skipped. +func TestLoadConfigurationFailsOnMalformedIdentityFile(t *testing.T) { + ctx := context.Background() + store, server := newPolicyTestStoreWithServer(t) + + identDir := filer.IamConfigDirectory + "/" + IamIdentitiesDirectory + putEntry(t, server, identDir, "alice.json", []byte(`{"name":"alice"}`)) + putEntry(t, server, identDir, "bob.json", []byte(`{"name":"bob"`)) // truncated + + _, err := store.LoadConfiguration(ctx) + require.Error(t, err) + assert.Contains(t, err.Error(), "bob.json") +} + +// A mid-rewrite policy file must fail the snapshot rather than be dropped. +func TestLoadManagedPoliciesFailsOnEmptyPolicyFile(t *testing.T) { + ctx := context.Background() + store, server := newPolicyTestStoreWithServer(t) + + polDir := filer.IamConfigDirectory + "/" + IamPoliciesDirectory + putEntry(t, server, polDir, "good.json", []byte(`{"Version":"2012-10-17","Statement":[]}`)) + putEntry(t, server, polDir, "bad.json", []byte{}) // mid-rewrite: empty + + _, err := store.LoadManagedPolicies(ctx) + require.Error(t, err) + assert.Contains(t, err.Error(), "bad.json") +} + +// A non-JSON auxiliary file (README, .DS_Store, a migration backup) in an IAM +// directory must not fail the snapshot: only *.json files are IAM objects. +func TestLoadConfigurationIgnoresNonJsonAuxiliaryFiles(t *testing.T) { + ctx := context.Background() + store, server := newPolicyTestStoreWithServer(t) + + identDir := filer.IamConfigDirectory + "/" + IamIdentitiesDirectory + putEntry(t, server, identDir, "alice.json", []byte(`{"name":"alice","credentials":[{"accessKey":"AK","secretKey":"SK"}]}`)) + putEntry(t, server, identDir, "README.txt", []byte(`not an identity`)) + putEntry(t, server, identDir, "alice.json.old", []byte(`{"name":"alice"}`)) + + polDir := filer.IamConfigDirectory + "/" + IamPoliciesDirectory + putEntry(t, server, polDir, "p.json", []byte(`{"Version":"2012-10-17","Statement":[]}`)) + putEntry(t, server, polDir, "notes.md", []byte(`# policies`)) + + cfg, err := store.LoadConfiguration(ctx) + require.NoError(t, err) + require.Len(t, cfg.Identities, 1) + assert.Equal(t, "alice", cfg.Identities[0].Name) +} + +// A valid JSON file with an empty name (e.g. `{}`) would unmarshal cleanly but +// install a garbage empty-key record that can displace a real one. Reject it. +func TestLoadConfigurationFailsOnEmptyIdentityName(t *testing.T) { + ctx := context.Background() + store, server := newPolicyTestStoreWithServer(t) + + identDir := filer.IamConfigDirectory + "/" + IamIdentitiesDirectory + putEntry(t, server, identDir, "alice.json", []byte(`{"name":"alice"}`)) + putEntry(t, server, identDir, "empty.json", []byte(`{}`)) + + _, err := store.LoadConfiguration(ctx) + require.Error(t, err) + assert.Contains(t, err.Error(), "empty.json") +} + +func TestLoadConfigurationFailsOnEmptyGroupName(t *testing.T) { + ctx := context.Background() + store, server := newPolicyTestStoreWithServer(t) + + groupDir := filer.IamConfigDirectory + "/" + IamGroupsDirectory + putEntry(t, server, groupDir, "empty.json", []byte(`{}`)) + + _, err := store.LoadConfiguration(ctx) + require.Error(t, err) + assert.Contains(t, err.Error(), "empty.json") +} + +func TestLoadConfigurationFailsOnEmptyServiceAccountId(t *testing.T) { + ctx := context.Background() + store, server := newPolicyTestStoreWithServer(t) + + saDir := filer.IamConfigDirectory + "/" + IamServiceAccountsDirectory + putEntry(t, server, saDir, "empty.json", []byte(`{}`)) + + _, err := store.LoadConfiguration(ctx) + require.Error(t, err) + assert.Contains(t, err.Error(), "empty.json") +} + +// A valid full snapshot loads without error (regression guard). +func TestLoadConfigurationSucceedsOnValidFiles(t *testing.T) { + ctx := context.Background() + store, server := newPolicyTestStoreWithServer(t) + + identDir := filer.IamConfigDirectory + "/" + IamIdentitiesDirectory + putEntry(t, server, identDir, "alice.json", []byte(`{"name":"alice","credentials":[{"accessKey":"AK","secretKey":"SK"}]}`)) + putEntry(t, server, identDir, "bob.json", []byte(`{"name":"bob","credentials":[{"accessKey":"BK","secretKey":"SK"}]}`)) + + cfg, err := store.LoadConfiguration(ctx) + require.NoError(t, err) + names := make([]string, 0, len(cfg.Identities)) + for _, id := range cfg.Identities { + names = append(names, id.Name) + } + assert.ElementsMatch(t, []string{"alice", "bob"}, names) +} diff --git a/weed/credential/filer_etc/filer_etc_policy.go b/weed/credential/filer_etc/filer_etc_policy.go index 00dbe3540..d4544b749 100644 --- a/weed/credential/filer_etc/filer_etc_policy.go +++ b/weed/credential/filer_etc/filer_etc_policy.go @@ -181,6 +181,9 @@ func (store *FilerEtcStore) loadPoliciesFromMultiFile(ctx context.Context, polic if entry.IsDirectory { continue } + if !strings.HasSuffix(entry.Name, ".json") { + continue + } var content []byte if len(entry.Content) > 0 { @@ -188,26 +191,21 @@ func (store *FilerEtcStore) loadPoliciesFromMultiFile(ctx context.Context, polic } else { c, err := filer.ReadInsideFiler(ctx, client, dir, entry.Name) if err != nil { - // 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 } - if len(content) > 0 { - var policy policy_engine.PolicyDocument - if err := json.Unmarshal(content, &policy); err != nil { - glog.Warningf("Failed to unmarshal policy %s: %v", entry.Name, err) - continue - } - - // The file name is "policyName.json" - policyName := entry.Name - if strings.HasSuffix(policyName, ".json") { - policyName = policyName[:len(policyName)-5] - policies[policyName] = policy - } + if len(content) == 0 { + return fmt.Errorf("policy file %s is empty", entry.Name) } + var policy policy_engine.PolicyDocument + if err := json.Unmarshal(content, &policy); err != nil { + return fmt.Errorf("failed to unmarshal policy %s: %w", entry.Name, err) + } + + policyName := strings.TrimSuffix(entry.Name, ".json") + policies[policyName] = policy } return nil }) @@ -498,10 +496,10 @@ func (store *FilerEtcStore) ListPolicyNames(ctx context.Context) ([]string, erro if entry.IsDirectory { continue } - name := entry.Name - if strings.HasSuffix(name, ".json") { - name = name[:len(name)-5] + if !strings.HasSuffix(entry.Name, ".json") { + continue } + name := entry.Name[:len(entry.Name)-5] if _, found := seenNames[name]; found { continue } diff --git a/weed/credential/filer_etc/filer_etc_policy_test.go b/weed/credential/filer_etc/filer_etc_policy_test.go index d64cb5fd5..47b034ebc 100644 --- a/weed/credential/filer_etc/filer_etc_policy_test.go +++ b/weed/credential/filer_etc/filer_etc_policy_test.go @@ -192,6 +192,22 @@ func TestFilerEtcStoreListPolicyNamesIncludesLegacyPolicies(t *testing.T) { assert.ElementsMatch(t, []string{"legacy-only", "multi-file-only", "shared"}, names) } +// A non-JSON auxiliary file in the policies directory must not be listed as a +// policy name (it cannot be retrieved by GetPolicy either). +func TestFilerEtcStoreListPolicyNamesSkipsNonJsonAuxiliary(t *testing.T) { + ctx := context.Background() + store, server := newPolicyTestStoreWithServer(t) + + require.NoError(t, store.savePolicy(ctx, "real", testPolicyDocument("s3:GetObject", "arn:aws:s3:::real/*"))) + + polDir := filer.IamConfigDirectory + "/" + IamPoliciesDirectory + putEntry(t, server, polDir, "notes.md", []byte(`# not a policy`)) + + names, err := store.ListPolicyNames(ctx) + require.NoError(t, err) + assert.ElementsMatch(t, []string{"real"}, names) +} + func TestFilerEtcStoreDeletePolicyRemovesLegacyManagedCopy(t *testing.T) { ctx := context.Background() store := newPolicyTestStore(t) diff --git a/weed/credential/filer_etc/filer_etc_service_account.go b/weed/credential/filer_etc/filer_etc_service_account.go index 2f8b3a480..20c6a3843 100644 --- a/weed/credential/filer_etc/filer_etc_service_account.go +++ b/weed/credential/filer_etc/filer_etc_service_account.go @@ -33,6 +33,9 @@ func (store *FilerEtcStore) loadServiceAccountsFromMultiFile(ctx context.Context if entry.IsDirectory { continue } + if !strings.HasSuffix(entry.Name, ".json") { + continue + } var content []byte if len(entry.Content) > 0 { @@ -40,20 +43,22 @@ func (store *FilerEtcStore) loadServiceAccountsFromMultiFile(ctx context.Context } else { c, err := filer.ReadInsideFiler(ctx, client, dir, entry.Name) if err != nil { - glog.Warningf("Failed to read service account file %s: %v", entry.Name, err) - continue + return fmt.Errorf("failed to read service account file %s: %w", entry.Name, err) } content = c } - if len(content) > 0 { - sa := &iam_pb.ServiceAccount{} - if err := json.Unmarshal(content, sa); err != nil { - glog.Warningf("Failed to unmarshal service account %s: %v", entry.Name, err) - continue - } - s3cfg.ServiceAccounts = append(s3cfg.ServiceAccounts, sa) + if len(content) == 0 { + return fmt.Errorf("service account file %s is empty", entry.Name) } + sa := &iam_pb.ServiceAccount{} + if err := json.Unmarshal(content, sa); err != nil { + return fmt.Errorf("failed to unmarshal service account %s: %w", entry.Name, err) + } + if err := validateServiceAccountId(sa.Id); err != nil { + return fmt.Errorf("service account file %s: %w", entry.Name, err) + } + s3cfg.ServiceAccounts = append(s3cfg.ServiceAccounts, sa) } return nil }) diff --git a/weed/s3api/auth_credentials.go b/weed/s3api/auth_credentials.go index 21d0833f4..f1e9d6afb 100644 --- a/weed/s3api/auth_credentials.go +++ b/weed/s3api/auth_credentials.go @@ -455,10 +455,11 @@ func (iam *IdentityAccessManagement) markStaticIdentities(config *iam_pb.S3ApiCo 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() { +// scheduleReload queues a coalesced full configuration reload that retries +// until it succeeds. Safe to call for every IAM config change event: bursts +// collapse into a single reload via the buffered reloadCh. +func (iam *IdentityAccessManagement) scheduleReload(reason string) { + glog.V(1).Infof("IAM change detected in %s, scheduling reload", reason) select { case iam.reloadCh <- struct{}{}: default: diff --git a/weed/s3api/auth_credentials_static_config_test.go b/weed/s3api/auth_credentials_static_config_test.go index fe66efc3c..463ca7448 100644 --- a/weed/s3api/auth_credentials_static_config_test.go +++ b/weed/s3api/auth_credentials_static_config_test.go @@ -38,9 +38,7 @@ func TestIamConfigWithoutIdentitiesIsNotStatic(t *testing.T) { 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 load after filer change with -iam.config-only setup") - } + waitForIdentity(t, s3a.iam, "alice") } // A -config identity file protects its identities but must not block live @@ -70,9 +68,7 @@ func TestConfigWithIdentitiesStillLiveReloadsDynamic(t *testing.T) { 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") - } + waitForIdentity(t, s3a.iam, "alice") if !hasIdentity(s3a.iam, "static-admin") { t.Fatalf("static-admin must survive the dynamic reload") } @@ -84,9 +80,7 @@ func TestConfigWithIdentitiesStillLiveReloadsDynamic(t *testing.T) { 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("expected alice to be removed after deletion on the filer") - } + waitForIdentityGone(t, s3a.iam, "alice") if !hasIdentity(s3a.iam, "static-admin") { t.Fatalf("static-admin must survive the deletion reload") } @@ -285,7 +279,8 @@ func (f *flakyStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3ApiConfig return f.CredentialStore.LoadConfiguration(ctx) } -// A failed event-driven reload must keep retrying until the store recovers. +// A failed reload queued from an IAM config change must keep retrying until +// the store recovers. func TestFailedReloadRetriesUntilSuccess(t *testing.T) { s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{}) @@ -293,26 +288,17 @@ func TestFailedReloadRetriesUntilSuccess(t *testing.T) { 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) + // onIamConfigChange only queues a coalesced reload; the retry loop does the + // work and retries through the transient store failures. + if err := s3a.onIamConfigChange(filer.IamConfigDirectory+"/identities", nil, &filer_pb.Entry{Name: "alice.json"}); err != nil { + t.Fatalf("onIamConfigChange returned error: %v", err) } + waitForIdentity(t, s3a.iam, "alice") } func hasPolicy(iam *IdentityAccessManagement, name string) bool { diff --git a/weed/s3api/auth_credentials_subscribe.go b/weed/s3api/auth_credentials_subscribe.go index 50be9104e..8f2519ae6 100644 --- a/weed/s3api/auth_credentials_subscribe.go +++ b/weed/s3api/auth_credentials_subscribe.go @@ -84,44 +84,27 @@ func (s3a *S3ApiServer) onIamConfigChange(dir string, oldEntry *filer_pb.Entry, return nil } - 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 - } + // Coalesce bursts through the reload queue instead of one full synchronous + // reload per event: independently-refreshing credentials can rewrite several + // files within the same second. // 1. Handle traditional single identity.json file if dir == filer.IamConfigDirectory { - // Handle create/update/delete events on legacy identity.json. - // During migration this file is renamed, which emits a delete event. - // Always reload from the credential manager so we keep the migrated identities. if (oldEntry != nil && oldEntry.Name == filer.IamIdentityFile) || (newEntry != nil && newEntry.Name == filer.IamIdentityFile) { - if err := reloadIamConfig(dir + "/" + filer.IamIdentityFile); err != nil { - return err - } + s3a.iam.scheduleReload(dir + "/" + filer.IamIdentityFile) } return nil } - // 2. Handle multiple-file identities and policies - // Watch /etc/iam/{identities,policies,service_accounts} + // 2. Handle multiple-file identities, policies, service accounts and groups isIdentityDir := dir == filer.IamConfigDirectory+"/identities" || strings.HasPrefix(dir, filer.IamConfigDirectory+"/identities/") isPolicyDir := dir == filer.IamConfigDirectory+"/policies" || strings.HasPrefix(dir, filer.IamConfigDirectory+"/policies/") isServiceAccountDir := dir == filer.IamConfigDirectory+"/service_accounts" || strings.HasPrefix(dir, filer.IamConfigDirectory+"/service_accounts/") isGroupDir := dir == filer.IamConfigDirectory+"/groups" || strings.HasPrefix(dir, filer.IamConfigDirectory+"/groups/") if isIdentityDir || isPolicyDir || isServiceAccountDir || isGroupDir { - // For multiple-file mode, any change in these directories should trigger a full reload - // from the credential manager (which handles the details of loading from multiple files). - if err := reloadIamConfig(dir); err != nil { - return err - } + s3a.iam.scheduleReload(dir) } return nil diff --git a/weed/s3api/auth_credentials_subscribe_test.go b/weed/s3api/auth_credentials_subscribe_test.go index 7ee236594..59520f1e8 100644 --- a/weed/s3api/auth_credentials_subscribe_test.go +++ b/weed/s3api/auth_credentials_subscribe_test.go @@ -2,8 +2,11 @@ package s3api import ( "context" + "fmt" "sync" + "sync/atomic" "testing" + "time" "github.com/seaweedfs/seaweedfs/weed/credential" _ "github.com/seaweedfs/seaweedfs/weed/credential/memory" @@ -53,18 +56,15 @@ func TestOnIamConfigChangeReloadsOnIamIdentityDirectoryChanges(t *testing.T) { t.Fatalf("failed to create alice in memory credential manager: %v", err) } - err := s3a.onIamConfigChange( + if err := s3a.onIamConfigChange( filer.IamConfigDirectory+"/identities", nil, &filer_pb.Entry{Name: "alice.json"}, - ) - if err != nil { + ); err != nil { t.Fatalf("onIamConfigChange returned error for identities directory update: %v", err) } - if !hasIdentity(s3a.iam, "alice") { - t.Fatalf("expected alice identity to be loaded after /etc/iam/identities update") - } + waitForIdentity(t, s3a.iam, "alice") } func newTestS3ApiServerWithMemoryIAM(t *testing.T, identities []*iam_pb.Identity) *S3ApiServer { @@ -101,9 +101,12 @@ func newTestS3ApiServerWithMemoryIAM(t *testing.T, identities []*iam_pb.Identity hashCounters: make(map[string]*int32), isAuthEnabled: false, stopChan: make(chan struct{}), + reloadCh: make(chan struct{}, 1), useStaticConfig: false, credentialManager: cm, } + go iam.reloadRetryLoop() + t.Cleanup(iam.Shutdown) // Load test configuration if err := iam.ReplaceS3ApiConfiguration(config); err != nil { @@ -122,3 +125,78 @@ func hasIdentity(iam *IdentityAccessManagement, identityName string) bool { _, ok := iam.nameToIdentity[identityName] return ok } + +func waitForIdentity(t *testing.T, iam *IdentityAccessManagement, name string) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for !hasIdentity(iam, name) { + if time.Now().After(deadline) { + t.Fatalf("expected identity %s to be loaded", name) + } + time.Sleep(10 * time.Millisecond) + } +} + +func waitForIdentityGone(t *testing.T, iam *IdentityAccessManagement, name string) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for hasIdentity(iam, name) { + if time.Now().After(deadline) { + t.Fatalf("expected identity %s to be gone", name) + } + time.Sleep(10 * time.Millisecond) + } +} + +// countingStore wraps a store and counts LoadConfiguration calls. +type countingStore struct { + credential.CredentialStore + loads int64 +} + +func (c *countingStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3ApiConfiguration, error) { + atomic.AddInt64(&c.loads, 1) + return c.CredentialStore.LoadConfiguration(ctx) +} + +// A burst of IAM config change events must coalesce into a handful of reloads, +// not one full reload per event. +func TestOnIamConfigChangeCoalescesBurstReloads(t *testing.T) { + s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{{Name: "anonymous"}}) + + counter := &countingStore{CredentialStore: s3a.iam.credentialManager.Store} + s3a.iam.credentialManager.Store = counter + + const burst = 50 + for i := 0; i < burst; i++ { + if err := s3a.onIamConfigChange( + filer.IamConfigDirectory+"/identities", + nil, + &filer_pb.Entry{Name: fmt.Sprintf("u%d.json", i)}, + ); err != nil { + t.Fatalf("onIamConfigChange returned error: %v", err) + } + } + + // Wait for the queue to drain: no pending signal. + deadline := time.Now().Add(5 * time.Second) + for { + s3a.iam.reloadMu.Lock() + empty := len(s3a.iam.reloadCh) == 0 + s3a.iam.reloadMu.Unlock() + if empty { + break + } + if time.Now().After(deadline) { + t.Fatalf("reload queue did not drain") + } + time.Sleep(10 * time.Millisecond) + } + // Let any final coalesced reload finish. + time.Sleep(50 * time.Millisecond) + + loads := atomic.LoadInt64(&counter.loads) + if loads > 3 { + t.Fatalf("expected a burst of %d events to coalesce into <=3 reloads, got %d", burst, loads) + } +}