IAM Policy Management via gRPC (#8109)

* Add IAM gRPC service definition

- Add GetConfiguration/PutConfiguration for config management
- Add CreateUser/GetUser/UpdateUser/DeleteUser/ListUsers for user management
- Add CreateAccessKey/DeleteAccessKey/GetUserByAccessKey for access key management
- Methods mirror existing IAM HTTP API functionality

* Add IAM gRPC handlers on filer server

- Implement IamGrpcServer with CredentialManager integration
- Handle configuration get/put operations
- Handle user CRUD operations
- Handle access key create/delete operations
- All methods delegate to CredentialManager for actual storage

* Wire IAM gRPC service to filer server

- Add CredentialManager field to FilerOption and FilerServer
- Import credential store implementations in filer command
- Initialize CredentialManager from credential.toml if available
- Register IAM gRPC service on filer gRPC server
- Enable credential management via gRPC alongside existing filer services

* Regenerate IAM protobuf with gRPC service methods

* iam_pb: add Policy Management to protobuf definitions

* credential: implement PolicyManager in credential stores

* filer: implement IAM Policy Management RPCs

* shell: add s3.policy command

* test: add integration test for s3.policy

* test: fix compilation errors in policy_test

* pb

* fmt

* test

* weed shell: add -policies flag to s3.configure

This allows linking/unlinking IAM policies to/from identities
directly from the s3.configure command.

* test: verify s3.configure policy linking and fix port allocation

- Added test case for linking policies to users via s3.configure
- Implemented findAvailablePortPair to ensure HTTP and gRPC ports
  are both available, avoiding conflicts with randomized port assignments.
- Updated assertion to match jsonpb output (policyNames)

* credential: add StoreTypeGrpc constant

* credential: add IAM gRPC store boilerplate

* credential: implement identity methods in gRPC store

* credential: implement policy methods in gRPC store

* admin: use gRPC credential store for AdminServer

This ensures that all IAM and policy changes made through the Admin UI
are persisted via the Filer's IAM gRPC service instead of direct file manipulation.

* shell: s3.configure use granular IAM gRPC APIs instead of full config patching

* shell: s3.configure use granular IAM gRPC APIs

* shell: replace deprecated ioutil with os in s3.policy

* filer: use gRPC FailedPrecondition for unconfigured credential manager

* test: improve s3.policy integration tests and fix error checks

* ci: add s3 policy shell integration tests to github workflow

* filer: fix LoadCredentialConfiguration error handling

* credential/grpc: propagate unmarshal errors in GetPolicies

* filer/grpc: improve error handling and validation

* shell: use gRPC status codes in s3.configure

* credential: document PutPolicy as create-or-replace

* credential/postgres: reuse CreatePolicy in PutPolicy to deduplicate logic

* shell: add timeout context and strictly enforce flags in s3.policy

* iam: standardize policy content field naming in gRPC and proto

* shell: extract slice helper functions in s3.configure

* filer: map credential store errors to gRPC status codes

* filer: add input validation for UpdateUser and CreateAccessKey

* iam: improve validation in policy and config handlers

* filer: ensure IAM service registration by defaulting credential manager

* credential: add GetStoreName method to manager

* test: verify policy deletion in integration test

weed/credential/credential_manager.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
@@ -46,6 +47,14 @@ func (cm *CredentialManager) GetStore() CredentialStore {
 	return cm.store
 }
 
+// GetStoreName returns the name of the underlying credential store
+func (cm *CredentialManager) GetStoreName() string {
+	if cm.store != nil {
+		return string(cm.store.GetName())
+	}
+	return ""
+}
+
 // LoadConfiguration loads the S3 API configuration
 func (cm *CredentialManager) LoadConfiguration(ctx context.Context) (*iam_pb.S3ApiConfiguration, error) {
 	return cm.store.LoadConfiguration(ctx)
@@ -96,6 +105,26 @@ func (cm *CredentialManager) DeleteAccessKey(ctx context.Context, username strin
 	return cm.store.DeleteAccessKey(ctx, username, accessKey)
 }
 
+// GetPolicies returns all policies
+func (cm *CredentialManager) GetPolicies(ctx context.Context) (map[string]policy_engine.PolicyDocument, error) {
+	return cm.store.GetPolicies(ctx)
+}
+
+// PutPolicy creates or updates a policy
+func (cm *CredentialManager) PutPolicy(ctx context.Context, name string, document policy_engine.PolicyDocument) error {
+	return cm.store.PutPolicy(ctx, name, document)
+}
+
+// DeletePolicy removes a policy
+func (cm *CredentialManager) DeletePolicy(ctx context.Context, name string) error {
+	return cm.store.DeletePolicy(ctx, name)
+}
+
+// GetPolicy retrieves a policy by name
+func (cm *CredentialManager) GetPolicy(ctx context.Context, name string) (*policy_engine.PolicyDocument, error) {
+	return cm.store.GetPolicy(ctx, name)
+}
+
 // Shutdown performs cleanup
 func (cm *CredentialManager) Shutdown() {
 	if cm.store != nil {

weed/credential/credential_store.go

@@ -25,6 +25,7 @@ const (
 	StoreTypeFilerEtc      CredentialStoreTypeName = "filer_etc"
 	StoreTypeFilerMultiple CredentialStoreTypeName = "filer_multiple"
 	StoreTypePostgres      CredentialStoreTypeName = "postgres"
+	StoreTypeGrpc          CredentialStoreTypeName = "grpc"
 )
 
 // CredentialStore defines the interface for user credential storage and retrieval
@@ -65,6 +66,13 @@ type CredentialStore interface {
 	// DeleteAccessKey removes an access key for a user
 	DeleteAccessKey(ctx context.Context, username string, accessKey string) error
 
+	// Policy Management
+	GetPolicies(ctx context.Context) (map[string]policy_engine.PolicyDocument, error)
+	// PutPolicy creates or replaces a policy document.
+	PutPolicy(ctx context.Context, name string, document policy_engine.PolicyDocument) error
+	DeletePolicy(ctx context.Context, name string) error
+	GetPolicy(ctx context.Context, name string) (*policy_engine.PolicyDocument, error)
+
 	// Shutdown performs cleanup when the store is being shut down
 	Shutdown()
 }

weed/credential/filer_etc/filer_etc_policy.go

@@ -98,6 +98,11 @@ func (store *FilerEtcStore) UpdatePolicy(ctx context.Context, name string, docum
 	})
 }
 
+// PutPolicy creates or updates an IAM policy in the filer
+func (store *FilerEtcStore) PutPolicy(ctx context.Context, name string, document policy_engine.PolicyDocument) error {
+	return store.UpdatePolicy(ctx, name, document)
+}
+
 // DeletePolicy deletes an IAM policy from the filer
 func (store *FilerEtcStore) DeletePolicy(ctx context.Context, name string) error {
 	return store.updatePolicies(ctx, func(policies map[string]policy_engine.PolicyDocument) {

weed/credential/filer_multiple/filer_multiple_store.go

@@ -429,6 +429,14 @@ func (store *FilerMultipleStore) CreatePolicy(ctx context.Context, name string,
 	})
 }
 
+func (store *FilerMultipleStore) PutPolicy(ctx context.Context, name string, document policy_engine.PolicyDocument) error {
+	return store.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+		// We can just overwrite. The distinction between Create and Update in filer_multiple was just checking existence.
+		// Put implies "create or replace".
+		return store.savePolicy(ctx, client, name, document)
+	})
+}
+
 func (store *FilerMultipleStore) UpdatePolicy(ctx context.Context, name string, document policy_engine.PolicyDocument) error {
 	return store.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 		filename := name + ".json"

weed/credential/grpc/grpc_identity.go

@@ -0,0 +1,120 @@
+package grpc
+
+import (
+	"context"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+)
+
+func (store *IamGrpcStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3ApiConfiguration, error) {
+	var config *iam_pb.S3ApiConfiguration
+	err := store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		resp, err := client.GetConfiguration(ctx, &iam_pb.GetConfigurationRequest{})
+		if err != nil {
+			return err
+		}
+		config = resp.Configuration
+		return nil
+	})
+	return config, err
+}
+
+func (store *IamGrpcStore) SaveConfiguration(ctx context.Context, config *iam_pb.S3ApiConfiguration) error {
+	return store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		_, err := client.PutConfiguration(ctx, &iam_pb.PutConfigurationRequest{
+			Configuration: config,
+		})
+		return err
+	})
+}
+
+func (store *IamGrpcStore) CreateUser(ctx context.Context, identity *iam_pb.Identity) error {
+	return store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		_, err := client.CreateUser(ctx, &iam_pb.CreateUserRequest{
+			Identity: identity,
+		})
+		return err
+	})
+}
+
+func (store *IamGrpcStore) GetUser(ctx context.Context, username string) (*iam_pb.Identity, error) {
+	var identity *iam_pb.Identity
+	err := store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		resp, err := client.GetUser(ctx, &iam_pb.GetUserRequest{
+			Username: username,
+		})
+		if err != nil {
+			return err
+		}
+		identity = resp.Identity
+		return nil
+	})
+	return identity, err
+}
+
+func (store *IamGrpcStore) UpdateUser(ctx context.Context, username string, identity *iam_pb.Identity) error {
+	return store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		_, err := client.UpdateUser(ctx, &iam_pb.UpdateUserRequest{
+			Username: username,
+			Identity: identity,
+		})
+		return err
+	})
+}
+
+func (store *IamGrpcStore) DeleteUser(ctx context.Context, username string) error {
+	return store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		_, err := client.DeleteUser(ctx, &iam_pb.DeleteUserRequest{
+			Username: username,
+		})
+		return err
+	})
+}
+
+func (store *IamGrpcStore) ListUsers(ctx context.Context) ([]string, error) {
+	var usernames []string
+	err := store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		resp, err := client.ListUsers(ctx, &iam_pb.ListUsersRequest{})
+		if err != nil {
+			return err
+		}
+		usernames = resp.Usernames
+		return nil
+	})
+	return usernames, err
+}
+
+func (store *IamGrpcStore) GetUserByAccessKey(ctx context.Context, accessKey string) (*iam_pb.Identity, error) {
+	var identity *iam_pb.Identity
+	err := store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		resp, err := client.GetUserByAccessKey(ctx, &iam_pb.GetUserByAccessKeyRequest{
+			AccessKey: accessKey,
+		})
+		if err != nil {
+			return err
+		}
+		identity = resp.Identity
+		return nil
+	})
+	return identity, err
+}
+
+func (store *IamGrpcStore) CreateAccessKey(ctx context.Context, username string, credential *iam_pb.Credential) error {
+	return store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		_, err := client.CreateAccessKey(ctx, &iam_pb.CreateAccessKeyRequest{
+			Username:   username,
+			Credential: credential,
+		})
+		return err
+	})
+}
+
+func (store *IamGrpcStore) DeleteAccessKey(ctx context.Context, username string, accessKey string) error {
+	return store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		_, err := client.DeleteAccessKey(ctx, &iam_pb.DeleteAccessKeyRequest{
+			Username:  username,
+			AccessKey: accessKey,
+		})
+		return err
+	})
+}

