fix(auth): 修复生产登录卡死与有界超时 #11

Merged
chengcheng merged 2 commits from codex/fix-api-key-pool-deadlock into main 2026-07-21 12:09:40 +08:00
12 changed files with 653 additions and 54 deletions
+6
View File
@@ -5208,6 +5208,12 @@
"schema": {
"$ref": "#/definitions/httpapi.ErrorEnvelope"
}
},
"503": {
"description": "Service Unavailable",
"schema": {
"$ref": "#/definitions/httpapi.ErrorEnvelope"
}
}
}
}
+4
View File
@@ -6441,6 +6441,10 @@ paths:
description: Internal Server Error
schema:
$ref: '#/definitions/httpapi.ErrorEnvelope'
"503":
description: Service Unavailable
schema:
$ref: '#/definitions/httpapi.ErrorEnvelope'
summary: 本地登录
tags:
- auth
@@ -0,0 +1,111 @@
package httpapi
import (
"bytes"
"context"
"encoding/json"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"net/url"
"os"
"strings"
"testing"
"time"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
)
func TestReadyReturnsPostgresUnavailableWithinTwoSeconds(t *testing.T) {
db := newExhaustedPostgresStore(t)
server := &Server{store: db, logger: slog.New(slog.NewJSONHandler(io.Discard, nil))}
requestContext, cancel := context.WithTimeout(context.Background(), 4*time.Second)
defer cancel()
request := httptest.NewRequest(http.MethodGet, "/readyz", nil).WithContext(requestContext)
recorder := httptest.NewRecorder()
startedAt := time.Now()
server.ready(recorder, request)
elapsed := time.Since(startedAt)
assertUnavailableResponse(t, recorder, "POSTGRES_UNAVAILABLE", "postgres unavailable")
if elapsed > 3*time.Second {
t.Fatalf("readiness timeout took %s, want no more than 3s", elapsed)
}
}
func TestLoginReturnsAuthStoreUnavailableWithinFiveSeconds(t *testing.T) {
db := newExhaustedPostgresStore(t)
var logs bytes.Buffer
server := &Server{store: db, logger: slog.New(slog.NewJSONHandler(&logs, nil))}
requestContext, cancel := context.WithTimeout(context.Background(), 7*time.Second)
defer cancel()
request := httptest.NewRequest(http.MethodPost, "/api/v1/auth/login", strings.NewReader(`{"account":"timeout-test-account","password":"timeout-test-password"}`)).WithContext(requestContext)
recorder := httptest.NewRecorder()
startedAt := time.Now()
server.login(recorder, request)
elapsed := time.Since(startedAt)
assertUnavailableResponse(t, recorder, "AUTH_STORE_UNAVAILABLE", "authentication service temporarily unavailable")
if elapsed > 6*time.Second {
t.Fatalf("login timeout took %s, want no more than 6s", elapsed)
}
logOutput := logs.String()
for _, field := range []string{"postgres_pool_max_connections", "postgres_pool_acquired_connections", "postgres_pool_idle_connections", "postgres_pool_empty_acquire_count", "postgres_pool_canceled_acquire_count"} {
if !strings.Contains(logOutput, field) {
t.Fatalf("login failure log did not include %q: %s", field, logOutput)
}
}
if strings.Contains(logOutput, "timeout-test-account") || strings.Contains(logOutput, "timeout-test-password") {
t.Fatalf("login failure log exposed credentials: %s", logOutput)
}
}
func newExhaustedPostgresStore(t *testing.T) *store.Store {
t.Helper()
databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL"))
if databaseURL == "" {
t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run PostgreSQL availability timeout tests")
}
parsed, err := url.Parse(databaseURL)
if err != nil {
t.Fatalf("parse test database URL: %v", err)
}
query := parsed.Query()
query.Set("pool_max_conns", "1")
query.Set("pool_min_conns", "0")
parsed.RawQuery = query.Encode()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
db, err := store.Connect(ctx, parsed.String())
if err != nil {
t.Fatalf("connect timeout test store: %v", err)
}
t.Cleanup(db.Close)
connection, err := db.Pool().Acquire(ctx)
if err != nil {
t.Fatalf("exhaust timeout test pool: %v", err)
}
t.Cleanup(connection.Release)
return db
}
func assertUnavailableResponse(t *testing.T, recorder *httptest.ResponseRecorder, expectedCode, expectedMessage string) {
t.Helper()
if recorder.Code != http.StatusServiceUnavailable {
t.Fatalf("status = %d, want 503; body=%s", recorder.Code, recorder.Body.String())
}
var envelope ErrorEnvelope
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
t.Fatalf("decode unavailable response: %v", err)
}
if envelope.Error.Code != expectedCode {
t.Fatalf("error code = %q, want %q; body=%s", envelope.Error.Code, expectedCode, recorder.Body.String())
}
if envelope.Error.Message != expectedMessage {
t.Fatalf("error message = %q, want %q; body=%s", envelope.Error.Message, expectedMessage, recorder.Body.String())
}
}
+38 -3
View File
@@ -17,6 +17,14 @@ import (
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
)
const (
postgresReadinessTimeout = 2 * time.Second
localLoginStoreTimeout = 5 * time.Second
errorCodePostgresDown = "POSTGRES_UNAVAILABLE"
errorCodeAuthStoreDown = "AUTH_STORE_UNAVAILABLE"
authStoreUnavailableError = "authentication service temporarily unavailable"
)
// health godoc
// @Summary 健康检查
// @Description 返回服务进程、运行环境和身份模式,供负载均衡或人工排障使用。
@@ -42,8 +50,11 @@ func (s *Server) health(w http.ResponseWriter, r *http.Request) {
// @Failure 503 {object} ErrorEnvelope
// @Router /readyz [get]
func (s *Server) ready(w http.ResponseWriter, r *http.Request) {
if err := s.store.Ping(r.Context()); err != nil {
writeError(w, http.StatusServiceUnavailable, "postgres unavailable")
ctx, cancel := context.WithTimeout(r.Context(), postgresReadinessTimeout)
defer cancel()
if err := s.store.Ping(ctx); err != nil {
s.logPostgresUnavailable("postgres readiness check failed")
writeError(w, http.StatusServiceUnavailable, "postgres unavailable", errorCodePostgresDown)
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
@@ -121,6 +132,7 @@ func (s *Server) register(w http.ResponseWriter, r *http.Request) {
// @Failure 401 {object} ErrorEnvelope
// @Failure 403 {object} ErrorEnvelope
// @Failure 500 {object} ErrorEnvelope
// @Failure 503 {object} ErrorEnvelope
// @Router /api/v1/auth/login [post]
func (s *Server) login(w http.ResponseWriter, r *http.Request) {
var input store.LocalLoginInput
@@ -128,12 +140,19 @@ func (s *Server) login(w http.ResponseWriter, r *http.Request) {
writeError(w, http.StatusBadRequest, "invalid json body")
return
}
user, err := s.store.AuthenticateLocalUser(r.Context(), input)
ctx, cancel := context.WithTimeout(r.Context(), localLoginStoreTimeout)
defer cancel()
user, err := s.store.AuthenticateLocalUser(ctx, input)
if err != nil {
if errors.Is(err, store.ErrInvalidCredentials) {
writeError(w, http.StatusUnauthorized, "invalid account or password")
return
}
if store.IsPostgresUnavailable(err) {
s.logPostgresUnavailable("login authentication store unavailable")
writeError(w, http.StatusServiceUnavailable, authStoreUnavailableError, errorCodeAuthStoreDown)
return
}
s.logger.Error("login local user failed", "error", err)
writeError(w, http.StatusInternalServerError, "login failed")
return
@@ -145,6 +164,22 @@ func (s *Server) login(w http.ResponseWriter, r *http.Request) {
s.writeAuthResponse(w, http.StatusOK, user)
}
func (s *Server) logPostgresUnavailable(message string) {
if s.logger == nil || s.store == nil || s.store.Pool() == nil {
return
}
statistics := s.store.Pool().Stat()
s.logger.Error(message,
"error_category", "postgres_unavailable",
"postgres_pool_max_connections", statistics.MaxConns(),
"postgres_pool_total_connections", statistics.TotalConns(),
"postgres_pool_acquired_connections", statistics.AcquiredConns(),
"postgres_pool_idle_connections", statistics.IdleConns(),
"postgres_pool_empty_acquire_count", statistics.EmptyAcquireCount(),
"postgres_pool_canceled_acquire_count", statistics.CanceledAcquireCount(),
)
}
func (s *Server) localIdentityEnabled() bool {
mode := strings.ToLower(strings.TrimSpace(s.cfg.IdentityMode))
return mode == "" || mode == "standalone" || mode == "hybrid"
@@ -0,0 +1,99 @@
package store
import (
"context"
"sync"
"testing"
"time"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
"github.com/jackc/pgx/v5/pgxpool"
)
func TestVerifyLocalAPIKeyWorksWithSingleConnectionPool(t *testing.T) {
db, verificationStore, created, user := newLocalAPIKeyVerificationFixture(t, 1)
ctx := context.Background()
verifyCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
defer cancel()
verified, err := verificationStore.VerifyLocalAPIKey(verifyCtx, created.Secret)
if err != nil {
t.Fatalf("verify local API key with one connection: %v", err)
}
if verified.APIKeyID != created.APIKey.ID || verified.GatewayUserID != user.ID {
t.Fatalf("verified identity = %+v, want API key %q and user %q", verified, created.APIKey.ID, user.ID)
}
var lastUsedAt *time.Time
if err := db.pool.QueryRow(ctx, `SELECT last_used_at FROM gateway_api_keys WHERE id=$1::uuid`, created.APIKey.ID).Scan(&lastUsedAt); err != nil {
t.Fatalf("read API key last_used_at: %v", err)
}
if lastUsedAt == nil {
t.Fatal("successful API key verification did not update last_used_at")
}
}
func TestVerifyLocalAPIKeyHandlesEightConcurrentRequestsWithFourConnections(t *testing.T) {
_, verificationStore, created, _ := newLocalAPIKeyVerificationFixture(t, 4)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
const requestCount = 8
start := make(chan struct{})
errorsByRequest := make(chan error, requestCount)
var requests sync.WaitGroup
requests.Add(requestCount)
for range requestCount {
go func() {
defer requests.Done()
<-start
_, err := verificationStore.VerifyLocalAPIKey(ctx, created.Secret)
errorsByRequest <- err
}()
}
close(start)
requests.Wait()
close(errorsByRequest)
for err := range errorsByRequest {
if err != nil {
t.Fatalf("concurrent API key verification failed: %v", err)
}
}
if acquired := verificationStore.pool.Stat().AcquiredConns(); acquired != 0 {
t.Fatalf("API key verification left %d connections acquired", acquired)
}
}
func newLocalAPIKeyVerificationFixture(t *testing.T, maxConnections int32) (*Store, *Store, CreatedAPIKey, GatewayUser) {
t.Helper()
db := newIdentityPairingPostgresTestStore(t)
ctx := context.Background()
user, err := db.RegisterLocalUser(ctx, LocalRegisterInput{
Username: "api-key-verification-user",
Password: "api-key-verification-password",
})
if err != nil {
t.Fatalf("register API key verification user: %v", err)
}
created, err := db.CreateAPIKey(ctx, CreateAPIKeyInput{Name: "API key verification fixture"}, &auth.User{
ID: user.ID,
GatewayUserID: user.ID,
GatewayTenantID: user.GatewayTenantID,
TenantID: user.TenantID,
TenantKey: user.TenantKey,
})
if err != nil {
t.Fatalf("create API key verification fixture: %v", err)
}
config := db.pool.Config()
config.MaxConns = maxConnections
config.MinConns = 0
pool, err := pgxpool.NewWithConfig(ctx, config)
if err != nil {
t.Fatalf("create verification pool with %d connections: %v", maxConnections, err)
}
t.Cleanup(pool.Close)
return db, &Store{pool: pool}, created, user
}
@@ -0,0 +1,169 @@
package store
import (
"context"
"errors"
"testing"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
"golang.org/x/crypto/bcrypt"
)
func TestVerifyLocalAPIKeyClosesCandidateRowsBeforeUpdatingUsage(t *testing.T) {
secret := "sk-gw-matching-secret"
wrongHash, err := bcrypt.GenerateFromPassword([]byte("sk-gw-different-secret"), bcrypt.MinCost)
if err != nil {
t.Fatalf("hash non-matching API key: %v", err)
}
matchingHash, err := bcrypt.GenerateFromPassword([]byte(secret), bcrypt.MinCost)
if err != nil {
t.Fatalf("hash matching API key: %v", err)
}
rows := &fakeLocalAPIKeyRows{candidates: []localAPIKeyCandidate{
{apiKeyID: "wrong-key", hash: string(wrongHash), keyPrefix: apiKeyPrefix(secret)},
{
apiKeyID: "matching-key",
hash: string(matchingHash),
keyPrefix: apiKeyPrefix(secret),
keyName: "Matching key",
scopesBytes: []byte(`["chat"]`),
userGroupID: "group-id",
gatewayUserID: "user-id",
username: "api-key-user",
rolesBytes: []byte(`["user"]`),
gatewayTenantID: "gateway-tenant-id",
tenantID: "tenant-id",
tenantKey: "tenant-key",
},
}}
database := &fakeLocalAPIKeyDatabase{rows: rows}
user, err := verifyLocalAPIKey(context.Background(), database, secret)
if err != nil {
t.Fatalf("verify local API key: %v", err)
}
if !rows.closed {
t.Fatal("candidate rows remained open after API key verification")
}
if database.updatedAPIKeyID != "matching-key" {
t.Fatalf("updated API key = %q, want matching-key", database.updatedAPIKeyID)
}
if user.APIKeyID != "matching-key" || user.GatewayUserID != "user-id" {
t.Fatalf("verified user = %+v", user)
}
}
func TestVerifyLocalAPIKeyReturnsUnauthorizedAfterClosingCandidateRows(t *testing.T) {
secret := "sk-gw-unknown-secret"
wrongHash, err := bcrypt.GenerateFromPassword([]byte("sk-gw-different-secret"), bcrypt.MinCost)
if err != nil {
t.Fatalf("hash non-matching API key: %v", err)
}
rows := &fakeLocalAPIKeyRows{candidates: []localAPIKeyCandidate{{
apiKeyID: "wrong-key",
hash: string(wrongHash),
}}}
database := &fakeLocalAPIKeyDatabase{rows: rows}
_, err = verifyLocalAPIKey(context.Background(), database, secret)
if !errors.Is(err, auth.ErrUnauthorized) {
t.Fatalf("verify error = %v, want unauthorized", err)
}
if !rows.closed {
t.Fatal("candidate rows remained open after unsuccessful API key verification")
}
if database.updatedAPIKeyID != "" {
t.Fatalf("unexpected API key usage update for %q", database.updatedAPIKeyID)
}
}
type fakeLocalAPIKeyDatabase struct {
rows *fakeLocalAPIKeyRows
updatedAPIKeyID string
}
func (database *fakeLocalAPIKeyDatabase) Query(context.Context, string, ...any) (pgx.Rows, error) {
return database.rows, nil
}
func (database *fakeLocalAPIKeyDatabase) Exec(_ context.Context, _ string, arguments ...any) (pgconn.CommandTag, error) {
if !database.rows.closed {
return pgconn.CommandTag{}, errors.New("API key usage update started before candidate rows closed")
}
database.updatedAPIKeyID, _ = arguments[0].(string)
return pgconn.NewCommandTag("UPDATE 1"), nil
}
type fakeLocalAPIKeyRows struct {
candidates []localAPIKeyCandidate
current int
closed bool
}
func (rows *fakeLocalAPIKeyRows) Close() {
rows.closed = true
}
func (rows *fakeLocalAPIKeyRows) Err() error {
return nil
}
func (rows *fakeLocalAPIKeyRows) CommandTag() pgconn.CommandTag {
return pgconn.CommandTag{}
}
func (rows *fakeLocalAPIKeyRows) FieldDescriptions() []pgconn.FieldDescription {
return nil
}
func (rows *fakeLocalAPIKeyRows) Next() bool {
if rows.current >= len(rows.candidates) {
rows.Close()
return false
}
rows.current++
return true
}
func (rows *fakeLocalAPIKeyRows) Scan(destinations ...any) error {
candidate := rows.candidates[rows.current-1]
values := []any{
candidate.apiKeyID,
candidate.hash,
candidate.keyPrefix,
candidate.keyName,
candidate.scopesBytes,
candidate.userGroupID,
candidate.gatewayUserID,
candidate.username,
candidate.rolesBytes,
candidate.gatewayTenantID,
candidate.tenantID,
candidate.tenantKey,
}
for index, value := range values {
switch destination := destinations[index].(type) {
case *string:
*destination = value.(string)
case *[]byte:
*destination = value.([]byte)
default:
return errors.New("unsupported fake row destination")
}
}
return nil
}
func (rows *fakeLocalAPIKeyRows) Values() ([]any, error) {
return nil, errors.New("not implemented")
}
func (rows *fakeLocalAPIKeyRows) RawValues() [][]byte {
return nil
}
func (rows *fakeLocalAPIKeyRows) Conn() *pgx.Conn {
return nil
}
@@ -685,11 +685,13 @@ func newIdentityPairingPostgresTestStore(t *testing.T) *Store {
migrationDirectory := filepath.Join(filepath.Dir(filename), "..", "..", "migrations")
for _, migrationName := range []string{
"0001_init.sql",
"0017_task_record_enrichment.sql",
"0061_oidc_server_sessions.sql",
"0065_identity_configuration_revisions.sql",
"0066_identity_onboarding_exchanges.sql",
"0067_identity_secret_cleanup_queue.sql",
"0068_identity_pairing_start_reservation.sql",
"0069_billing_correctness_v2.sql",
} {
migration, err := os.ReadFile(filepath.Join(migrationDirectory, migrationName))
if err != nil {
+104 -36
View File
@@ -7,6 +7,7 @@ import (
"encoding/base64"
"encoding/json"
"errors"
"net"
"strings"
"time"
"unicode"
@@ -22,6 +23,11 @@ type Store struct {
pool *pgxpool.Pool
}
const (
postgresApplicationName = "easyai-ai-gateway"
postgresConnectTimeout = 5 * time.Second
)
func defaultAPIKeyScopes() []string {
return []string{"chat", "embedding", "rerank", "image", "video", "music", "audio", "voice_clone"}
}
@@ -67,7 +73,11 @@ var (
)
func Connect(ctx context.Context, databaseURL string) (*Store, error) {
pool, err := pgxpool.New(ctx, databaseURL)
config, err := postgresPoolConfig(databaseURL)
if err != nil {
return nil, err
}
pool, err := pgxpool.NewWithConfig(ctx, config)
if err != nil {
return nil, err
}
@@ -78,6 +88,38 @@ func Connect(ctx context.Context, databaseURL string) (*Store, error) {
return &Store{pool: pool}, nil
}
func postgresPoolConfig(databaseURL string) (*pgxpool.Config, error) {
config, err := pgxpool.ParseConfig(databaseURL)
if err != nil {
return nil, err
}
config.ConnConfig.ConnectTimeout = postgresConnectTimeout
config.ConnConfig.RuntimeParams["application_name"] = postgresApplicationName
return config, nil
}
func IsPostgresUnavailable(err error) bool {
if err == nil {
return false
}
if errors.Is(err, context.DeadlineExceeded) {
return true
}
var connectError *pgconn.ConnectError
if errors.As(err, &connectError) {
return true
}
var networkError net.Error
if errors.As(err, &networkError) {
return true
}
var postgresError *pgconn.PgError
if errors.As(err, &postgresError) {
return strings.HasPrefix(postgresError.Code, "08") || postgresError.Code == "53300" || strings.HasPrefix(postgresError.Code, "57P0")
}
return pgconn.SafeToRetry(err)
}
func (s *Store) Close() {
s.pool.Close()
}
@@ -1519,11 +1561,35 @@ WHERE subject_type = 'api_key' AND subject_id = $1::uuid`, apiKeyID); err != nil
}
func (s *Store) VerifyLocalAPIKey(ctx context.Context, secret string) (*auth.User, error) {
return verifyLocalAPIKey(ctx, s.pool, secret)
}
type localAPIKeyDatabase interface {
Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
}
type localAPIKeyCandidate struct {
apiKeyID string
hash string
keyPrefix string
keyName string
scopesBytes []byte
userGroupID string
gatewayUserID string
username string
rolesBytes []byte
gatewayTenantID string
tenantID string
tenantKey string
}
func verifyLocalAPIKey(ctx context.Context, database localAPIKeyDatabase, secret string) (*auth.User, error) {
prefix := apiKeyPrefix(secret)
if prefix == "" {
return nil, auth.ErrUnauthorized
}
rows, err := s.pool.Query(ctx, `
rows, err := database.Query(ctx, `
SELECT k.id::text, k.key_hash, k.key_prefix, k.name, k.scopes, COALESCE(k.user_group_id::text, u.default_user_group_id::text, ''),
u.id::text, u.username, u.roles, COALESCE(u.gateway_tenant_id::text, ''),
COALESCE(u.tenant_id, ''), COALESCE(u.tenant_key, '')
@@ -1538,49 +1604,51 @@ WHERE k.key_prefix = $1
if err != nil {
return nil, err
}
defer rows.Close()
candidates, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (localAPIKeyCandidate, error) {
var candidate localAPIKeyCandidate
err := row.Scan(
&candidate.apiKeyID,
&candidate.hash,
&candidate.keyPrefix,
&candidate.keyName,
&candidate.scopesBytes,
&candidate.userGroupID,
&candidate.gatewayUserID,
&candidate.username,
&candidate.rolesBytes,
&candidate.gatewayTenantID,
&candidate.tenantID,
&candidate.tenantKey,
)
return candidate, err
})
if err != nil {
return nil, err
}
for rows.Next() {
var apiKeyID string
var hash string
var keyPrefix string
var keyName string
var scopesBytes []byte
var userGroupID string
var gatewayUserID string
var username string
var rolesBytes []byte
var gatewayTenantID string
var tenantID string
var tenantKey string
if err := rows.Scan(&apiKeyID, &hash, &keyPrefix, &keyName, &scopesBytes, &userGroupID, &gatewayUserID, &username, &rolesBytes, &gatewayTenantID, &tenantID, &tenantKey); err != nil {
return nil, err
}
if bcrypt.CompareHashAndPassword([]byte(hash), []byte(secret)) != nil {
for _, candidate := range candidates {
if bcrypt.CompareHashAndPassword([]byte(candidate.hash), []byte(secret)) != nil {
continue
}
if _, err := s.pool.Exec(ctx, `UPDATE gateway_api_keys SET last_used_at = now(), updated_at = now() WHERE id = $1::uuid`, apiKeyID); err != nil {
if _, err := database.Exec(ctx, `UPDATE gateway_api_keys SET last_used_at = now(), updated_at = now() WHERE id = $1::uuid`, candidate.apiKeyID); err != nil {
return nil, err
}
return &auth.User{
ID: gatewayUserID,
Username: username,
Roles: decodeStringArray(rolesBytes),
TenantID: tenantID,
GatewayTenantID: gatewayTenantID,
TenantKey: tenantKey,
ID: candidate.gatewayUserID,
Username: candidate.username,
Roles: decodeStringArray(candidate.rolesBytes),
TenantID: candidate.tenantID,
GatewayTenantID: candidate.gatewayTenantID,
TenantKey: candidate.tenantKey,
Source: "gateway",
GatewayUserID: gatewayUserID,
UserGroupID: userGroupID,
APIKeyID: apiKeyID,
APIKeyName: keyName,
APIKeyPrefix: keyPrefix,
APIKeyScopes: decodeStringArray(scopesBytes),
GatewayUserID: candidate.gatewayUserID,
UserGroupID: candidate.userGroupID,
APIKeyID: candidate.apiKeyID,
APIKeyName: candidate.keyName,
APIKeyPrefix: candidate.keyPrefix,
APIKeyScopes: decodeStringArray(candidate.scopesBytes),
}, nil
}
if err := rows.Err(); err != nil {
return nil, err
}
return nil, auth.ErrUnauthorized
}
@@ -0,0 +1,49 @@
package store
import (
"context"
"testing"
"time"
"github.com/jackc/pgx/v5/pgconn"
)
func TestPostgresPoolConfigSetsDiagnosticAndConnectTimeout(t *testing.T) {
config, err := postgresPoolConfig("postgresql://gateway:password@localhost:5432/gateway?sslmode=disable")
if err != nil {
t.Fatalf("parse PostgreSQL pool config: %v", err)
}
if config.ConnConfig.ConnectTimeout != 5*time.Second {
t.Fatalf("connect timeout = %s, want 5s", config.ConnConfig.ConnectTimeout)
}
if applicationName := config.ConnConfig.RuntimeParams["application_name"]; applicationName != "easyai-ai-gateway" {
t.Fatalf("application_name = %q, want easyai-ai-gateway", applicationName)
}
}
func TestPostgresPoolConfigRejectsMalformedURL(t *testing.T) {
if _, err := postgresPoolConfig("://malformed"); err == nil {
t.Fatal("expected malformed PostgreSQL URL to fail")
}
}
func TestIsPostgresUnavailableClassifiesConnectivityFailures(t *testing.T) {
for _, testCase := range []struct {
name string
err error
}{
{name: "deadline", err: context.DeadlineExceeded},
{name: "connection exception", err: &pgconn.PgError{Code: "08006"}},
{name: "too many connections", err: &pgconn.PgError{Code: "53300"}},
{name: "cannot connect now", err: &pgconn.PgError{Code: "57P03"}},
} {
t.Run(testCase.name, func(t *testing.T) {
if !IsPostgresUnavailable(testCase.err) {
t.Fatalf("error %v was not classified as PostgreSQL unavailable", testCase.err)
}
})
}
if IsPostgresUnavailable(&pgconn.PgError{Code: "42601"}) {
t.Fatal("SQL syntax error was incorrectly classified as PostgreSQL unavailable")
}
}
+27
View File
@@ -9,12 +9,39 @@ import {
getAPITask,
getCurrentUser,
getOpsManagementSkillMetadata,
loginLocalAccount,
OIDC_BROWSER_SESSION_CREDENTIAL,
startIdentityPairing,
retireIdentityPairingSecurityEventConflict,
validateIdentityRevision,
} from './api';
describe('local login transport', () => {
afterEach(() => {
vi.useRealTimers();
vi.unstubAllGlobals();
});
it('aborts after ten seconds and returns a stable login timeout message', async () => {
vi.useFakeTimers();
const fetchMock = vi.fn((_url: string, init?: RequestInit) => new Promise<Response>((_resolve, reject) => {
init?.signal?.addEventListener('abort', () => reject(new DOMException('aborted', 'AbortError')));
}));
vi.stubGlobal('fetch', fetchMock);
const login = loginLocalAccount({ account: 'timeout-test-account', password: 'timeout-test-password' });
const rejection = expect(login).rejects.toMatchObject({
message: '登录请求超时,请稍后重试',
});
await vi.advanceTimersByTimeAsync(10_000);
await rejection;
const [, init] = fetchMock.mock.calls[0] as [string, RequestInit];
expect(init.signal).toBeInstanceOf(AbortSignal);
expect(init.signal?.aborted).toBe(true);
});
});
describe('Gateway provisioning errors', () => {
const cases = [
['GATEWAY_USER_NOT_PROVISIONED', '该账号尚未开通 EasyAI Gateway'],
+30 -15
View File
@@ -121,6 +121,7 @@ export async function loginLocalAccount(input: { account: string; password: stri
auth: false,
body: input,
method: 'POST',
timeoutMs: 10_000,
});
}
@@ -1206,7 +1207,7 @@ export async function deleteFileStorageChannel(token: string, channelId: string)
async function request<T>(
path: string,
options: { token?: string; auth?: boolean; method?: string; body?: unknown; headers?: Record<string, string>; signal?: AbortSignal } = {},
options: { token?: string; auth?: boolean; method?: string; body?: unknown; headers?: Record<string, string>; signal?: AbortSignal; timeoutMs?: number } = {},
): Promise<T> {
const headers: Record<string, string> = { ...(options.headers ?? {}) };
if (options.auth !== false && options.token && options.token !== OIDC_BROWSER_SESSION_CREDENTIAL) {
@@ -1215,21 +1216,35 @@ async function request<T>(
if (options.body !== undefined) {
headers['Content-Type'] = 'application/json';
}
const response = await fetch(`${API_BASE}${path}`, {
method: options.method ?? 'GET',
headers,
body: options.body === undefined ? undefined : JSON.stringify(options.body),
credentials: 'include',
signal: options.signal,
});
if (!response.ok) {
const body = await response.text();
throw new GatewayApiError(parseErrorDetails(body, response.status, `Request failed: ${response.status}`));
const controller = options.timeoutMs ? new AbortController() : undefined;
const timeout = controller ? globalThis.setTimeout(() => controller.abort(), options.timeoutMs) : undefined;
const signal = controller && options.signal
? AbortSignal.any([controller.signal, options.signal])
: controller?.signal ?? options.signal;
try {
const response = await fetch(`${API_BASE}${path}`, {
method: options.method ?? 'GET',
headers,
body: options.body === undefined ? undefined : JSON.stringify(options.body),
credentials: 'include',
signal,
});
if (!response.ok) {
const body = await response.text();
throw new GatewayApiError(parseErrorDetails(body, response.status, `Request failed: ${response.status}`));
}
if (response.status === 204) {
return undefined as T;
}
return response.json() as Promise<T>;
} catch (error) {
if (controller?.signal.aborted) {
throw new GatewayApiError('登录请求超时,请稍后重试');
}
throw error;
} finally {
if (timeout !== undefined) globalThis.clearTimeout(timeout);
}
if (response.status === 204) {
return undefined as T;
}
return response.json() as Promise<T>;
}
function authorizationHeader(token: string): Record<string, string> {
+14
View File
@@ -33,6 +33,20 @@ server {
return 404;
}
location = /gateway-api/api/v1/auth/login {
proxy_http_version 1.1;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header Connection "";
proxy_connect_timeout 3s;
proxy_read_timeout 15s;
proxy_send_timeout 15s;
proxy_redirect off;
proxy_pass http://api:8088/api/v1/auth/login;
}
location /gateway-api/ {
proxy_http_version 1.1;
proxy_set_header Host $host;