BCAI/easyai-ai-gateway
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(identity): 增加统一认证管理接口

Browse Source 
提供接入码配对、配置查询、本地策略修改、验证、激活、回滚、禁用和公开运行时状态接口。写操作强制 Idempotency-Key 与 If-Match，后台 Exchange 支持重启恢复和过期收敛，并记录脱敏 Trace 与审计。\n\n验证：go test 相关 identity、store、identityruntime、httpapi 包；go vet ./apps/api/...
This commit is contained in:
chengcheng
2026-07-17 12:10:44 +08:00
parent a9e23cb237
commit 1fce3a535d
10 changed files with 767 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/internal/store/identity_configurations.go
+68 -5
View File
@@ -4,11 +4,22 @@ import (
"context"
"encoding/json"
"errors"
"time"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/identity"
"github.com/jackc/pgx/v5"
)
var ErrIdentityManagementRequestNotFound = errors.New("identity management request not found")
type IdentityManagementRequest struct {
Operation string
Key string
RequestHash string
Response []byte
CreatedAt time.Time
}
const identityRevisionColumns = `
id::text,state,schema_version,auth_center_url,COALESCE(issuer,''),COALESCE(tenant_id,''),COALESCE(application_id,''),
COALESCE(audience,''),COALESCE(browser_client_id,''),COALESCE(machine_client_id,''),scopes,capabilities,role_prefix,
@@ -45,6 +56,57 @@ FROM gateway_identity_configuration_revisions WHERE state='active'`))
return revision, normalizeIdentityRevisionError(err)
}
func (s *Store) LatestInactiveIdentityConfigurationRevision(ctx context.Context) (identity.Revision, error) {
revision, err := scanIdentityRevision(s.pool.QueryRow(ctx, `SELECT `+identityRevisionColumns+`
FROM gateway_identity_configuration_revisions WHERE state IN ('draft','validated','failed') ORDER BY created_at DESC LIMIT 1`))
return revision, normalizeIdentityRevisionError(err)
}
func (s *Store) LatestSupersededIdentityConfigurationRevision(ctx context.Context) (identity.Revision, error) {
revision, err := scanIdentityRevision(s.pool.QueryRow(ctx, `SELECT `+identityRevisionColumns+`
FROM gateway_identity_configuration_revisions WHERE state='superseded' ORDER BY superseded_at DESC NULLS LAST LIMIT 1`))
return revision, normalizeIdentityRevisionError(err)
}
func (s *Store) UpdateIdentityRevisionPolicy(ctx context.Context, id string, expectedVersion int64, policy identity.RevisionPolicy) (identity.Revision, error) {
if err := policy.Validate(); err != nil {
return identity.Revision{}, err
}
revision, err := scanIdentityRevision(s.pool.QueryRow(ctx, `
UPDATE gateway_identity_configuration_revisions SET local_tenant_key=$3,role_prefix=$4,jit_enabled=$5,
legacy_jwt_enabled=$6,session_idle_seconds=$7,session_absolute_seconds=$8,session_refresh_seconds=$9,
version=version+1,updated_at=now()
WHERE id=$1::uuid AND version=$2 AND state='draft'
RETURNING `+identityRevisionColumns, id, expectedVersion, policy.LocalTenantKey, policy.RolePrefix, policy.JITEnabled,
policy.LegacyJWTEnabled, policy.SessionIdleSeconds, policy.SessionAbsoluteSeconds, policy.SessionRefreshSeconds))
if errors.Is(err, pgx.ErrNoRows) {
return identity.Revision{}, identity.ErrRevisionConflict
}
return revision, err
}
func (s *Store) IdentityManagementRequest(ctx context.Context, operation, key string) (IdentityManagementRequest, error) {
var request IdentityManagementRequest
err := s.pool.QueryRow(ctx, `SELECT operation,idempotency_key,request_hash,response,created_at
FROM gateway_identity_management_requests WHERE operation=$1 AND idempotency_key=$2`, operation, key).Scan(
&request.Operation, &request.Key, &request.RequestHash, &request.Response, &request.CreatedAt,
)
if errors.Is(err, pgx.ErrNoRows) {
return IdentityManagementRequest{}, ErrIdentityManagementRequestNotFound
}
return request, err
}
func (s *Store) RecordIdentityManagementRequest(ctx context.Context, request IdentityManagementRequest) error {
if !json.Valid(request.Response) {
return errors.New("identity management response is invalid")
}
_, err := s.pool.Exec(ctx, `INSERT INTO gateway_identity_management_requests(operation,idempotency_key,request_hash,response)
VALUES($1,$2,$3,$4::jsonb) ON CONFLICT(operation,idempotency_key) DO NOTHING`,
request.Operation, request.Key, request.RequestHash, request.Response)
return err
}
func (s *Store) ApplyIdentityManifest(ctx context.Context, id string, expectedVersion int64, applied identity.ManifestApplication) (identity.Revision, error) {
current, err := s.IdentityConfigurationRevision(ctx, id)
if err != nil {
@@ -103,7 +165,7 @@ RETURNING `+identityRevisionColumns, id, expectedVersion, category, traceID, aud
return revision, err
}
func (s *Store) ActivateIdentityRevision(ctx context.Context, id string, expectedVersion int64) (identity.Revision, bool, error) {
func (s *Store) ActivateIdentityRevision(ctx context.Context, id string, expectedVersion int64, traceID, auditID string) (identity.Revision, bool, error) {
tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.Serializable})
if err != nil {
return identity.Revision{}, false, err
@@ -143,8 +205,8 @@ version=version+1,updated_at=now() WHERE id=$1::uuid AND state='active'`, previo
}
}
revision, err := scanIdentityRevision(tx.QueryRow(ctx, `UPDATE gateway_identity_configuration_revisions SET state='active',
activated_at=now(),superseded_at=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid AND version=$2 AND state='validated'
RETURNING `+identityRevisionColumns, id, expectedVersion))
activated_at=now(),superseded_at=NULL,last_trace_id=NULLIF($3,''),last_audit_id=NULLIF($4,''),version=version+1,updated_at=now()
WHERE id=$1::uuid AND version=$2 AND state='validated' RETURNING `+identityRevisionColumns, id, expectedVersion, traceID, auditID))
if err != nil {
return identity.Revision{}, false, err
}
@@ -160,7 +222,7 @@ RETURNING `+identityRevisionColumns, id, expectedVersion))
return revision, identityChanged, nil
}
func (s *Store) DisableActiveIdentityRevision(ctx context.Context, expectedVersion int64) (identity.Revision, error) {
func (s *Store) DisableActiveIdentityRevision(ctx context.Context, expectedVersion int64, traceID, auditID string) (identity.Revision, error) {
tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.Serializable})
if err != nil {
return identity.Revision{}, err
@@ -172,7 +234,8 @@ func (s *Store) DisableActiveIdentityRevision(ctx context.Context, expectedVersi
return identity.Revision{}, identity.ErrBreakGlassRequired
}
revision, err := scanIdentityRevision(tx.QueryRow(ctx, `UPDATE gateway_identity_configuration_revisions SET state='superseded',
superseded_at=now(),version=version+1,updated_at=now() WHERE state='active' AND version=$1 RETURNING `+identityRevisionColumns, expectedVersion))
superseded_at=now(),last_trace_id=NULLIF($2,''),last_audit_id=NULLIF($3,''),version=version+1,updated_at=now()
WHERE state='active' AND version=$1 RETURNING `+identityRevisionColumns, expectedVersion, traceID, auditID))
if errors.Is(err, pgx.ErrNoRows) {
return identity.Revision{}, identity.ErrRevisionConflict
}
apps/api/internal/store/identity_pairing.go
+28
View File
@@ -32,6 +32,34 @@ FROM gateway_identity_onboarding_exchanges WHERE id=$1::uuid`, id))
return exchange, err
}
func (s *Store) IdentityPairingExchangeForRevision(ctx context.Context, revisionID string) (identity.PairingExchange, error) {
exchange, err := scanIdentityPairing(s.pool.QueryRow(ctx, `SELECT `+identityPairingColumns+`
FROM gateway_identity_onboarding_exchanges WHERE revision_id=$1::uuid`, revisionID))
if errors.Is(err, pgx.ErrNoRows) {
return identity.PairingExchange{}, identity.ErrRevisionNotFound
}
return exchange, err
}
func (s *Store) PendingIdentityPairingExchanges(ctx context.Context) ([]identity.PairingExchange, error) {
rows, err := s.pool.Query(ctx, `SELECT `+identityPairingColumns+`
FROM gateway_identity_onboarding_exchanges
WHERE status IN ('metadata_pending','preparing','ready','credentials_saved') ORDER BY created_at`)
if err != nil {
return nil, err
}
defer rows.Close()
exchanges := make([]identity.PairingExchange, 0)
for rows.Next() {
exchange, scanErr := scanIdentityPairing(rows)
if scanErr != nil {
return nil, scanErr
}
exchanges = append(exchanges, exchange)
}
return exchanges, rows.Err()
}
func (s *Store) UpdateIdentityPairingExchange(ctx context.Context, id string, expectedVersion int64, update identity.PairingExchangeUpdate) (identity.PairingExchange, error) {
exchange, err := scanIdentityPairing(s.pool.QueryRow(ctx, `
UPDATE gateway_identity_onboarding_exchanges SET status=$3,remote_version=$4,last_error_category=NULLIF($5,''),