weed/credential/grpc/grpc_policy.go

@@ -0,0 +1,69 @@
+package grpc
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
+)
+
+func (store *IamGrpcStore) GetPolicies(ctx context.Context) (map[string]policy_engine.PolicyDocument, error) {
+	policies := make(map[string]policy_engine.PolicyDocument)
+	err := store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		resp, err := client.ListPolicies(ctx, &iam_pb.ListPoliciesRequest{})
+		if err != nil {
+			return err
+		}
+		for _, p := range resp.Policies {
+			var doc policy_engine.PolicyDocument
+			if err := json.Unmarshal([]byte(p.Content), &doc); err != nil {
+				return fmt.Errorf("failed to unmarshal policy %s: %v", p.Name, err)
+			}
+			policies[p.Name] = doc
+		}
+		return nil
+	})
+	return policies, err
+}
+
+func (store *IamGrpcStore) PutPolicy(ctx context.Context, name string, document policy_engine.PolicyDocument) error {
+	content, err := json.Marshal(document)
+	if err != nil {
+		return err
+	}
+	return store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		_, err := client.PutPolicy(ctx, &iam_pb.PutPolicyRequest{
+			Name:    name,
+			Content: string(content),
+		})
+		return err
+	})
+}
+
+func (store *IamGrpcStore) DeletePolicy(ctx context.Context, name string) error {
+	return store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		_, err := client.DeletePolicy(ctx, &iam_pb.DeletePolicyRequest{
+			Name: name,
+		})
+		return err
+	})
+}
+
+func (store *IamGrpcStore) GetPolicy(ctx context.Context, name string) (*policy_engine.PolicyDocument, error) {
+	var doc policy_engine.PolicyDocument
+	err := store.withIamClient(func(client iam_pb.SeaweedIdentityAccessManagementClient) error {
+		resp, err := client.GetPolicy(ctx, &iam_pb.GetPolicyRequest{
+			Name: name,
+		})
+		if err != nil {
+			return err
+		}
+		return json.Unmarshal([]byte(resp.Content), &doc)
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &doc, nil
+}

weed/credential/grpc/grpc_store.go

@@ -0,0 +1,72 @@
+package grpc
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/seaweedfs/seaweedfs/weed/credential"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"google.golang.org/grpc"
+)
+
+func init() {
+	credential.Stores = append(credential.Stores, &IamGrpcStore{})
+}
+
+// IamGrpcStore implements CredentialStore using SeaweedFS IAM gRPC service
+type IamGrpcStore struct {
+	filerAddressFunc func() pb.ServerAddress // Function to get current active filer
+	grpcDialOption   grpc.DialOption
+	mu               sync.RWMutex // Protects filerAddressFunc and grpcDialOption
+}
+
+func (store *IamGrpcStore) GetName() credential.CredentialStoreTypeName {
+	return credential.StoreTypeGrpc
+}
+
+func (store *IamGrpcStore) Initialize(configuration util.Configuration, prefix string) error {
+	if configuration != nil {
+		filerAddr := configuration.GetString(prefix + "filer")
+		if filerAddr != "" {
+			store.mu.Lock()
+			store.filerAddressFunc = func() pb.ServerAddress {
+				return pb.ServerAddress(filerAddr)
+			}
+			store.mu.Unlock()
+		}
+	}
+	return nil
+}
+
+func (store *IamGrpcStore) SetFilerAddressFunc(getFiler func() pb.ServerAddress, grpcDialOption grpc.DialOption) {
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	store.filerAddressFunc = getFiler
+	store.grpcDialOption = grpcDialOption
+}
+
+func (store *IamGrpcStore) withIamClient(fn func(client iam_pb.SeaweedIdentityAccessManagementClient) error) error {
+	store.mu.RLock()
+	if store.filerAddressFunc == nil {
+		store.mu.RUnlock()
+		return fmt.Errorf("iam_grpc: filer not yet available")
+	}
+
+	filerAddress := store.filerAddressFunc()
+	dialOption := store.grpcDialOption
+	store.mu.RUnlock()
+
+	if filerAddress == "" {
+		return fmt.Errorf("iam_grpc: no filer discovered yet")
+	}
+
+	return pb.WithGrpcClient(false, 0, func(conn *grpc.ClientConn) error {
+		client := iam_pb.NewSeaweedIdentityAccessManagementClient(conn)
+		return fn(client)
+	}, filerAddress.ToGrpcAddress(), false, dialOption)
+}
+
+func (store *IamGrpcStore) Shutdown() {
+}

weed/credential/memory/memory_policy.go

@@ -63,6 +63,19 @@ func (store *MemoryStore) UpdatePolicy(ctx context.Context, name string, documen
 	return nil
 }
 
+// PutPolicy creates or updates an IAM policy in memory
+func (store *MemoryStore) PutPolicy(ctx context.Context, name string, document policy_engine.PolicyDocument) error {
+	store.mu.Lock()
+	defer store.mu.Unlock()
+
+	if !store.initialized {
+		return fmt.Errorf("store not initialized")
+	}
+
+	store.policies[name] = document
+	return nil
+}
+
 // DeletePolicy deletes an IAM policy from memory
 func (store *MemoryStore) DeletePolicy(ctx context.Context, name string) error {
 	store.mu.Lock()

weed/credential/postgres/postgres_policy.go

@@ -62,6 +62,11 @@ func (store *PostgresStore) CreatePolicy(ctx context.Context, name string, docum
 	return nil
 }
 
+// PutPolicy creates or updates an IAM policy in PostgreSQL
+func (store *PostgresStore) PutPolicy(ctx context.Context, name string, document policy_engine.PolicyDocument) error {
+	return store.CreatePolicy(ctx, name, document)
+}
+
 // UpdatePolicy updates an existing IAM policy in PostgreSQL
 func (store *PostgresStore) UpdatePolicy(ctx context.Context, name string, document policy_engine.PolicyDocument) error {
 	if !store.configured {