feat(identity): 增加统一认证配置版本

新增统一认证 Revision 状态机、单 Active 数据库约束、SecretStore 引用字段、Break-glass 与本地租户门禁，并在关键身份变化或禁用时清理旧 BFF Session。\n\n同时实现标准应用接入 Manifest v1 消费端，接入码只进入请求 Body，Exchange Token 只进入 Authorization Header，禁用重定向并限制响应大小。\n\n验证：go test ./...；go vet ./...
2026-07-17 11:52:01 +08:00
parent 2a73e18123
commit c2ce42fead
7 changed files with 1025 additions and 0 deletions
@@ -0,0 +1,240 @@
+package store
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/identity"
+	"github.com/jackc/pgx/v5"
+)
+
+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,
+local_tenant_key,public_base_url,web_base_url,jit_enabled,legacy_jwt_enabled,token_introspection,session_revocation,
+COALESCE(machine_credential_ref,''),COALESCE(session_encryption_key_ref,''),session_idle_seconds,session_absolute_seconds,
+session_refresh_seconds,version,COALESCE(last_error_category,''),COALESCE(last_trace_id,''),COALESCE(last_audit_id,''),
+validated_at,activated_at,superseded_at,created_at,updated_at`
+
+func (s *Store) CreateIdentityConfigurationRevision(ctx context.Context, revision identity.Revision) (identity.Revision, error) {
+	scopes, _ := json.Marshal(revision.Scopes)
+	capabilities, _ := json.Marshal(revision.Capabilities)
+	return scanIdentityRevision(s.pool.QueryRow(ctx, `
+INSERT INTO gateway_identity_configuration_revisions (
+  id,state,schema_version,auth_center_url,role_prefix,local_tenant_key,public_base_url,web_base_url,
+  jit_enabled,legacy_jwt_enabled,scopes,capabilities,session_idle_seconds,session_absolute_seconds,session_refresh_seconds
+) VALUES ($1::uuid,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11::jsonb,$12::jsonb,$13,$14,$15)
+RETURNING `+identityRevisionColumns,
+		revision.ID, revision.State, revision.SchemaVersion, revision.AuthCenterURL, revision.RolePrefix,
+		revision.LocalTenantKey, revision.PublicBaseURL, revision.WebBaseURL, revision.JITEnabled, revision.LegacyJWTEnabled,
+		string(scopes), string(capabilities), revision.SessionIdleSeconds, revision.SessionAbsoluteSeconds, revision.SessionRefreshSeconds,
+	))
+}
+
+func (s *Store) IdentityConfigurationRevision(ctx context.Context, id string) (identity.Revision, error) {
+	revision, err := scanIdentityRevision(s.pool.QueryRow(ctx, `SELECT `+identityRevisionColumns+`
+FROM gateway_identity_configuration_revisions WHERE id=$1::uuid`, id))
+	return revision, normalizeIdentityRevisionError(err)
+}
+
+func (s *Store) ActiveIdentityConfigurationRevision(ctx context.Context) (identity.Revision, error) {
+	revision, err := scanIdentityRevision(s.pool.QueryRow(ctx, `SELECT `+identityRevisionColumns+`
+FROM gateway_identity_configuration_revisions WHERE state='active'`))
+	return revision, normalizeIdentityRevisionError(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 {
+		return identity.Revision{}, err
+	}
+	if current.Version != expectedVersion {
+		return identity.Revision{}, identity.ErrRevisionConflict
+	}
+	updated, err := identity.ApplyManifest(current, applied)
+	if err != nil {
+		return identity.Revision{}, err
+	}
+	scopes, _ := json.Marshal(updated.Scopes)
+	capabilities, _ := json.Marshal(updated.Capabilities)
+	revision, err := scanIdentityRevision(s.pool.QueryRow(ctx, `
+UPDATE gateway_identity_configuration_revisions SET
+  issuer=$3,tenant_id=$4,application_id=$5,audience=NULLIF($6,''),browser_client_id=NULLIF($7,''),machine_client_id=NULLIF($8,''),
+  scopes=$9::jsonb,capabilities=$10::jsonb,token_introspection=$11,session_revocation=$12,
+  machine_credential_ref=NULLIF($13,''),session_encryption_key_ref=NULLIF($14,''),last_trace_id=NULLIF($15,''),
+  last_audit_id=NULLIF($16,''),last_error_category=NULL,version=version+1,updated_at=now()
+WHERE id=$1::uuid AND version=$2 AND state='draft'
+RETURNING `+identityRevisionColumns,
+		id, expectedVersion, updated.Issuer, updated.TenantID, updated.ApplicationID, updated.Audience,
+		updated.BrowserClientID, updated.MachineClientID, string(scopes), string(capabilities), updated.TokenIntrospection,
+		updated.SessionRevocation, updated.MachineCredentialRef, updated.SessionEncryptionKeyRef, updated.LastTraceID, updated.LastAuditID,
+	))
+	if errors.Is(err, pgx.ErrNoRows) {
+		return identity.Revision{}, identity.ErrRevisionConflict
+	}
+	return revision, err
+}
+
+func (s *Store) MarkIdentityRevisionValidated(ctx context.Context, id string, expectedVersion int64, traceID, auditID string) (identity.Revision, error) {
+	revision, err := scanIdentityRevision(s.pool.QueryRow(ctx, `
+UPDATE gateway_identity_configuration_revisions SET state='validated',validated_at=now(),last_error_category=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 IN ('draft','superseded')
+RETURNING `+identityRevisionColumns, id, expectedVersion, traceID, auditID))
+	if errors.Is(err, pgx.ErrNoRows) {
+		return identity.Revision{}, identity.ErrRevisionConflict
+	}
+	return revision, err
+}
+
+func (s *Store) MarkIdentityRevisionFailed(ctx context.Context, id string, expectedVersion int64, category, traceID, auditID string) (identity.Revision, error) {
+	revision, err := scanIdentityRevision(s.pool.QueryRow(ctx, `
+UPDATE gateway_identity_configuration_revisions SET state='failed',last_error_category=NULLIF($3,''),
+  last_trace_id=NULLIF($4,''),last_audit_id=NULLIF($5,''),version=version+1,updated_at=now()
+WHERE id=$1::uuid AND version=$2 AND state IN ('draft','validated')
+RETURNING `+identityRevisionColumns, id, expectedVersion, category, traceID, auditID))
+	if errors.Is(err, pgx.ErrNoRows) {
+		return identity.Revision{}, identity.ErrRevisionConflict
+	}
+	return revision, err
+}
+
+func (s *Store) ActivateIdentityRevision(ctx context.Context, id string, expectedVersion int64) (identity.Revision, bool, error) {
+	tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.Serializable})
+	if err != nil {
+		return identity.Revision{}, false, err
+	}
+	defer tx.Rollback(ctx)
+	if ok, err := hasBreakGlassManager(ctx, tx); err != nil {
+		return identity.Revision{}, false, err
+	} else if !ok {
+		return identity.Revision{}, false, identity.ErrBreakGlassRequired
+	}
+	var tenantExists bool
+	if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway_tenants WHERE tenant_key=(
+SELECT local_tenant_key FROM gateway_identity_configuration_revisions WHERE id=$1::uuid) AND status='active')`, id).Scan(&tenantExists); err != nil {
+		return identity.Revision{}, false, err
+	}
+	if !tenantExists {
+		return identity.Revision{}, false, identity.ErrLocalTenantInvalid
+	}
+	var previousID, previousIssuer, previousTenant, previousAudience, previousClient, previousSessionRef string
+	_ = tx.QueryRow(ctx, `SELECT id::text,COALESCE(issuer,''),COALESCE(tenant_id,''),COALESCE(audience,''),
+COALESCE(browser_client_id,''),COALESCE(session_encryption_key_ref,'') FROM gateway_identity_configuration_revisions
+WHERE state='active' FOR UPDATE`).Scan(&previousID, &previousIssuer, &previousTenant, &previousAudience, &previousClient, &previousSessionRef)
+	var nextIssuer, nextTenant, nextAudience, nextClient, nextSessionRef string
+	if err := tx.QueryRow(ctx, `SELECT COALESCE(issuer,''),COALESCE(tenant_id,''),COALESCE(audience,''),
+COALESCE(browser_client_id,''),COALESCE(session_encryption_key_ref,'') FROM gateway_identity_configuration_revisions
+WHERE id=$1::uuid AND version=$2 AND state='validated' FOR UPDATE`, id, expectedVersion).Scan(
+		&nextIssuer, &nextTenant, &nextAudience, &nextClient, &nextSessionRef,
+	); errors.Is(err, pgx.ErrNoRows) {
+		return identity.Revision{}, false, identity.ErrRevisionConflict
+	} else if err != nil {
+		return identity.Revision{}, false, err
+	}
+	if previousID != "" {
+		if _, err := tx.Exec(ctx, `UPDATE gateway_identity_configuration_revisions SET state='superseded',superseded_at=now(),
+version=version+1,updated_at=now() WHERE id=$1::uuid AND state='active'`, previousID); err != nil {
+			return identity.Revision{}, false, err
+		}
+	}
+	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))
+	if err != nil {
+		return identity.Revision{}, false, err
+	}
+	identityChanged := previousID != "" && (previousIssuer != nextIssuer || previousTenant != nextTenant || previousAudience != nextAudience || previousClient != nextClient || previousSessionRef != nextSessionRef)
+	if identityChanged {
+		if _, err := tx.Exec(ctx, `DELETE FROM gateway_oidc_sessions`); err != nil {
+			return identity.Revision{}, false, err
+		}
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return identity.Revision{}, false, err
+	}
+	return revision, identityChanged, nil
+}
+
+func (s *Store) DisableActiveIdentityRevision(ctx context.Context, expectedVersion int64) (identity.Revision, error) {
+	tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.Serializable})
+	if err != nil {
+		return identity.Revision{}, err
+	}
+	defer tx.Rollback(ctx)
+	if ok, err := hasBreakGlassManager(ctx, tx); err != nil {
+		return identity.Revision{}, err
+	} else if !ok {
+		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))
+	if errors.Is(err, pgx.ErrNoRows) {
+		return identity.Revision{}, identity.ErrRevisionConflict
+	}
+	if err != nil {
+		return identity.Revision{}, err
+	}
+	if _, err := tx.Exec(ctx, `DELETE FROM gateway_oidc_sessions`); err != nil {
+		return identity.Revision{}, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return identity.Revision{}, err
+	}
+	return revision, nil
+}
+
+func (s *Store) HasBreakGlassManager(ctx context.Context) (bool, error) {
+	return hasBreakGlassManager(ctx, s.pool)
+}
+
+func (s *Store) HasActiveTenantKey(ctx context.Context, tenantKey string) (bool, error) {
+	var exists bool
+	err := s.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway_tenants WHERE tenant_key=$1 AND status='active')`, tenantKey).Scan(&exists)
+	return exists, err
+}
+
+func hasBreakGlassManager(ctx context.Context, query interface {
+	QueryRow(context.Context, string, ...any) pgx.Row
+}) (bool, error) {
+	var exists bool
+	err := query.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway_users WHERE source='gateway' AND status='active'
+AND deleted_at IS NULL AND password_hash IS NOT NULL AND password_hash <> '' AND (roles ? 'manager' OR roles ? 'admin'))`).Scan(&exists)
+	return exists, err
+}
+
+func scanIdentityRevision(row scanner) (identity.Revision, error) {
+	var revision identity.Revision
+	var state string
+	var scopes, capabilities []byte
+	if err := row.Scan(
+		&revision.ID, &state, &revision.SchemaVersion, &revision.AuthCenterURL, &revision.Issuer, &revision.TenantID,
+		&revision.ApplicationID, &revision.Audience, &revision.BrowserClientID, &revision.MachineClientID, &scopes,
+		&capabilities, &revision.RolePrefix, &revision.LocalTenantKey, &revision.PublicBaseURL, &revision.WebBaseURL,
+		&revision.JITEnabled, &revision.LegacyJWTEnabled, &revision.TokenIntrospection, &revision.SessionRevocation,
+		&revision.MachineCredentialRef, &revision.SessionEncryptionKeyRef, &revision.SessionIdleSeconds,
+		&revision.SessionAbsoluteSeconds, &revision.SessionRefreshSeconds, &revision.Version, &revision.LastErrorCategory,
+		&revision.LastTraceID, &revision.LastAuditID, &revision.ValidatedAt, &revision.ActivatedAt, &revision.SupersededAt,
+		&revision.CreatedAt, &revision.UpdatedAt,
+	); err != nil {
+		return identity.Revision{}, err
+	}
+	revision.State = identity.RevisionState(state)
+	_ = json.Unmarshal(scopes, &revision.Scopes)
+	_ = json.Unmarshal(capabilities, &revision.Capabilities)
+	if revision.Scopes == nil {
+		revision.Scopes = []string{}
+	}
+	if revision.Capabilities == nil {
+		revision.Capabilities = []string{}
+	}
+	return revision, nil
+}
+
+func normalizeIdentityRevisionError(err error) error {
+	if errors.Is(err, pgx.ErrNoRows) {
+		return identity.ErrRevisionNotFound
+	}
+	return err
+}