From dc462a80d712b1179c8841c519d2ec47152b6503 Mon Sep 17 00:00:00 2001 From: Jon E Nesvold <29639563+JonNesvold@users.noreply.github.com> Date: Sun, 26 Apr 2026 23:54:53 +0200 Subject: [PATCH] feat(credential/postgres): inline policies, mTLS and pgbouncer connection support (#9226) * feat(credential/postgres): mTLS + pgbouncer support, InlinePolicyStore implementation, upsert SaveConfiguration * fix(credential/postgres): add rows.Err() checks, inline policy tests, memory store LoadInlinePolicies * fix(credential/postgres): cast JSONB params to string for pgbouncer simple protocol * fix(credential/postgres): wrap tx.Commit errors with context * fix(credential/postgres): use any type for JSONB params to preserve SQL NULL for nil fields --- weed/credential/memory/memory_policy.go | 20 ++ weed/credential/postgres/postgres_identity.go | 231 +++++++++++------- .../postgres/postgres_inline_policy.go | 155 ++++++++++++ weed/credential/postgres/postgres_store.go | 69 ++++-- weed/credential/test/inline_policy_test.go | 170 +++++++++++++ 5 files changed, 536 insertions(+), 109 deletions(-) create mode 100644 weed/credential/postgres/postgres_inline_policy.go create mode 100644 weed/credential/test/inline_policy_test.go diff --git a/weed/credential/memory/memory_policy.go b/weed/credential/memory/memory_policy.go index 228b09844..078982793 100644 --- a/weed/credential/memory/memory_policy.go +++ b/weed/credential/memory/memory_policy.go @@ -173,3 +173,23 @@ func (store *MemoryStore) ListUserInlinePolicies(ctx context.Context, userName s } return names, nil } + +// LoadInlinePolicies returns all inline policies keyed by username then policy name. +func (store *MemoryStore) LoadInlinePolicies(ctx context.Context) (map[string]map[string]policy_engine.PolicyDocument, error) { + store.mu.RLock() + defer store.mu.RUnlock() + + if !store.initialized { + return nil, fmt.Errorf("store not initialized") + } + + result := make(map[string]map[string]policy_engine.PolicyDocument, len(store.inlinePolicies)) + for userName, userPolicies := range store.inlinePolicies { + copied := make(map[string]policy_engine.PolicyDocument, len(userPolicies)) + for policyName, doc := range userPolicies { + copied[policyName] = doc + } + result[userName] = copied + } + return result, nil +} diff --git a/weed/credential/postgres/postgres_identity.go b/weed/credential/postgres/postgres_identity.go index 74934c9c4..288734ead 100644 --- a/weed/credential/postgres/postgres_identity.go +++ b/weed/credential/postgres/postgres_identity.go @@ -7,6 +7,7 @@ import ( "fmt" "github.com/seaweedfs/seaweedfs/weed/credential" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/iam_pb" ) @@ -17,9 +18,9 @@ func (store *PostgresStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3Ap config := &iam_pb.S3ApiConfiguration{} - // Query all users rows, err := store.db.QueryContext(ctx, "SELECT username, email, account_data, actions, policy_names FROM users") if err != nil { + glog.Errorf("credential postgres: LoadConfiguration query failed: %v", err) return nil, fmt.Errorf("failed to query users: %w", err) } defer rows.Close() @@ -29,6 +30,7 @@ func (store *PostgresStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3Ap var accountDataJSON, actionsJSON, policyNamesJSON []byte if err := rows.Scan(&username, &email, &accountDataJSON, &actionsJSON, &policyNamesJSON); err != nil { + glog.Errorf("credential postgres: LoadConfiguration scan failed: %v", err) return nil, fmt.Errorf("failed to scan user row: %w", err) } @@ -36,28 +38,24 @@ func (store *PostgresStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3Ap Name: username, } - // Parse account data if len(accountDataJSON) > 0 { if err := json.Unmarshal(accountDataJSON, &identity.Account); err != nil { return nil, fmt.Errorf("failed to unmarshal account data for user %s: %v", username, err) } } - // Parse actions if len(actionsJSON) > 0 { if err := json.Unmarshal(actionsJSON, &identity.Actions); err != nil { return nil, fmt.Errorf("failed to unmarshal actions for user %s: %v", username, err) } } - // Parse policy names if len(policyNamesJSON) > 0 { if err := json.Unmarshal(policyNamesJSON, &identity.PolicyNames); err != nil { return nil, fmt.Errorf("failed to unmarshal policy names for user %s: %v", username, err) } } - // Query credentials for this user credRows, err := store.db.QueryContext(ctx, "SELECT access_key, secret_key FROM credentials WHERE username = $1", username) if err != nil { return nil, fmt.Errorf("failed to query credentials for user %s: %v", username, err) @@ -76,10 +74,16 @@ func (store *PostgresStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3Ap }) } credRows.Close() - + if err := credRows.Err(); err != nil { + return nil, fmt.Errorf("failed iterating credential rows for user %s: %w", username, err) + } config.Identities = append(config.Identities, identity) } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed iterating user rows: %w", err) + } + glog.V(0).Infof("credential postgres: LoadConfiguration loaded %d identities", len(config.Identities)) return config, nil } @@ -88,59 +92,62 @@ func (store *PostgresStore) SaveConfiguration(ctx context.Context, config *iam_p return fmt.Errorf("store not configured") } - // Start transaction tx, err := store.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } defer tx.Rollback() - // Clear existing data - if _, err := tx.ExecContext(ctx, "DELETE FROM credentials"); err != nil { - return fmt.Errorf("failed to clear credentials: %w", err) - } - if _, err := tx.ExecContext(ctx, "DELETE FROM users"); err != nil { - return fmt.Errorf("failed to clear users: %w", err) - } + // Track which usernames are in the incoming config for pruning + configUsernames := make(map[string]bool, len(config.Identities)) - // Insert all identities for _, identity := range config.Identities { - // Marshal account data - var accountDataJSON []byte + configUsernames[identity.Name] = true + + var accountDataParam any if identity.Account != nil { - accountDataJSON, err = json.Marshal(identity.Account) + b, err := json.Marshal(identity.Account) if err != nil { return fmt.Errorf("failed to marshal account data for user %s: %v", identity.Name, err) } + accountDataParam = string(b) } - - // Marshal actions - var actionsJSON []byte + var actionsParam any if identity.Actions != nil { - actionsJSON, err = json.Marshal(identity.Actions) + b, err := json.Marshal(identity.Actions) if err != nil { return fmt.Errorf("failed to marshal actions for user %s: %v", identity.Name, err) } + actionsParam = string(b) } - - // Marshal policy names - var policyNamesJSON []byte + var policyNamesParam any if identity.PolicyNames != nil { - policyNamesJSON, err = json.Marshal(identity.PolicyNames) + b, err := json.Marshal(identity.PolicyNames) if err != nil { return fmt.Errorf("failed to marshal policy names for user %s: %v", identity.Name, err) } + policyNamesParam = string(b) } - - // Insert user - _, err := tx.ExecContext(ctx, - "INSERT INTO users (username, email, account_data, actions, policy_names) VALUES ($1, $2, $3, $4, $5)", - identity.Name, "", accountDataJSON, actionsJSON, policyNamesJSON) + // Upsert user — preserves the row (and its CASCADE dependents) if it already exists + _, err = tx.ExecContext(ctx, + `INSERT INTO users (username, email, account_data, actions, policy_names) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (username) DO UPDATE SET + email = EXCLUDED.email, + account_data = EXCLUDED.account_data, + actions = EXCLUDED.actions, + policy_names = EXCLUDED.policy_names, + updated_at = CURRENT_TIMESTAMP`, + identity.Name, "", accountDataParam, actionsParam, policyNamesParam) if err != nil { - return fmt.Errorf("failed to insert user %s: %v", identity.Name, err) + return fmt.Errorf("failed to upsert user %s: %v", identity.Name, err) } - // Insert credentials + // Replace credentials for this user — credentials carry no independent + // state worth preserving (unlike inline policies) + if _, err := tx.ExecContext(ctx, "DELETE FROM credentials WHERE username = $1", identity.Name); err != nil { + return fmt.Errorf("failed to clear credentials for user %s: %v", identity.Name, err) + } for _, cred := range identity.Credentials { _, err := tx.ExecContext(ctx, "INSERT INTO credentials (username, access_key, secret_key) VALUES ($1, $2, $3)", @@ -151,6 +158,35 @@ func (store *PostgresStore) SaveConfiguration(ctx context.Context, config *iam_p } } + // Prune users no longer in config — CASCADE correctly removes their + // credentials and inline policies since they were intentionally deleted + rows, err := tx.QueryContext(ctx, "SELECT username FROM users") + if err != nil { + return fmt.Errorf("failed to list existing users for pruning: %w", err) + } + var toDelete []string + for rows.Next() { + var username string + if err := rows.Scan(&username); err != nil { + rows.Close() + return fmt.Errorf("failed to scan username for pruning: %w", err) + } + if !configUsernames[username] { + toDelete = append(toDelete, username) + } + } + rows.Close() + if err := rows.Err(); err != nil { + return fmt.Errorf("failed iterating user rows for pruning: %w", err) + } + + for _, username := range toDelete { + if _, err := tx.ExecContext(ctx, "DELETE FROM users WHERE username = $1", username); err != nil { + return fmt.Errorf("failed to prune user %s: %v", username, err) + } + } + + glog.V(0).Infof("credential postgres: SaveConfiguration saved %d identities, pruned %d", len(config.Identities), len(toDelete)) return tx.Commit() } @@ -159,69 +195,72 @@ func (store *PostgresStore) CreateUser(ctx context.Context, identity *iam_pb.Ide return fmt.Errorf("store not configured") } - // Check if user already exists var count int err := store.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM users WHERE username = $1", identity.Name).Scan(&count) if err != nil { + glog.Errorf("credential postgres: CreateUser check failed user=%s: %v", identity.Name, err) return fmt.Errorf("failed to check user existence: %w", err) } if count > 0 { + glog.V(1).Infof("credential postgres: CreateUser user=%s already exists", identity.Name) return credential.ErrUserAlreadyExists } - // Start transaction tx, err := store.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } defer tx.Rollback() - // Marshal account data - var accountDataJSON []byte + var accountDataParam any if identity.Account != nil { - accountDataJSON, err = json.Marshal(identity.Account) + b, err := json.Marshal(identity.Account) if err != nil { return fmt.Errorf("failed to marshal account data: %w", err) } + accountDataParam = string(b) } - - // Marshal actions - var actionsJSON []byte + var actionsParam any if identity.Actions != nil { - actionsJSON, err = json.Marshal(identity.Actions) + b, err := json.Marshal(identity.Actions) if err != nil { return fmt.Errorf("failed to marshal actions: %w", err) } + actionsParam = string(b) } - - // Marshal policy names - var policyNamesJSON []byte + var policyNamesParam any if identity.PolicyNames != nil { - policyNamesJSON, err = json.Marshal(identity.PolicyNames) + b, err := json.Marshal(identity.PolicyNames) if err != nil { return fmt.Errorf("failed to marshal policy names: %w", err) } + policyNamesParam = string(b) } - - // Insert user _, err = tx.ExecContext(ctx, "INSERT INTO users (username, email, account_data, actions, policy_names) VALUES ($1, $2, $3, $4, $5)", - identity.Name, "", accountDataJSON, actionsJSON, policyNamesJSON) + identity.Name, "", accountDataParam, actionsParam, policyNamesParam) if err != nil { + glog.Errorf("credential postgres: CreateUser insert failed user=%s: %v", identity.Name, err) return fmt.Errorf("failed to insert user: %w", err) } - // Insert credentials for _, cred := range identity.Credentials { _, err = tx.ExecContext(ctx, "INSERT INTO credentials (username, access_key, secret_key) VALUES ($1, $2, $3)", identity.Name, cred.AccessKey, cred.SecretKey) if err != nil { + glog.Errorf("credential postgres: CreateUser insert credential failed user=%s accessKey=%s: %v", identity.Name, cred.AccessKey, err) return fmt.Errorf("failed to insert credential: %w", err) } } - return tx.Commit() + if err := tx.Commit(); err != nil { + glog.Errorf("credential postgres: CreateUser commit failed user=%s: %v", identity.Name, err) + return fmt.Errorf("failed to commit: %w", err) + } + + glog.V(0).Infof("credential postgres: CreateUser user=%s credentials=%d actions=%d", identity.Name, len(identity.Credentials), len(identity.Actions)) + return nil } func (store *PostgresStore) GetUser(ctx context.Context, username string) (*iam_pb.Identity, error) { @@ -237,8 +276,10 @@ func (store *PostgresStore) GetUser(ctx context.Context, username string) (*iam_ username).Scan(&email, &accountDataJSON, &actionsJSON, &policyNamesJSON) if err != nil { if err == sql.ErrNoRows { + glog.V(2).Infof("credential postgres: GetUser user=%s not found", username) return nil, credential.ErrUserNotFound } + glog.Errorf("credential postgres: GetUser query failed user=%s: %v", username, err) return nil, fmt.Errorf("failed to query user: %w", err) } @@ -246,28 +287,24 @@ func (store *PostgresStore) GetUser(ctx context.Context, username string) (*iam_ Name: username, } - // Parse account data if len(accountDataJSON) > 0 { if err := json.Unmarshal(accountDataJSON, &identity.Account); err != nil { return nil, fmt.Errorf("failed to unmarshal account data: %w", err) } } - // Parse actions if len(actionsJSON) > 0 { if err := json.Unmarshal(actionsJSON, &identity.Actions); err != nil { return nil, fmt.Errorf("failed to unmarshal actions: %w", err) } } - // Parse policy names if len(policyNamesJSON) > 0 { if err := json.Unmarshal(policyNamesJSON, &identity.PolicyNames); err != nil { return nil, fmt.Errorf("failed to unmarshal policy names: %w", err) } } - // Query credentials rows, err := store.db.QueryContext(ctx, "SELECT access_key, secret_key FROM credentials WHERE username = $1", username) if err != nil { return nil, fmt.Errorf("failed to query credentials: %w", err) @@ -279,29 +316,31 @@ func (store *PostgresStore) GetUser(ctx context.Context, username string) (*iam_ if err := rows.Scan(&accessKey, &secretKey); err != nil { return nil, fmt.Errorf("failed to scan credential: %w", err) } - identity.Credentials = append(identity.Credentials, &iam_pb.Credential{ AccessKey: accessKey, SecretKey: secretKey, }) } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed iterating credential rows: %w", err) + } + glog.V(2).Infof("credential postgres: GetUser user=%s credentials=%d actions=%d", username, len(identity.Credentials), len(identity.Actions)) return identity, nil } + func (store *PostgresStore) UpdateUser(ctx context.Context, username string, identity *iam_pb.Identity) error { if !store.configured { return fmt.Errorf("store not configured") } - // Start transaction tx, err := store.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } defer tx.Rollback() - // Check if user exists var count int err = tx.QueryRowContext(ctx, "SELECT COUNT(*) FROM users WHERE username = $1", username).Scan(&count) if err != nil { @@ -311,48 +350,43 @@ func (store *PostgresStore) UpdateUser(ctx context.Context, username string, ide return credential.ErrUserNotFound } - // Marshal account data - var accountDataJSON []byte + var accountDataParam any if identity.Account != nil { - accountDataJSON, err = json.Marshal(identity.Account) + b, err := json.Marshal(identity.Account) if err != nil { return fmt.Errorf("failed to marshal account data: %w", err) } + accountDataParam = string(b) } - - // Marshal actions - var actionsJSON []byte + var actionsParam any if identity.Actions != nil { - actionsJSON, err = json.Marshal(identity.Actions) + b, err := json.Marshal(identity.Actions) if err != nil { return fmt.Errorf("failed to marshal actions: %w", err) } + actionsParam = string(b) } - - // Marshal policy names - var policyNamesJSON []byte + var policyNamesParam any if identity.PolicyNames != nil { - policyNamesJSON, err = json.Marshal(identity.PolicyNames) + b, err := json.Marshal(identity.PolicyNames) if err != nil { return fmt.Errorf("failed to marshal policy names: %w", err) } + policyNamesParam = string(b) } - - // Update user _, err = tx.ExecContext(ctx, "UPDATE users SET email = $2, account_data = $3, actions = $4, policy_names = $5, updated_at = CURRENT_TIMESTAMP WHERE username = $1", - username, "", accountDataJSON, actionsJSON, policyNamesJSON) + username, "", accountDataParam, actionsParam, policyNamesParam) if err != nil { + glog.Errorf("credential postgres: UpdateUser failed user=%s: %v", username, err) return fmt.Errorf("failed to update user: %w", err) } - // Delete existing credentials _, err = tx.ExecContext(ctx, "DELETE FROM credentials WHERE username = $1", username) if err != nil { return fmt.Errorf("failed to delete existing credentials: %w", err) } - // Insert new credentials for _, cred := range identity.Credentials { _, err = tx.ExecContext(ctx, "INSERT INTO credentials (username, access_key, secret_key) VALUES ($1, $2, $3)", @@ -362,7 +396,13 @@ func (store *PostgresStore) UpdateUser(ctx context.Context, username string, ide } } - return tx.Commit() + if err := tx.Commit(); err != nil { + glog.Errorf("credential postgres: UpdateUser commit failed user=%s: %v", username, err) + return fmt.Errorf("failed to commit: %w", err) + } + + glog.V(0).Infof("credential postgres: UpdateUser user=%s credentials=%d", username, len(identity.Credentials)) + return nil } func (store *PostgresStore) DeleteUser(ctx context.Context, username string) error { @@ -372,6 +412,7 @@ func (store *PostgresStore) DeleteUser(ctx context.Context, username string) err result, err := store.db.ExecContext(ctx, "DELETE FROM users WHERE username = $1", username) if err != nil { + glog.Errorf("credential postgres: DeleteUser failed user=%s: %v", username, err) return fmt.Errorf("failed to delete user: %w", err) } @@ -381,9 +422,11 @@ func (store *PostgresStore) DeleteUser(ctx context.Context, username string) err } if rowsAffected == 0 { + glog.V(1).Infof("credential postgres: DeleteUser user=%s not found", username) return credential.ErrUserNotFound } + glog.V(0).Infof("credential postgres: DeleteUser user=%s", username) return nil } @@ -394,6 +437,7 @@ func (store *PostgresStore) ListUsers(ctx context.Context) ([]string, error) { rows, err := store.db.QueryContext(ctx, "SELECT username FROM users ORDER BY username") if err != nil { + glog.Errorf("credential postgres: ListUsers query failed: %v", err) return nil, fmt.Errorf("failed to query users: %w", err) } defer rows.Close() @@ -406,7 +450,11 @@ func (store *PostgresStore) ListUsers(ctx context.Context) ([]string, error) { } usernames = append(usernames, username) } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed iterating user rows: %w", err) + } + glog.V(1).Infof("credential postgres: ListUsers count=%d", len(usernames)) return usernames, nil } @@ -419,11 +467,14 @@ func (store *PostgresStore) GetUserByAccessKey(ctx context.Context, accessKey st err := store.db.QueryRowContext(ctx, "SELECT username FROM credentials WHERE access_key = $1", accessKey).Scan(&username) if err != nil { if err == sql.ErrNoRows { + glog.V(2).Infof("credential postgres: GetUserByAccessKey accessKey=%s not found", accessKey) return nil, credential.ErrAccessKeyNotFound } + glog.Errorf("credential postgres: GetUserByAccessKey query failed accessKey=%s: %v", accessKey, err) return nil, fmt.Errorf("failed to query access key: %w", err) } + glog.V(2).Infof("credential postgres: GetUserByAccessKey accessKey=%s resolved to user=%s", accessKey, username) return store.GetUser(ctx, username) } @@ -432,7 +483,6 @@ func (store *PostgresStore) CreateAccessKey(ctx context.Context, username string return fmt.Errorf("store not configured") } - // Check if user exists var count int err := store.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM users WHERE username = $1", username).Scan(&count) if err != nil { @@ -442,14 +492,15 @@ func (store *PostgresStore) CreateAccessKey(ctx context.Context, username string return credential.ErrUserNotFound } - // Insert credential _, err = store.db.ExecContext(ctx, "INSERT INTO credentials (username, access_key, secret_key) VALUES ($1, $2, $3)", username, cred.AccessKey, cred.SecretKey) if err != nil { + glog.Errorf("credential postgres: CreateAccessKey failed user=%s accessKey=%s: %v", username, cred.AccessKey, err) return fmt.Errorf("failed to insert credential: %w", err) } + glog.V(0).Infof("credential postgres: CreateAccessKey user=%s accessKey=%s", username, cred.AccessKey) return nil } @@ -462,6 +513,7 @@ func (store *PostgresStore) DeleteAccessKey(ctx context.Context, username string "DELETE FROM credentials WHERE username = $1 AND access_key = $2", username, accessKey) if err != nil { + glog.Errorf("credential postgres: DeleteAccessKey failed user=%s accessKey=%s: %v", username, accessKey, err) return fmt.Errorf("failed to delete access key: %w", err) } @@ -471,7 +523,6 @@ func (store *PostgresStore) DeleteAccessKey(ctx context.Context, username string } if rowsAffected == 0 { - // Check if user exists var count int err = store.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM users WHERE username = $1", username).Scan(&count) if err != nil { @@ -483,22 +534,20 @@ func (store *PostgresStore) DeleteAccessKey(ctx context.Context, username string return credential.ErrAccessKeyNotFound } + glog.V(0).Infof("credential postgres: DeleteAccessKey user=%s accessKey=%s", username, accessKey) return nil } -// AttachUserPolicy attaches a managed policy to a user by policy name func (store *PostgresStore) AttachUserPolicy(ctx context.Context, username string, policyName string) error { if !store.configured { return fmt.Errorf("store not configured") } - // Get user identity, err := store.GetUser(ctx, username) if err != nil { return err } - // Verify policy exists policy, err := store.GetPolicy(ctx, policyName) if err != nil { return err @@ -507,31 +556,31 @@ func (store *PostgresStore) AttachUserPolicy(ctx context.Context, username strin return credential.ErrPolicyNotFound } - // Check if already attached for _, p := range identity.PolicyNames { if p == policyName { return credential.ErrPolicyAlreadyAttached } } - // Append policy name and update identity.PolicyNames = append(identity.PolicyNames, policyName) - return store.UpdateUser(ctx, username, identity) + if err := store.UpdateUser(ctx, username, identity); err != nil { + return err + } + + glog.V(0).Infof("credential postgres: AttachUserPolicy user=%s policy=%s", username, policyName) + return nil } -// DetachUserPolicy detaches a managed policy from a user func (store *PostgresStore) DetachUserPolicy(ctx context.Context, username string, policyName string) error { if !store.configured { return fmt.Errorf("store not configured") } - // Get user identity, err := store.GetUser(ctx, username) if err != nil { return err } - // Find and remove policy found := false var newPolicyNames []string for _, p := range identity.PolicyNames { @@ -547,10 +596,14 @@ func (store *PostgresStore) DetachUserPolicy(ctx context.Context, username strin } identity.PolicyNames = newPolicyNames - return store.UpdateUser(ctx, username, identity) + if err := store.UpdateUser(ctx, username, identity); err != nil { + return err + } + + glog.V(0).Infof("credential postgres: DetachUserPolicy user=%s policy=%s", username, policyName) + return nil } -// ListAttachedUserPolicies returns the list of policy names attached to a user func (store *PostgresStore) ListAttachedUserPolicies(ctx context.Context, username string) ([]string, error) { if !store.configured { return nil, fmt.Errorf("store not configured") diff --git a/weed/credential/postgres/postgres_inline_policy.go b/weed/credential/postgres/postgres_inline_policy.go new file mode 100644 index 000000000..8e47f9a4b --- /dev/null +++ b/weed/credential/postgres/postgres_inline_policy.go @@ -0,0 +1,155 @@ +package postgres + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine" +) + +func (store *PostgresStore) PutUserInlinePolicy(ctx context.Context, userName, policyName string, document policy_engine.PolicyDocument) error { + if !store.configured { + return fmt.Errorf("store not configured") + } + + docJSON, err := json.Marshal(document) + if err != nil { + glog.Errorf("credential postgres: PutUserInlinePolicy marshal failed user=%s policy=%s: %v", userName, policyName, err) + return fmt.Errorf("failed to marshal policy document: %w", err) + } + + _, err = store.db.ExecContext(ctx, + `INSERT INTO user_inline_policies (username, policy_name, document) + VALUES ($1, $2, $3) + ON CONFLICT (username, policy_name) + DO UPDATE SET document = $3, updated_at = CURRENT_TIMESTAMP`, + userName, policyName, string(docJSON)) + if err != nil { + glog.Errorf("credential postgres: PutUserInlinePolicy failed user=%s policy=%s: %v", userName, policyName, err) + return fmt.Errorf("failed to upsert inline policy: %w", err) + } + + glog.V(0).Infof("credential postgres: PutUserInlinePolicy user=%s policy=%s", userName, policyName) + return nil +} + +func (store *PostgresStore) GetUserInlinePolicy(ctx context.Context, userName, policyName string) (*policy_engine.PolicyDocument, error) { + if !store.configured { + return nil, fmt.Errorf("store not configured") + } + + var docJSON []byte + err := store.db.QueryRowContext(ctx, + "SELECT document FROM user_inline_policies WHERE username = $1 AND policy_name = $2", + userName, policyName).Scan(&docJSON) + if err != nil { + if err == sql.ErrNoRows { + glog.V(2).Infof("credential postgres: GetUserInlinePolicy user=%s policy=%s not found", userName, policyName) + return nil, nil + } + glog.Errorf("credential postgres: GetUserInlinePolicy query failed user=%s policy=%s: %v", userName, policyName, err) + return nil, fmt.Errorf("failed to query inline policy: %w", err) + } + + var doc policy_engine.PolicyDocument + if err := json.Unmarshal(docJSON, &doc); err != nil { + return nil, fmt.Errorf("failed to unmarshal inline policy: %w", err) + } + + glog.V(2).Infof("credential postgres: GetUserInlinePolicy user=%s policy=%s found", userName, policyName) + return &doc, nil +} + +func (store *PostgresStore) DeleteUserInlinePolicy(ctx context.Context, userName, policyName string) error { + if !store.configured { + return fmt.Errorf("store not configured") + } + + result, err := store.db.ExecContext(ctx, + "DELETE FROM user_inline_policies WHERE username = $1 AND policy_name = $2", + userName, policyName) + if err != nil { + glog.Errorf("credential postgres: DeleteUserInlinePolicy failed user=%s policy=%s: %v", userName, policyName, err) + return fmt.Errorf("failed to delete inline policy: %w", err) + } + + rowsAffected, _ := result.RowsAffected() + glog.V(0).Infof("credential postgres: DeleteUserInlinePolicy user=%s policy=%s deleted=%d", userName, policyName, rowsAffected) + return nil +} + +func (store *PostgresStore) ListUserInlinePolicies(ctx context.Context, userName string) ([]string, error) { + if !store.configured { + return nil, fmt.Errorf("store not configured") + } + + rows, err := store.db.QueryContext(ctx, + "SELECT policy_name FROM user_inline_policies WHERE username = $1 ORDER BY policy_name", + userName) + if err != nil { + glog.Errorf("credential postgres: ListUserInlinePolicies query failed user=%s: %v", userName, err) + return nil, fmt.Errorf("failed to query inline policies: %w", err) + } + defer rows.Close() + + var names []string + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + return nil, fmt.Errorf("failed to scan policy name: %w", err) + } + names = append(names, name) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed iterating inline policy rows: %w", err) + } + + glog.V(1).Infof("credential postgres: ListUserInlinePolicies user=%s count=%d", userName, len(names)) + return names, nil +} + + +func (store *PostgresStore) LoadInlinePolicies(ctx context.Context) (map[string]map[string]policy_engine.PolicyDocument, error) { + if !store.configured { + return nil, fmt.Errorf("store not configured") + } + + rows, err := store.db.QueryContext(ctx, + "SELECT username, policy_name, document FROM user_inline_policies ORDER BY username, policy_name") + if err != nil { + glog.Errorf("credential postgres: LoadInlinePolicies query failed: %v", err) + return nil, fmt.Errorf("failed to query inline policies: %w", err) + } + defer rows.Close() + + result := make(map[string]map[string]policy_engine.PolicyDocument) + count := 0 + for rows.Next() { + var username, policyName string + var docJSON []byte + if err := rows.Scan(&username, &policyName, &docJSON); err != nil { + return nil, fmt.Errorf("failed to scan inline policy row: %w", err) + } + + var doc policy_engine.PolicyDocument + if err := json.Unmarshal(docJSON, &doc); err != nil { + glog.Warningf("credential postgres: LoadInlinePolicies unmarshal failed user=%s policy=%s: %v", username, policyName, err) + continue + } + + if result[username] == nil { + result[username] = make(map[string]policy_engine.PolicyDocument) + } + result[username][policyName] = doc + count++ + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed iterating inline policy rows: %w", err) + } + + glog.V(0).Infof("credential postgres: LoadInlinePolicies loaded %d policies for %d users", count, len(result)) + return result, nil +} diff --git a/weed/credential/postgres/postgres_store.go b/weed/credential/postgres/postgres_store.go index 3ba4810d6..0f7a43f28 100644 --- a/weed/credential/postgres/postgres_store.go +++ b/weed/credential/postgres/postgres_store.go @@ -6,6 +6,7 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/credential" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/util" _ "github.com/jackc/pgx/v5/stdlib" @@ -37,6 +38,10 @@ func (store *PostgresStore) Initialize(configuration util.Configuration, prefix database := configuration.GetString(prefix + "database") schema := configuration.GetString(prefix + "schema") sslmode := configuration.GetString(prefix + "sslmode") + sslcert := configuration.GetString(prefix + "sslcert") + sslkey := configuration.GetString(prefix + "sslkey") + sslrootcert := configuration.GetString(prefix + "sslrootcert") + pgbouncerCompatible := configuration.GetBool(prefix + "pgbouncer_compatible") // Set defaults if hostname == "" { @@ -45,48 +50,64 @@ func (store *PostgresStore) Initialize(configuration util.Configuration, prefix if port == 0 { port = 5432 } - if schema == "" { - schema = "public" - } if sslmode == "" { sslmode = "disable" } - // Build pgx-optimized connection string - // Note: prefer_simple_protocol=true is only needed for PgBouncer, not direct PostgreSQL connections - connStr := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=%s search_path=%s", - hostname, port, username, password, database, sslmode, schema) + glog.V(0).Infof("credential postgres: initializing store host=%s port=%d user=%s db=%s sslmode=%s pgbouncer=%v", + hostname, port, username, database, sslmode, pgbouncerCompatible) + + connStr := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=%s", + hostname, port, username, password, database, sslmode) + if schema != "" { + connStr += fmt.Sprintf(" search_path=%s", schema) + } + if sslcert != "" { + connStr += fmt.Sprintf(" sslcert=%s", sslcert) + } + if sslkey != "" { + connStr += fmt.Sprintf(" sslkey=%s", sslkey) + } + if sslrootcert != "" { + connStr += fmt.Sprintf(" sslrootcert=%s", sslrootcert) + } + if pgbouncerCompatible { + connStr += " default_query_exec_mode=simple_protocol" + } db, err := sql.Open("pgx", connStr) if err != nil { + glog.Errorf("credential postgres: failed to open database: %v", err) return fmt.Errorf("failed to open database: %w", err) } - // Test connection if err := db.Ping(); err != nil { db.Close() + glog.Errorf("credential postgres: failed to ping database: %v", err) return fmt.Errorf("failed to ping database: %w", err) } - // Set connection pool settings + glog.V(0).Infof("credential postgres: connection established") + db.SetMaxOpenConns(25) db.SetMaxIdleConns(5) db.SetConnMaxLifetime(5 * time.Minute) store.db = db - // Create tables if they don't exist if err := store.createTables(); err != nil { db.Close() + glog.Errorf("credential postgres: failed to create tables: %v", err) return fmt.Errorf("failed to create tables: %w", err) } + glog.V(0).Infof("credential postgres: tables verified, store ready") + store.configured = true return nil } func (store *PostgresStore) createTables() error { - // Create users table usersTable := ` CREATE TABLE IF NOT EXISTS users ( username VARCHAR(255) PRIMARY KEY, @@ -100,12 +121,10 @@ func (store *PostgresStore) createTables() error { CREATE INDEX IF NOT EXISTS idx_users_email ON users(email); ` - // Migration: Add policy_names column if it doesn't exist (for existing installations) addPolicyNamesColumn := ` ALTER TABLE users ADD COLUMN IF NOT EXISTS policy_names JSONB DEFAULT '[]'; ` - // Create credentials table credentialsTable := ` CREATE TABLE IF NOT EXISTS credentials ( id SERIAL PRIMARY KEY, @@ -118,7 +137,6 @@ func (store *PostgresStore) createTables() error { CREATE INDEX IF NOT EXISTS idx_credentials_access_key ON credentials(access_key); ` - // Create policies table policiesTable := ` CREATE TABLE IF NOT EXISTS policies ( name VARCHAR(255) PRIMARY KEY, @@ -129,7 +147,6 @@ func (store *PostgresStore) createTables() error { CREATE INDEX IF NOT EXISTS idx_policies_name ON policies(name); ` - // Create service_accounts table serviceAccountsTable := ` CREATE TABLE IF NOT EXISTS service_accounts ( id VARCHAR(255) PRIMARY KEY, @@ -140,7 +157,18 @@ func (store *PostgresStore) createTables() error { ); ` - // Create groups table + inlinePoliciesTable := ` + CREATE TABLE IF NOT EXISTS user_inline_policies ( + username VARCHAR(255) REFERENCES users(username) ON DELETE CASCADE, + policy_name VARCHAR(255) NOT NULL, + document JSONB NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (username, policy_name) + ); + CREATE INDEX IF NOT EXISTS idx_user_inline_policies_username ON user_inline_policies(username); + ` + groupsTable := ` CREATE TABLE IF NOT EXISTS groups ( name VARCHAR(255) PRIMARY KEY, @@ -152,12 +180,10 @@ func (store *PostgresStore) createTables() error { ); ` - // Execute table creation if _, err := store.db.Exec(usersTable); err != nil { return fmt.Errorf("failed to create users table: %w", err) } - // Run migration to add policy_names column for existing installations if _, err := store.db.Exec(addPolicyNamesColumn); err != nil { return fmt.Errorf("failed to add policy_names column: %w", err) } @@ -174,17 +200,19 @@ func (store *PostgresStore) createTables() error { return fmt.Errorf("failed to create service_accounts table: %w", err) } + if _, err := store.db.Exec(inlinePoliciesTable); err != nil { + return fmt.Errorf("failed to create user_inline_policies table: %w", err) + } + if _, err := store.db.Exec(groupsTable); err != nil { return fmt.Errorf("failed to create groups table: %w", err) } - // Create index on groups disabled column for filtering groupsDisabledIndex := `CREATE INDEX IF NOT EXISTS idx_groups_disabled ON groups (disabled);` if _, err := store.db.Exec(groupsDisabledIndex); err != nil { return fmt.Errorf("failed to create groups disabled index: %w", err) } - // Create GIN index on groups members JSONB for membership lookups groupsMembersIndex := `CREATE INDEX IF NOT EXISTS idx_groups_members_gin ON groups USING GIN (members);` if _, err := store.db.Exec(groupsMembersIndex); err != nil { return fmt.Errorf("failed to create groups members index: %w", err) @@ -195,6 +223,7 @@ func (store *PostgresStore) createTables() error { func (store *PostgresStore) Shutdown() { if store.db != nil { + glog.V(0).Infof("credential postgres: shutting down") store.db.Close() store.db = nil } diff --git a/weed/credential/test/inline_policy_test.go b/weed/credential/test/inline_policy_test.go new file mode 100644 index 000000000..f5fa3f2ab --- /dev/null +++ b/weed/credential/test/inline_policy_test.go @@ -0,0 +1,170 @@ +package test + +import ( + "context" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/credential" + "github.com/seaweedfs/seaweedfs/weed/credential/memory" + "github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine" + + _ "github.com/seaweedfs/seaweedfs/weed/credential/filer_etc" + _ "github.com/seaweedfs/seaweedfs/weed/credential/memory" + _ "github.com/seaweedfs/seaweedfs/weed/credential/postgres" +) + +func TestInlinePolicyOperations(t *testing.T) { + ctx := context.Background() + + credentialManager, err := credential.NewCredentialManager(credential.StoreTypeMemory, nil, "") + if err != nil { + t.Fatalf("Failed to create credential manager: %v", err) + } + + store, ok := credentialManager.GetStore().(*memory.MemoryStore) + if !ok { + t.Fatal("Store is not a memory store") + } + + userName := "testuser" + policyName := "read-bucket" + doc := policy_engine.PolicyDocument{ + Version: "2012-10-17", + Statement: []policy_engine.PolicyStatement{ + { + Effect: policy_engine.PolicyEffectAllow, + Action: policy_engine.NewStringOrStringSlice("s3:GetObject"), + Resource: policy_engine.NewStringOrStringSlicePtr("arn:aws:s3:::test-bucket/*"), + }, + }, + } + + // Put + if err := store.PutUserInlinePolicy(ctx, userName, policyName, doc); err != nil { + t.Fatalf("PutUserInlinePolicy failed: %v", err) + } + + // Get + got, err := store.GetUserInlinePolicy(ctx, userName, policyName) + if err != nil { + t.Fatalf("GetUserInlinePolicy failed: %v", err) + } + if got == nil { + t.Fatal("GetUserInlinePolicy returned nil") + } + if got.Version != "2012-10-17" { + t.Errorf("Expected version '2012-10-17', got '%s'", got.Version) + } + if len(got.Statement) != 1 { + t.Errorf("Expected 1 statement, got %d", len(got.Statement)) + } + + // Get non-existent + missing, err := store.GetUserInlinePolicy(ctx, userName, "no-such-policy") + if err != nil { + t.Fatalf("GetUserInlinePolicy for missing policy failed: %v", err) + } + if missing != nil { + t.Error("Expected nil for non-existent policy") + } + + // Put second policy, same user + doc2 := policy_engine.PolicyDocument{ + Version: "2012-10-17", + Statement: []policy_engine.PolicyStatement{ + { + Effect: policy_engine.PolicyEffectAllow, + Action: policy_engine.NewStringOrStringSlice("s3:PutObject"), + Resource: policy_engine.NewStringOrStringSlicePtr("arn:aws:s3:::other-bucket/*"), + }, + }, + } + if err := store.PutUserInlinePolicy(ctx, userName, "write-bucket", doc2); err != nil { + t.Fatalf("PutUserInlinePolicy second policy failed: %v", err) + } + + // List + names, err := store.ListUserInlinePolicies(ctx, userName) + if err != nil { + t.Fatalf("ListUserInlinePolicies failed: %v", err) + } + if len(names) != 2 { + t.Errorf("Expected 2 policies, got %d", len(names)) + } + + // List for non-existent user + emptyNames, err := store.ListUserInlinePolicies(ctx, "nobody") + if err != nil { + t.Fatalf("ListUserInlinePolicies for missing user failed: %v", err) + } + if len(emptyNames) != 0 { + t.Errorf("Expected 0 policies for missing user, got %d", len(emptyNames)) + } + + // LoadInlinePolicies (bulk) + all, err := store.LoadInlinePolicies(ctx) + if err != nil { + t.Fatalf("LoadInlinePolicies failed: %v", err) + } + if len(all) != 1 { + t.Errorf("Expected 1 user in LoadInlinePolicies, got %d", len(all)) + } + if len(all[userName]) != 2 { + t.Errorf("Expected 2 policies for user in LoadInlinePolicies, got %d", len(all[userName])) + } + + // Overwrite existing policy (upsert) + updatedDoc := policy_engine.PolicyDocument{ + Version: "2012-10-17", + Statement: []policy_engine.PolicyStatement{ + { + Effect: policy_engine.PolicyEffectDeny, + Action: policy_engine.NewStringOrStringSlice("s3:GetObject"), + Resource: policy_engine.NewStringOrStringSlicePtr("arn:aws:s3:::test-bucket/secret/*"), + }, + }, + } + if err := store.PutUserInlinePolicy(ctx, userName, policyName, updatedDoc); err != nil { + t.Fatalf("PutUserInlinePolicy overwrite failed: %v", err) + } + overwritten, err := store.GetUserInlinePolicy(ctx, userName, policyName) + if err != nil { + t.Fatalf("GetUserInlinePolicy after overwrite failed: %v", err) + } + if overwritten.Statement[0].Effect != policy_engine.PolicyEffectDeny { + t.Errorf("Expected Deny after overwrite, got %s", overwritten.Statement[0].Effect) + } + + // Delete one policy + if err := store.DeleteUserInlinePolicy(ctx, userName, policyName); err != nil { + t.Fatalf("DeleteUserInlinePolicy failed: %v", err) + } + deleted, err := store.GetUserInlinePolicy(ctx, userName, policyName) + if err != nil { + t.Fatalf("GetUserInlinePolicy after delete failed: %v", err) + } + if deleted != nil { + t.Error("Expected nil after delete") + } + + // Remaining policy still there + remaining, err := store.ListUserInlinePolicies(ctx, userName) + if err != nil { + t.Fatalf("ListUserInlinePolicies after delete failed: %v", err) + } + if len(remaining) != 1 { + t.Errorf("Expected 1 remaining policy, got %d", len(remaining)) + } + + // Delete last policy — user entry should be cleaned up + if err := store.DeleteUserInlinePolicy(ctx, userName, "write-bucket"); err != nil { + t.Fatalf("DeleteUserInlinePolicy last policy failed: %v", err) + } + allAfter, err := store.LoadInlinePolicies(ctx) + if err != nil { + t.Fatalf("LoadInlinePolicies after full cleanup failed: %v", err) + } + if len(allAfter) != 0 { + t.Errorf("Expected empty LoadInlinePolicies after cleanup, got %d users", len(allAfter)) + } +}