mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-20 22:27:04 +00:00
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
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user