credential: keep IAM changes made during the legacy-file migration (#12614)

* credential: migrate the legacy IAM file without overwriting newer entries

The migration wrote every identity, service account and group from the
snapshot the load read, over whatever the per-entry files held by then, so
a revoke or delete landing in between was undone for good. It now migrates
only the legacy file's own entries, creates each per-entry file only when
none exists, and waits for the next load when the legacy file changed
after the snapshot.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* credential: test changes landing between the IAM load and its migration

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* credential: keep a per-entry file deleted after the IAM load deleted

An entry the load read from its own file is already migrated. When that
file is removed before the migration runs, the legacy copy must not
recreate it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* credential: test a deleted per-entry file the legacy file also holds

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/credential/filer_etc/filer_etc_identity.go

@@ -1,10 +1,14 @@
 package filer_etc
 
 import (
+	"bytes"
 	"context"
+	"crypto/md5"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/credential"
 	"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -35,6 +39,17 @@ func (store *FilerEtcStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3Ap
 		}
 	}
 
+	fromLegacy := map[any]bool{}
+	for _, identity := range s3cfg.Identities {
+		fromLegacy[identity] = true
+	}
+	for _, sa := range s3cfg.ServiceAccounts {
+		fromLegacy[sa] = true
+	}
+	for _, g := range s3cfg.Groups {
+		fromLegacy[g] = true
+	}
+
 	// 2. Load from multi-file structure (high priority, overrides legacy details)
 	if _, err := store.loadFromMultiFile(ctx, s3cfg); err != nil {
 		return s3cfg, err
@@ -54,7 +69,23 @@ func (store *FilerEtcStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3Ap
 	// This ensures that all identities (including legacy ones) are written to individual files
 	// and the legacy file is renamed.
 	if foundLegacy {
-		if err := store.migrateToMultiFile(ctx, s3cfg); err != nil {
+		loaded := map[string]bool{}
+		for _, identity := range s3cfg.Identities {
+			if !fromLegacy[identity] {
+				loaded[IamIdentitiesDirectory+"/"+identity.Name] = true
+			}
+		}
+		for _, sa := range s3cfg.ServiceAccounts {
+			if !fromLegacy[sa] {
+				loaded[IamServiceAccountsDirectory+"/"+sa.Id] = true
+			}
+		}
+		for _, g := range s3cfg.Groups {
+			if !fromLegacy[g] {
+				loaded[IamGroupsDirectory+"/"+g.Name] = true
+			}
+		}
+		if err := store.migrateToMultiFile(ctx, content, loaded); err != nil {
 			glog.Errorf("Failed to migrate IAM configuration to multi-file layout: %v", err)
 			return s3cfg, nil
 		}
@@ -137,26 +168,49 @@ func (store *FilerEtcStore) loadFromMultiFile(ctx context.Context, s3cfg *iam_pb
 	return hasIdentities, nil
 }
 
-func (store *FilerEtcStore) migrateToMultiFile(ctx context.Context, s3cfg *iam_pb.S3ApiConfiguration) error {
+func (store *FilerEtcStore) migrateToMultiFile(ctx context.Context, legacyContent []byte, loaded map[string]bool) error {
+	current, found, err := store.readInsideFiler(ctx, filer.IamConfigDirectory, IamLegacyIdentityFile)
+	if err != nil {
+		return err
+	}
+	if !found || !bytes.Equal(current, legacyContent) {
+		glog.V(0).Infof("%s changed during the IAM load, migrating it at the next load", IamLegacyIdentityFile)
+		return nil
+	}
+	legacy := &iam_pb.S3ApiConfiguration{}
+	if err := filer.ParseS3ConfigurationFromBytes(current, legacy); err != nil {
+		return err
+	}
+
 	glog.Infof("Migrating IAM configuration to multi-file layout...")
 
-	// 1. Save all identities
-	for _, identity := range s3cfg.Identities {
-		if err := store.saveIdentity(ctx, identity); err != nil {
+	// A per-entry file is newer than the legacy copy, and may have been
+	// rewritten or removed after this load read it: never write it back.
+	createIfAbsent := func(subDir, name string, v any) error {
+		if loaded[subDir+"/"+name] {
+			return nil
+		}
+		return store.createIfAbsent(ctx, subDir, name, v)
+	}
+	for _, identity := range legacy.Identities {
+		if err := createIfAbsent(IamIdentitiesDirectory, identity.Name, identity); err != nil {
 			return err
 		}
 	}
-
-	// 2. Save all service accounts
-	for _, sa := range s3cfg.ServiceAccounts {
-		if err := store.saveServiceAccount(ctx, sa); err != nil {
+	for _, sa := range legacy.ServiceAccounts {
+		if err := validateServiceAccountId(sa.Id); err != nil {
+			return err
+		}
+		if err := createIfAbsent(IamServiceAccountsDirectory, sa.Id, sa); err != nil {
 			return err
 		}
 	}
-
-	// 3. Save all groups
-	for _, g := range s3cfg.Groups {
-		if err := store.saveGroup(ctx, g); err != nil {
+	for _, g := range legacy.Groups {
+		g.Name = strings.TrimSpace(g.Name)
+		if g.Name == "" {
+			return fmt.Errorf("group name is required")
+		}
+		if err := createIfAbsent(IamGroupsDirectory, g.Name, g); err != nil {
 			return err
 		}
 	}
@@ -487,6 +541,35 @@ func (store *FilerEtcStore) saveIdentity(ctx context.Context, identity *iam_pb.I
 	})
 }
 
+func (store *FilerEtcStore) createIfAbsent(ctx context.Context, subDir, name string, v any) error {
+	data, err := json.MarshalIndent(v, "", "  ")
+	if err != nil {
+		return err
+	}
+	contentMd5 := md5.Sum(data)
+	return store.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+		err := filer_pb.CreateEntry(ctx, client, &filer_pb.CreateEntryRequest{
+			Directory: filer.IamConfigDirectory + "/" + subDir,
+			Entry: &filer_pb.Entry{
+				Name:    name + ".json",
+				Content: data,
+				Attributes: &filer_pb.FuseAttributes{
+					Mtime:    time.Now().Unix(),
+					Crtime:   time.Now().Unix(),
+					FileMode: uint32(0644),
+					FileSize: uint64(len(data)),
+					Md5:      contentMd5[:],
+				},
+			},
+			OExcl: true,
+		})
+		if errors.Is(err, filer_pb.ErrEntryAlreadyExists) {
+			return nil
+		}
+		return err
+	})
+}
+
 func (store *FilerEtcStore) readInsideFiler(ctx context.Context, dir string, name string) ([]byte, bool, error) {
 	var content []byte
 	found := false

weed/credential/filer_etc/filer_etc_identity_test.go

@@ -2,10 +2,13 @@ package filer_etc
 
 import (
 	"context"
+	"encoding/json"
+	"sync/atomic"
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -142,3 +145,111 @@ func TestLoadConfigurationSucceedsOnValidFiles(t *testing.T) {
 	}
 	assert.ElementsMatch(t, []string{"alice", "bob"}, names)
 }
+
+// changeAfterLoad runs change once, at the first filer lookup after the load
+// read the legacy file, so it lands before the migration writes anything.
+func changeAfterLoad(server *policyTestFilerServer, change func()) {
+	var lookups atomic.Int32
+	server.beforeLookup = func(context.Context, string, string) error {
+		if lookups.Add(1) == 2 {
+			change()
+		}
+		return nil
+	}
+}
+
+func setEntry(server *policyTestFilerServer, dir, name string, content []byte) {
+	_, _ = server.CreateEntry(context.Background(), &filer_pb.CreateEntryRequest{Directory: dir, Entry: &filer_pb.Entry{Name: name, Content: content, Attributes: &filer_pb.FuseAttributes{}}})
+}
+
+func storedIdentity(t *testing.T, server *policyTestFilerServer, name string) *iam_pb.Identity {
+	t.Helper()
+	server.mu.RLock()
+	entry, found := server.entries[filerEntryKey(filer.IamConfigDirectory+"/"+IamIdentitiesDirectory, name+".json")]
+	server.mu.RUnlock()
+	if !found {
+		return nil
+	}
+	identity := &iam_pb.Identity{}
+	require.NoError(t, json.Unmarshal(entry.Content, identity))
+	return identity
+}
+
+func storedFile(server *policyTestFilerServer, dir, name string) bool {
+	server.mu.RLock()
+	defer server.mu.RUnlock()
+	_, found := server.entries[filerEntryKey(dir, name)]
+	return found
+}
+
+// The legacy-file migration runs after the load read every IAM file. A
+// revoke that lands in between must not be undone by the migration.
+func TestMigrationKeepsChangesMadeAfterTheLoad(t *testing.T) {
+	ctx := context.Background()
+	identDir := filer.IamConfigDirectory + "/" + IamIdentitiesDirectory
+	granted := []byte(`{"name":"anonymous","actions":["Read:b"]}`)
+	revoked := []byte(`{"name":"anonymous","actions":["Read:c"]}`)
+	legacy := func(identities string) []byte { return []byte(`{"identities":[` + identities + `]}`) }
+
+	t.Run("rewritten identity file", func(t *testing.T) {
+		store, server := newPolicyTestStoreWithServer(t)
+		setEntry(server, filer.IamConfigDirectory, IamLegacyIdentityFile, legacy(string(granted)))
+		setEntry(server, identDir, "anonymous.json", granted)
+		changeAfterLoad(server, func() { setEntry(server, identDir, "anonymous.json", revoked) })
+
+		_, err := store.LoadConfiguration(ctx)
+		require.NoError(t, err)
+		assert.Equal(t, []string{"Read:c"}, storedIdentity(t, server, "anonymous").GetActions())
+		assert.True(t, storedFile(server, filer.IamConfigDirectory, IamLegacyIdentityOldFile))
+	})
+
+	t.Run("removed identity file", func(t *testing.T) {
+		store, server := newPolicyTestStoreWithServer(t)
+		setEntry(server, filer.IamConfigDirectory, IamLegacyIdentityFile, legacy(`{"name":"alice"}`))
+		setEntry(server, identDir, "bob.json", []byte(`{"name":"bob"}`))
+		changeAfterLoad(server, func() {
+			_, _ = server.DeleteEntry(ctx, &filer_pb.DeleteEntryRequest{Directory: identDir, Name: "bob.json"})
+		})
+
+		_, err := store.LoadConfiguration(ctx)
+		require.NoError(t, err)
+		assert.Nil(t, storedIdentity(t, server, "bob"))
+		assert.NotNil(t, storedIdentity(t, server, "alice"))
+	})
+
+	t.Run("removed identity file also in the legacy file", func(t *testing.T) {
+		store, server := newPolicyTestStoreWithServer(t)
+		setEntry(server, filer.IamConfigDirectory, IamLegacyIdentityFile, legacy(`{"name":"alice"},{"name":"bob","actions":["Read:b"]}`))
+		setEntry(server, identDir, "bob.json", []byte(`{"name":"bob","actions":["Read:b"]}`))
+		changeAfterLoad(server, func() {
+			_, _ = server.DeleteEntry(ctx, &filer_pb.DeleteEntryRequest{Directory: identDir, Name: "bob.json"})
+		})
+
+		_, err := store.LoadConfiguration(ctx)
+		require.NoError(t, err)
+		assert.Nil(t, storedIdentity(t, server, "bob"))
+		assert.NotNil(t, storedIdentity(t, server, "alice"))
+	})
+
+	t.Run("rewritten legacy file", func(t *testing.T) {
+		store, server := newPolicyTestStoreWithServer(t)
+		setEntry(server, filer.IamConfigDirectory, IamLegacyIdentityFile, legacy(string(granted)))
+		changeAfterLoad(server, func() {
+			setEntry(server, filer.IamConfigDirectory, IamLegacyIdentityFile, legacy(string(revoked)))
+		})
+
+		_, err := store.LoadConfiguration(ctx)
+		require.NoError(t, err)
+		if identity := storedIdentity(t, server, "anonymous"); identity != nil {
+			assert.Equal(t, []string{"Read:c"}, identity.GetActions())
+		}
+
+		s3cfg, err := store.LoadConfiguration(ctx)
+		require.NoError(t, err)
+		require.Len(t, s3cfg.Identities, 1)
+		assert.Equal(t, []string{"Read:c"}, s3cfg.Identities[0].GetActions())
+		assert.Equal(t, []string{"Read:c"}, storedIdentity(t, server, "anonymous").GetActions())
+		assert.False(t, storedFile(server, filer.IamConfigDirectory, IamLegacyIdentityFile))
+		assert.True(t, storedFile(server, filer.IamConfigDirectory, IamLegacyIdentityOldFile))
+	})
+}

weed/credential/filer_etc/filer_etc_policy_test.go

@@ -96,10 +96,29 @@ func (s *policyTestFilerServer) CreateEntry(_ context.Context, req *filer_pb.Cre
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	s.entries[filerEntryKey(req.Directory, req.Entry.Name)] = cloneEntry(req.Entry)
+	key := filerEntryKey(req.Directory, req.Entry.Name)
+	if _, found := s.entries[key]; found && req.OExcl {
+		return &filer_pb.CreateEntryResponse{Error: filer_pb.ErrEntryAlreadyExists.Error(), ErrorCode: filer_pb.FilerError_ENTRY_ALREADY_EXISTS}, nil
+	}
+	s.entries[key] = cloneEntry(req.Entry)
 	return &filer_pb.CreateEntryResponse{}, nil
 }
 
+func (s *policyTestFilerServer) AtomicRenameEntry(_ context.Context, req *filer_pb.AtomicRenameEntryRequest) (*filer_pb.AtomicRenameEntryResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	oldKey := filerEntryKey(req.OldDirectory, req.OldName)
+	entry, found := s.entries[oldKey]
+	if !found {
+		return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error())
+	}
+	delete(s.entries, oldKey)
+	entry.Name = req.NewName
+	s.entries[filerEntryKey(req.NewDirectory, req.NewName)] = entry
+	return &filer_pb.AtomicRenameEntryResponse{}, nil
+}
+
 func (s *policyTestFilerServer) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
 	s.mu.RLock()
 	beforeUpdate := s.beforeUpdate