From 1fce3a535d95fb4f66efffbd0eb62a35e086d616 Mon Sep 17 00:00:00 2001 From: chengcheng Date: Fri, 17 Jul 2026 12:10:44 +0800 Subject: [PATCH] =?UTF-8?q?feat(identity):=20=E5=A2=9E=E5=8A=A0=E7=BB=9F?= =?UTF-8?q?=E4=B8=80=E8=AE=A4=E8=AF=81=E7=AE=A1=E7=90=86=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 提供接入码配对、配置查询、本地策略修改、验证、激活、回滚、禁用和公开运行时状态接口。写操作强制 Idempotency-Key 与 If-Match，后台 Exchange 支持重启恢复和过期收敛，并记录脱敏 Trace 与审计。\n\n验证：go test 相关 identity、store、identityruntime、httpapi 包；go vet ./apps/api/... --- .../identity_configuration_handlers.go | 491 ++++++++++++++++++ .../identity_configuration_handlers_test.go | 33 ++ apps/api/internal/httpapi/server.go | 52 +- apps/api/internal/identity/configuration.go | 21 + apps/api/internal/identity/pairing_service.go | 9 + .../internal/identity/pairing_service_test.go | 23 + apps/api/internal/identityruntime/manager.go | 42 +- .../internal/identityruntime/manager_test.go | 28 +- .../internal/store/identity_configurations.go | 73 ++- apps/api/internal/store/identity_pairing.go | 28 + 10 files changed, 767 insertions(+), 33 deletions(-) create mode 100644 apps/api/internal/httpapi/identity_configuration_handlers.go create mode 100644 apps/api/internal/httpapi/identity_configuration_handlers_test.go diff --git a/apps/api/internal/httpapi/identity_configuration_handlers.go b/apps/api/internal/httpapi/identity_configuration_handlers.go new file mode 100644 index 0000000..aedf60a --- /dev/null +++ b/apps/api/internal/httpapi/identity_configuration_handlers.go @@ -0,0 +1,491 @@ +package httpapi + +import ( + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/auth" + "github.com/easyai/easyai-ai-gateway/apps/api/internal/identity" + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +type identityConfigurationView struct { + Active *identity.Revision `json:"active"` + Draft *identity.Revision `json:"draft"` + Previous *identity.Revision `json:"previous"` + Pairing *identity.PairingExchange `json:"pairing,omitempty"` + Runtime identityRuntimeStatus `json:"runtime"` +} + +type identityRuntimeStatus struct { + Enabled bool `json:"enabled"` + RevisionID string `json:"revisionId,omitempty"` + Login string `json:"login"` + JIT string `json:"jit"` + TokenIntrospection string `json:"tokenIntrospection"` + SessionRevocation string `json:"sessionRevocation"` + LastTraceID string `json:"lastTraceId,omitempty"` + LastAuditID string `json:"lastAuditId,omitempty"` + LastErrorCategory string `json:"lastErrorCategory,omitempty"` +} + +type publicIdentityConfiguration struct { + Enabled bool `json:"enabled"` + OIDCLogin bool `json:"oidcLogin"` + LoginURL string `json:"loginUrl,omitempty"` + LogoutURL string `json:"logoutUrl,omitempty"` + Status string `json:"status"` +} + +type identityPolicyPatch struct { + LocalTenantKey *string `json:"localTenantKey"` + RolePrefix *string `json:"rolePrefix"` + JITEnabled *bool `json:"jitEnabled"` + LegacyJWTEnabled *bool `json:"legacyJwtEnabled"` + SessionIdleSeconds *int `json:"sessionIdleSeconds"` + SessionAbsoluteSeconds *int `json:"sessionAbsoluteSeconds"` + SessionRefreshSeconds *int `json:"sessionRefreshSeconds"` +} + +type storedIdentityResponse struct { + Status int `json:"status"` + Body json.RawMessage `json:"body"` + ETag string `json:"etag,omitempty"` + AuditID string `json:"auditId,omitempty"` +} + +type identityWriteOperation struct { + operation, key, requestHash string + release func() + once sync.Once +} + +func (operation *identityWriteOperation) close() { + if operation != nil { + operation.once.Do(operation.release) + } +} + +func (s *Server) getPublicIdentityConfiguration(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Cache-Control", "no-store") + runtime := s.currentIdentityRuntime() + if runtime == nil || runtime.Revision.State != identity.RevisionActive { + writeJSON(w, http.StatusOK, publicIdentityConfiguration{Status: "disabled"}) + return + } + view := publicIdentityConfiguration{Enabled: true, Status: "active"} + if oidcRuntimeReady(runtime) { + view.OIDCLogin = true + view.LoginURL = "/api/v1/auth/oidc/login" + view.LogoutURL = "/api/v1/auth/oidc/logout" + } + writeJSON(w, http.StatusOK, view) +} + +func (s *Server) getIdentityConfiguration(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-store") + view := identityConfigurationView{Runtime: s.identityRuntimeStatus(r)} + if revision, err := s.store.ActiveIdentityConfigurationRevision(r.Context()); err == nil { + view.Active = &revision + } else if !errors.Is(err, identity.ErrRevisionNotFound) { + writeError(w, http.StatusServiceUnavailable, "统一认证配置暂时不可用", "IDENTITY_CONFIGURATION_UNAVAILABLE") + return + } + if revision, err := s.store.LatestInactiveIdentityConfigurationRevision(r.Context()); err == nil { + view.Draft = &revision + if pairing, pairingErr := s.store.IdentityPairingExchangeForRevision(r.Context(), revision.ID); pairingErr == nil { + view.Pairing = &pairing + } + } else if !errors.Is(err, identity.ErrRevisionNotFound) { + writeError(w, http.StatusServiceUnavailable, "统一认证草稿暂时不可用", "IDENTITY_CONFIGURATION_UNAVAILABLE") + return + } + if revision, err := s.store.LatestSupersededIdentityConfigurationRevision(r.Context()); err == nil { + view.Previous = &revision + } else if !errors.Is(err, identity.ErrRevisionNotFound) { + writeError(w, http.StatusServiceUnavailable, "统一认证历史版本暂时不可用", "IDENTITY_CONFIGURATION_UNAVAILABLE") + return + } + writeJSON(w, http.StatusOK, view) +} + +func (s *Server) startIdentityPairing(w http.ResponseWriter, r *http.Request) { + var input identity.PairingInput + if !decodeIdentityRequest(w, r, &input) { + return + } + operation, ok := s.beginIdentityWrite(w, r, "pairing.start", 0, input) + if !ok { + return + } + defer operation.close() + traceID := ensureIdentityTraceID(w, r) + pairing, err := s.identityPairing.Start(r.Context(), input, traceID) + if err != nil { + s.writeIdentityError(w, r, "pairing.start", "", traceID, err) + return + } + auditID := s.recordIdentityConfigurationAudit(r, "pairing.start", pairing.RevisionID, "accepted", traceID, "") + pairing.AuthCenterAuditID = "" + s.completeIdentityWrite(w, r, operation, http.StatusAccepted, pairing, pairing.Version, auditID) + s.startIdentityPairingWorker(pairing.ID) +} + +func (s *Server) getIdentityPairing(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-store") + pairing, err := s.store.IdentityPairingExchange(r.Context(), r.PathValue("pairingID")) + if err != nil { + s.writeIdentityError(w, r, "pairing.get", r.PathValue("pairingID"), ensureIdentityTraceID(w, r), err) + return + } + w.Header().Set("ETag", identityETag(pairing.Version)) + writeJSON(w, http.StatusOK, pairing) +} + +func (s *Server) updateIdentityDraftPolicy(w http.ResponseWriter, r *http.Request) { + var patch identityPolicyPatch + if !decodeIdentityRequest(w, r, &patch) { + return + } + expectedVersion, ok := requiredIdentityVersion(w, r) + if !ok { + return + } + operation, ok := s.beginIdentityWriteWithVersion(w, r, "revision.policy", expectedVersion, patch) + if !ok { + return + } + defer operation.close() + revision, err := s.store.IdentityConfigurationRevision(r.Context(), r.PathValue("revisionID")) + if err != nil || revision.Version != expectedVersion { + s.writeIdentityError(w, r, "revision.policy", r.PathValue("revisionID"), ensureIdentityTraceID(w, r), firstIdentityError(err, identity.ErrRevisionConflict)) + return + } + policy := identity.RevisionPolicy{ + LocalTenantKey: revision.LocalTenantKey, RolePrefix: revision.RolePrefix, JITEnabled: revision.JITEnabled, + LegacyJWTEnabled: revision.LegacyJWTEnabled, SessionIdleSeconds: revision.SessionIdleSeconds, + SessionAbsoluteSeconds: revision.SessionAbsoluteSeconds, SessionRefreshSeconds: revision.SessionRefreshSeconds, + } + applyIdentityPolicyPatch(&policy, patch) + updated, err := s.store.UpdateIdentityRevisionPolicy(r.Context(), revision.ID, expectedVersion, policy) + if err != nil { + s.writeIdentityError(w, r, "revision.policy", revision.ID, ensureIdentityTraceID(w, r), err) + return + } + traceID := ensureIdentityTraceID(w, r) + auditID := s.recordIdentityConfigurationAudit(r, "revision.policy", revision.ID, "success", traceID, "") + s.completeIdentityWrite(w, r, operation, http.StatusOK, updated, updated.Version, auditID) +} + +func (s *Server) validateIdentityRevision(w http.ResponseWriter, r *http.Request) { + s.runIdentityRevisionAction(w, r, "revision.validate", func(expected int64, traceID, auditID string) (identity.Revision, error) { + return s.identityRuntime.Validate(r.Context(), r.PathValue("revisionID"), expected, traceID, auditID) + }) +} + +func (s *Server) activateIdentityRevision(w http.ResponseWriter, r *http.Request) { + s.runIdentityRevisionAction(w, r, "revision.activate", func(expected int64, traceID, auditID string) (identity.Revision, error) { + return s.identityRuntime.Activate(r.Context(), r.PathValue("revisionID"), expected, traceID, auditID) + }) +} + +func (s *Server) rollbackIdentityRevision(w http.ResponseWriter, r *http.Request) { + s.runIdentityRevisionAction(w, r, "revision.rollback", func(expected int64, traceID, auditID string) (identity.Revision, error) { + return s.identityRuntime.Rollback(r.Context(), r.PathValue("revisionID"), expected, traceID, auditID) + }) +} + +func (s *Server) disableIdentityConfiguration(w http.ResponseWriter, r *http.Request) { + expectedVersion, ok := requiredIdentityVersion(w, r) + if !ok { + return + } + operation, ok := s.beginIdentityWriteWithVersion(w, r, "revision.disable", expectedVersion, struct{}{}) + if !ok { + return + } + defer operation.close() + traceID := ensureIdentityTraceID(w, r) + auditID := s.recordIdentityConfigurationAudit(r, "revision.disable", "active", "requested", traceID, "") + disabled, err := s.identityRuntime.Disable(r.Context(), expectedVersion, traceID, auditID) + if err != nil { + s.writeIdentityError(w, r, "revision.disable", "active", traceID, err) + return + } + s.completeIdentityWrite(w, r, operation, http.StatusOK, disabled, disabled.Version, auditID) +} + +func (s *Server) runIdentityRevisionAction(w http.ResponseWriter, r *http.Request, action string, run func(int64, string, string) (identity.Revision, error)) { + expectedVersion, ok := requiredIdentityVersion(w, r) + if !ok { + return + } + operation, ok := s.beginIdentityWriteWithVersion(w, r, action, expectedVersion, struct{}{}) + if !ok { + return + } + defer operation.close() + traceID := ensureIdentityTraceID(w, r) + auditID := s.recordIdentityConfigurationAudit(r, action, r.PathValue("revisionID"), "requested", traceID, "") + revision, err := run(expectedVersion, traceID, auditID) + if err != nil { + s.writeIdentityError(w, r, action, r.PathValue("revisionID"), traceID, err) + return + } + s.completeIdentityWrite(w, r, operation, http.StatusOK, revision, revision.Version, auditID) +} + +func (s *Server) identityRuntimeStatus(r *http.Request) identityRuntimeStatus { + runtime := s.currentIdentityRuntime() + if runtime == nil { + return identityRuntimeStatus{Login: "disabled", JIT: "disabled", TokenIntrospection: "disabled", SessionRevocation: "disabled"} + } + revision := runtime.Revision + status := identityRuntimeStatus{ + Enabled: true, RevisionID: revision.ID, Login: capabilityHealth(runtime.PublicClient != nil), + JIT: capabilityHealth(revision.JITEnabled), TokenIntrospection: capabilityHealth(revision.TokenIntrospection), + SessionRevocation: capabilityHealth(revision.SessionRevocation), LastTraceID: revision.LastTraceID, + LastAuditID: revision.LastAuditID, LastErrorCategory: revision.LastErrorCategory, + } + if revision.SessionRevocation && runtime.SecurityEvents != nil { + if connection, err := runtime.SecurityEvents.Get(r.Context()); err == nil && connection.LifecycleStatus != "active" { + status.SessionRevocation = connection.HealthMode + } + } + return status +} + +func capabilityHealth(enabled bool) string { + if enabled { + return "healthy" + } + return "disabled" +} + +func decodeIdentityRequest(w http.ResponseWriter, r *http.Request, output any) bool { + r.Body = http.MaxBytesReader(w, r.Body, 32*1024) + decoder := json.NewDecoder(r.Body) + decoder.DisallowUnknownFields() + if err := decoder.Decode(output); err != nil { + writeError(w, http.StatusBadRequest, "统一认证请求格式无效", "INVALID_REQUEST") + return false + } + if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) { + writeError(w, http.StatusBadRequest, "统一认证请求格式无效", "INVALID_REQUEST") + return false + } + return true +} + +func (s *Server) beginIdentityWrite(w http.ResponseWriter, r *http.Request, operation string, version int64, request any) (*identityWriteOperation, bool) { + if parsed, ok := requiredIdentityVersion(w, r); !ok || parsed != version { + if ok { + writeError(w, http.StatusPreconditionFailed, "统一认证配置版本已变化", "IDENTITY_VERSION_CONFLICT") + } + return nil, false + } + return s.beginIdentityWriteWithVersion(w, r, operation, version, request) +} + +func (s *Server) beginIdentityWriteWithVersion(w http.ResponseWriter, r *http.Request, operation string, version int64, request any) (*identityWriteOperation, bool) { + key := strings.TrimSpace(r.Header.Get("Idempotency-Key")) + if key == "" || len(key) > 255 { + writeError(w, http.StatusBadRequest, "Idempotency-Key is required", "IDEMPOTENCY_KEY_REQUIRED") + return nil, false + } + encoded, _ := json.Marshal(request) + digest := sha256.Sum256(append([]byte(fmt.Sprintf("%s\x00%d\x00", operation, version)), encoded...)) + requestHash := fmt.Sprintf("%x", digest[:]) + s.identityManagementMu.Lock() + write := &identityWriteOperation{operation: operation, key: key, requestHash: requestHash, release: s.identityManagementMu.Unlock} + if s.store == nil { + return write, true + } + recorded, err := s.store.IdentityManagementRequest(r.Context(), operation, key) + if errors.Is(err, store.ErrIdentityManagementRequestNotFound) { + return write, true + } + if err != nil { + write.close() + writeError(w, http.StatusServiceUnavailable, "幂等状态暂时不可用", "IDENTITY_IDEMPOTENCY_UNAVAILABLE") + return nil, false + } + if recorded.RequestHash != requestHash { + write.close() + writeError(w, http.StatusConflict, "Idempotency-Key 已用于其他请求", "IDEMPOTENCY_KEY_REUSED") + return nil, false + } + var response storedIdentityResponse + if json.Unmarshal(recorded.Response, &response) != nil { + write.close() + writeError(w, http.StatusServiceUnavailable, "幂等响应暂时不可用", "IDENTITY_IDEMPOTENCY_UNAVAILABLE") + return nil, false + } + write.close() + w.Header().Set("Cache-Control", "no-store") + w.Header().Set("Idempotent-Replayed", "true") + if response.ETag != "" { + w.Header().Set("ETag", response.ETag) + } + if response.AuditID != "" { + w.Header().Set("X-Audit-Id", response.AuditID) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(response.Status) + _, _ = w.Write(response.Body) + return nil, false +} + +func (s *Server) completeIdentityWrite(w http.ResponseWriter, r *http.Request, operation *identityWriteOperation, status int, payload any, version int64, auditID string) { + body, _ := json.Marshal(payload) + stored, _ := json.Marshal(storedIdentityResponse{Status: status, Body: body, ETag: identityETag(version), AuditID: auditID}) + if s.store != nil { + if err := s.store.RecordIdentityManagementRequest(r.Context(), store.IdentityManagementRequest{ + Operation: operation.operation, Key: operation.key, RequestHash: operation.requestHash, Response: stored, + }); err != nil && s.logger != nil { + s.logger.ErrorContext(r.Context(), "identity idempotency response could not be recorded", "operation", operation.operation, "error_category", "idempotency_store_failed") + } + } + operation.close() + w.Header().Set("Cache-Control", "no-store") + w.Header().Set("ETag", identityETag(version)) + if auditID != "" { + w.Header().Set("X-Audit-Id", auditID) + } + writeJSON(w, status, payload) +} + +func requiredIdentityVersion(w http.ResponseWriter, r *http.Request) (int64, bool) { + value := strings.Trim(strings.TrimPrefix(strings.TrimSpace(r.Header.Get("If-Match")), "W/"), `"`) + version, err := strconv.ParseInt(value, 10, 64) + if err != nil || version < 0 { + writeError(w, http.StatusPreconditionRequired, "If-Match is required", "IF_MATCH_REQUIRED") + return 0, false + } + return version, true +} + +func identityETag(version int64) string { return fmt.Sprintf(`W/"%d"`, version) } + +func applyIdentityPolicyPatch(policy *identity.RevisionPolicy, patch identityPolicyPatch) { + if patch.LocalTenantKey != nil { + policy.LocalTenantKey = strings.TrimSpace(*patch.LocalTenantKey) + } + if patch.RolePrefix != nil { + policy.RolePrefix = strings.TrimSpace(*patch.RolePrefix) + } + if patch.JITEnabled != nil { + policy.JITEnabled = *patch.JITEnabled + } + if patch.LegacyJWTEnabled != nil { + policy.LegacyJWTEnabled = *patch.LegacyJWTEnabled + } + if patch.SessionIdleSeconds != nil { + policy.SessionIdleSeconds = *patch.SessionIdleSeconds + } + if patch.SessionAbsoluteSeconds != nil { + policy.SessionAbsoluteSeconds = *patch.SessionAbsoluteSeconds + } + if patch.SessionRefreshSeconds != nil { + policy.SessionRefreshSeconds = *patch.SessionRefreshSeconds + } +} + +func firstIdentityError(actual, fallback error) error { + if actual != nil { + return actual + } + return fallback +} + +func ensureIdentityTraceID(w http.ResponseWriter, r *http.Request) string { + return ensureSecurityEventTraceID(w, r) +} + +func (s *Server) writeIdentityError(w http.ResponseWriter, r *http.Request, action, targetID, traceID string, err error) { + status, message, code := identityErrorProjection(err) + s.recordIdentityConfigurationAudit(r, action, targetID, "failure", traceID, code) + writeError(w, status, message, code) +} + +func identityErrorProjection(err error) (int, string, string) { + switch { + case errors.Is(err, identity.ErrRevisionNotFound): + return http.StatusNotFound, "统一认证配置不存在", "IDENTITY_CONFIGURATION_NOT_FOUND" + case errors.Is(err, identity.ErrRevisionConflict): + return http.StatusPreconditionFailed, "统一认证配置版本或状态已变化", "IDENTITY_VERSION_CONFLICT" + case errors.Is(err, identity.ErrBreakGlassRequired): + return http.StatusConflict, "请先保留至少一个可用的本地应急管理员凭据", "BREAK_GLASS_MANAGER_REQUIRED" + case errors.Is(err, identity.ErrLocalTenantInvalid): + return http.StatusConflict, "本地租户映射无效", "IDENTITY_LOCAL_TENANT_INVALID" + case err != nil && (strings.Contains(err.Error(), "invalid") || strings.Contains(err.Error(), "required")): + return http.StatusBadRequest, "统一认证配置无效", "IDENTITY_CONFIGURATION_INVALID" + default: + return http.StatusBadGateway, "认证中心或统一认证服务暂时不可用", "IDENTITY_SERVICE_UNAVAILABLE" + } +} + +func (s *Server) recordIdentityConfigurationAudit(r *http.Request, action, targetID, outcome, traceID, errorCategory string) string { + if s.store == nil { + return "" + } + actor, _ := auth.UserFromContext(r.Context()) + input := store.AuditLogInput{ + Category: "identity", Action: "identity." + action, TargetType: "identity_configuration_revision", + TargetID: firstNonEmptyText(targetID, "pending"), RequestIP: limitAuditText(requestIP(r), 128), + UserAgent: limitAuditText(r.UserAgent(), 512), Metadata: map[string]any{ + "outcome": outcome, "traceId": traceID, "errorCategory": errorCategory, + }, + } + if actor != nil { + input.ActorGatewayUserID = uuidText(firstNonEmptyText(actor.GatewayUserID, actor.ID)) + input.ActorUserID, input.ActorUsername, input.ActorSource, input.ActorRoles = actor.ID, actor.Username, actor.Source, actor.Roles + } + audit, err := s.store.RecordAuditLog(r.Context(), input) + if err != nil { + if s.logger != nil { + s.logger.WarnContext(r.Context(), "record identity audit failed", "action", action, "error_category", "audit_store_failed", "trace_id", traceID) + } + return "" + } + return audit.ID +} + +func (s *Server) startIdentityPairingWorker(pairingID string) { + if _, loaded := s.identityPairingWorkers.LoadOrStore(pairingID, struct{}{}); loaded { + return + } + go func() { + defer s.identityPairingWorkers.Delete(pairingID) + delay := time.Second + for { + pairing, err := s.identityPairing.Continue(s.ctx, pairingID) + if err == nil { + if pairing.Status == identity.PairingCompleted || pairing.Status == identity.PairingFailed || pairing.Status == identity.PairingExpired { + return + } + delay = time.Second + } else { + if s.logger != nil { + s.logger.Warn("identity pairing step failed and will retry", "pairing_id", pairingID, "error_category", "pairing_step_failed") + } + if delay < 15*time.Second { + delay *= 2 + } + } + select { + case <-s.ctx.Done(): + return + case <-time.After(delay): + } + } + }() +} diff --git a/apps/api/internal/httpapi/identity_configuration_handlers_test.go b/apps/api/internal/httpapi/identity_configuration_handlers_test.go new file mode 100644 index 0000000..4dc8a23 --- /dev/null +++ b/apps/api/internal/httpapi/identity_configuration_handlers_test.go @@ -0,0 +1,33 @@ +package httpapi + +import ( + "net/http" + "net/http/httptest" + "testing" +) + +func TestRequiredIdentityWriteHeaders(t *testing.T) { + request := httptest.NewRequest(http.MethodPost, "/identity", nil) + recorder := httptest.NewRecorder() + if _, ok := requiredIdentityVersion(recorder, request); ok || recorder.Code != http.StatusPreconditionRequired { + t.Fatalf("missing If-Match status=%d", recorder.Code) + } + + request.Header.Set("If-Match", `W/"7"`) + recorder = httptest.NewRecorder() + version, ok := requiredIdentityVersion(recorder, request) + if !ok || version != 7 { + t.Fatalf("weak ETag version=%d ok=%v", version, ok) + } +} + +func TestIdentityErrorProjectionProtectsInternalDetails(t *testing.T) { + status, message, code := identityErrorProjection(assertionError("upstream response contained secret-token")) + if status != http.StatusBadGateway || code != "IDENTITY_SERVICE_UNAVAILABLE" || message == "upstream response contained secret-token" { + t.Fatalf("unsafe error projection status=%d code=%q message=%q", status, code, message) + } +} + +type assertionError string + +func (err assertionError) Error() string { return string(err) } diff --git a/apps/api/internal/httpapi/server.go b/apps/api/internal/httpapi/server.go index fd4d7e5..24d4389 100644 --- a/apps/api/internal/httpapi/server.go +++ b/apps/api/internal/httpapi/server.go @@ -20,21 +20,23 @@ import ( ) type Server struct { - ctx context.Context - cfg config.Config - store *store.Store - oidcUserResolver oidcUserResolver - auth *auth.Authenticator - oidcClient oidcPublicClient - oidcSessions oidcSessionManager - oidcSessionCipher *oidcsession.Cipher - runner *runner.Service - logger *slog.Logger - geminiUploadSessions sync.Map - securityEventReceiver http.Handler - securityEventManager *ssfreceiver.ConnectionManager - identityRuntime *identityruntime.Manager - identityPairing *identity.PairingService + ctx context.Context + cfg config.Config + store *store.Store + oidcUserResolver oidcUserResolver + auth *auth.Authenticator + oidcClient oidcPublicClient + oidcSessions oidcSessionManager + oidcSessionCipher *oidcsession.Cipher + runner *runner.Service + logger *slog.Logger + geminiUploadSessions sync.Map + securityEventReceiver http.Handler + securityEventManager *ssfreceiver.ConnectionManager + identityRuntime *identityruntime.Manager + identityPairing *identity.PairingService + identityManagementMu sync.Mutex + identityPairingWorkers sync.Map } type oidcPublicClient interface { @@ -77,8 +79,8 @@ func NewServerWithContext(ctx context.Context, cfg config.Config, db *store.Stor runtimeBuilder := identityruntime.NewRuntimeBuilder(ctx, db, secretStore, identityruntime.RuntimeBuilderConfig{ AppEnv: cfg.AppEnv, JWKSCacheTTL: 5 * time.Minute, HeartbeatInterval: time.Duration(cfg.OIDCSecurityEventsHeartbeatIntervalSeconds) * time.Second, - StaleAfter: time.Duration(cfg.OIDCSecurityEventsStaleAfterSeconds) * time.Second, - ClockSkew: time.Duration(cfg.OIDCSecurityEventsClockSkewSeconds) * time.Second, + StaleAfter: time.Duration(cfg.OIDCSecurityEventsStaleAfterSeconds) * time.Second, + ClockSkew: time.Duration(cfg.OIDCSecurityEventsClockSkewSeconds) * time.Second, }, securityEventMetrics) server.identityRuntime = identityruntime.NewManager(db, runtimeBuilder) if err := server.identityRuntime.LoadActive(ctx); err != nil && logger != nil { @@ -87,6 +89,13 @@ func NewServerWithContext(ctx context.Context, cfg config.Config, db *store.Stor server.identityPairing = identity.NewPairingService(db, secretStore, func(baseURL string) (identity.OnboardingRemote, error) { return identity.NewOnboardingClient(baseURL, nil) }, server.identityRuntime) + if pending, pendingErr := db.PendingIdentityPairingExchanges(ctx); pendingErr == nil { + for _, pairing := range pending { + server.startIdentityPairingWorker(pairing.ID) + } + } else if logger != nil { + logger.Warn("pending identity pairings could not be resumed", "error_category", "pairing_resume_failed") + } server.auth.OIDCVerifierProvider = func() *auth.OIDCVerifier { if runtime := server.identityRuntime.Current(); runtime != nil { return runtime.Verifier @@ -124,6 +133,7 @@ func NewServerWithContext(ctx context.Context, cfg config.Config, db *store.Stor mux.HandleFunc("GET /api/v1/auth/oidc/callback", server.completeOIDCLogin) mux.HandleFunc("POST /api/v1/auth/oidc/logout", server.logoutOIDCSession) mux.HandleFunc("DELETE /api/v1/auth/oidc/session", server.deleteOIDCBrowserSession) + mux.HandleFunc("GET /api/v1/public/identity", server.getPublicIdentityConfiguration) mux.HandleFunc("POST /api/v1/security-events/ssf", server.receiveSecurityEvent) mux.Handle("GET /api/v1/me", server.requireUser(auth.PermissionBasic, http.HandlerFunc(server.me))) mux.Handle("GET /api/v1/public/catalog/providers", server.auth.Require(auth.PermissionPublic, http.HandlerFunc(server.listCatalogProviders))) @@ -194,6 +204,14 @@ func NewServerWithContext(ctx context.Context, cfg config.Config, db *store.Stor mux.Handle("PATCH /api/admin/system/file-storage/settings", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.updateFileStorageSettings))) mux.Handle("GET /api/admin/system/client-customization/settings", server.requireAdmin(auth.PermissionPower, http.HandlerFunc(server.getClientCustomizationSettings))) mux.Handle("PATCH /api/admin/system/client-customization/settings", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.updateClientCustomizationSettings))) + mux.Handle("GET /api/admin/system/identity/configuration", server.requireAdmin(auth.PermissionPower, http.HandlerFunc(server.getIdentityConfiguration))) + mux.Handle("POST /api/admin/system/identity/pairings", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.startIdentityPairing))) + mux.Handle("GET /api/admin/system/identity/pairings/{pairingID}", server.requireAdmin(auth.PermissionPower, http.HandlerFunc(server.getIdentityPairing))) + mux.Handle("PATCH /api/admin/system/identity/revisions/{revisionID}", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.updateIdentityDraftPolicy))) + mux.Handle("POST /api/admin/system/identity/revisions/{revisionID}/validate", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.validateIdentityRevision))) + mux.Handle("POST /api/admin/system/identity/revisions/{revisionID}/activate", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.activateIdentityRevision))) + mux.Handle("POST /api/admin/system/identity/revisions/{revisionID}/rollback", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.rollbackIdentityRevision))) + mux.Handle("POST /api/admin/system/identity/disable", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.disableIdentityConfiguration))) mux.Handle("GET /api/admin/system/identity/security-events/connection", server.requireAdmin(auth.PermissionPower, http.HandlerFunc(server.getSecurityEventConnection))) mux.Handle("PUT /api/admin/system/identity/security-events/connection", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.putSecurityEventConnection))) mux.Handle("POST /api/admin/system/identity/security-events/connection/verify", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.verifySecurityEventConnection))) diff --git a/apps/api/internal/identity/configuration.go b/apps/api/internal/identity/configuration.go index a43f78e..f3eb4ad 100644 --- a/apps/api/internal/identity/configuration.go +++ b/apps/api/internal/identity/configuration.go @@ -17,6 +17,27 @@ var ( ErrLocalTenantInvalid = errors.New("local tenant mapping is invalid") ) +type RevisionPolicy struct { + LocalTenantKey string `json:"localTenantKey"` + RolePrefix string `json:"rolePrefix"` + JITEnabled bool `json:"jitEnabled"` + LegacyJWTEnabled bool `json:"legacyJwtEnabled"` + SessionIdleSeconds int `json:"sessionIdleSeconds"` + SessionAbsoluteSeconds int `json:"sessionAbsoluteSeconds"` + SessionRefreshSeconds int `json:"sessionRefreshSeconds"` +} + +func (policy RevisionPolicy) Validate() error { + if strings.TrimSpace(policy.LocalTenantKey) == "" || strings.TrimSpace(policy.RolePrefix) == "" { + return errors.New("local tenant mapping and role prefix are required") + } + if policy.SessionIdleSeconds <= 0 || policy.SessionAbsoluteSeconds <= policy.SessionIdleSeconds || + policy.SessionRefreshSeconds <= 0 || policy.SessionRefreshSeconds >= policy.SessionIdleSeconds { + return errors.New("identity session policy is invalid") + } + return nil +} + type RevisionState string const ( diff --git a/apps/api/internal/identity/pairing_service.go b/apps/api/internal/identity/pairing_service.go index 7a63c73..7ddc79c 100644 --- a/apps/api/internal/identity/pairing_service.go +++ b/apps/api/internal/identity/pairing_service.go @@ -129,6 +129,15 @@ func (service *PairingService) Continue(ctx context.Context, pairingID string) ( if pairing.Status == PairingCompleted || pairing.Status == PairingFailed || pairing.Status == PairingExpired { return pairing, nil } + if !pairing.ExpiresAt.After(time.Now()) { + expired, updateErr := service.repository.UpdateIdentityPairingExchange(ctx, pairing.ID, pairing.Version, PairingExchangeUpdate{ + Status: PairingExpired, RemoteVersion: pairing.RemoteVersion, LastErrorCategory: "exchange_expired", + }) + if updateErr == nil { + _ = service.secrets.Delete(ctx, pairing.ExchangeTokenRef) + } + return expired, updateErr + } revision, err := service.repository.IdentityConfigurationRevision(ctx, pairing.RevisionID) if err != nil { return PairingExchange{}, err diff --git a/apps/api/internal/identity/pairing_service_test.go b/apps/api/internal/identity/pairing_service_test.go index f8823c8..dd73124 100644 --- a/apps/api/internal/identity/pairing_service_test.go +++ b/apps/api/internal/identity/pairing_service_test.go @@ -154,3 +154,26 @@ func TestPairingRetriesWithRotatedCredentialAfterSecretStoreFailure(t *testing.T t.Fatalf("revision did not retain SecretStore references: %#v", repository.revision) } } + +func TestPairingExpirationIsPersistedAndExchangeTokenIsDestroyed(t *testing.T) { + repository := &pairingRepositoryFake{exchange: PairingExchange{ + ID: "pairing", RevisionID: "revision", ExchangeTokenRef: "identity-exchange-pairing", + Status: PairingPreparing, RemoteVersion: 2, ExpiresAt: time.Now().Add(-time.Minute), Version: 4, + }} + secrets := &secretStoreFake{values: map[string][]byte{"identity-exchange-pairing": []byte("temporary-token")}} + service := NewPairingService(repository, secrets, func(string) (OnboardingRemote, error) { + t.Fatal("expired exchange must not call the remote service") + return nil, nil + }, nil) + + expired, err := service.Continue(context.Background(), "pairing") + if err != nil { + t.Fatal(err) + } + if expired.Status != PairingExpired || expired.LastErrorCategory != "exchange_expired" { + t.Fatalf("expired pairing was not persisted: %#v", expired) + } + if _, exists := secrets.values["identity-exchange-pairing"]; exists { + t.Fatal("expired exchange token remained in SecretStore") + } +} diff --git a/apps/api/internal/identityruntime/manager.go b/apps/api/internal/identityruntime/manager.go index 0d1d2af..5a9cb46 100644 --- a/apps/api/internal/identityruntime/manager.go +++ b/apps/api/internal/identityruntime/manager.go @@ -18,8 +18,8 @@ type Repository interface { ActiveIdentityConfigurationRevision(context.Context) (identity.Revision, error) MarkIdentityRevisionValidated(context.Context, string, int64, string, string) (identity.Revision, error) MarkIdentityRevisionFailed(context.Context, string, int64, string, string, string) (identity.Revision, error) - ActivateIdentityRevision(context.Context, string, int64) (identity.Revision, bool, error) - DisableActiveIdentityRevision(context.Context, int64) (identity.Revision, error) + ActivateIdentityRevision(context.Context, string, int64, string, string) (identity.Revision, bool, error) + DisableActiveIdentityRevision(context.Context, int64, string, string) (identity.Revision, error) } type Builder interface { @@ -96,7 +96,7 @@ func (manager *Manager) Validate(ctx context.Context, id string, expectedVersion return manager.repository.MarkIdentityRevisionValidated(ctx, id, expectedVersion, traceID, auditID) } -func (manager *Manager) Activate(ctx context.Context, id string, expectedVersion int64) (identity.Revision, error) { +func (manager *Manager) Activate(ctx context.Context, id string, expectedVersion int64, traceID, auditID string) (identity.Revision, error) { manager.operation.Lock() defer manager.operation.Unlock() revision, err := manager.repository.IdentityConfigurationRevision(ctx, id) @@ -110,7 +110,7 @@ func (manager *Manager) Activate(ctx context.Context, id string, expectedVersion if err != nil { return identity.Revision{}, err } - activated, _, err := manager.repository.ActivateIdentityRevision(ctx, id, expectedVersion) + activated, _, err := manager.repository.ActivateIdentityRevision(ctx, id, expectedVersion, traceID, auditID) if err != nil { candidate.Close() return identity.Revision{}, err @@ -121,10 +121,40 @@ func (manager *Manager) Activate(ctx context.Context, id string, expectedVersion return activated, nil } -func (manager *Manager) Disable(ctx context.Context, expectedVersion int64) (identity.Revision, error) { +func (manager *Manager) Rollback(ctx context.Context, id string, expectedVersion int64, traceID, auditID string) (identity.Revision, error) { manager.operation.Lock() defer manager.operation.Unlock() - disabled, err := manager.repository.DisableActiveIdentityRevision(ctx, expectedVersion) + revision, err := manager.repository.IdentityConfigurationRevision(ctx, id) + if err != nil { + return identity.Revision{}, err + } + if revision.Version != expectedVersion || revision.State != identity.RevisionSuperseded { + return identity.Revision{}, identity.ErrRevisionConflict + } + candidate, err := manager.builder.Build(ctx, revision) + if err != nil { + return identity.Revision{}, err + } + validated, err := manager.repository.MarkIdentityRevisionValidated(ctx, id, expectedVersion, traceID, auditID) + if err != nil { + candidate.Close() + return identity.Revision{}, err + } + activated, _, err := manager.repository.ActivateIdentityRevision(ctx, id, validated.Version, traceID, auditID) + if err != nil { + candidate.Close() + return identity.Revision{}, err + } + candidate.Revision = activated + old := manager.current.Swap(candidate) + retireRuntime(old) + return activated, nil +} + +func (manager *Manager) Disable(ctx context.Context, expectedVersion int64, traceID, auditID string) (identity.Revision, error) { + manager.operation.Lock() + defer manager.operation.Unlock() + disabled, err := manager.repository.DisableActiveIdentityRevision(ctx, expectedVersion, traceID, auditID) if err != nil { return identity.Revision{}, err } diff --git a/apps/api/internal/identityruntime/manager_test.go b/apps/api/internal/identityruntime/manager_test.go index 3494d70..4d1f9c0 100644 --- a/apps/api/internal/identityruntime/manager_test.go +++ b/apps/api/internal/identityruntime/manager_test.go @@ -45,7 +45,7 @@ func (f *runtimeRepositoryFake) MarkIdentityRevisionFailed(_ context.Context, id f.revisions[id] = revision return revision, nil } -func (f *runtimeRepositoryFake) ActivateIdentityRevision(_ context.Context, id string, expected int64) (identity.Revision, bool, error) { +func (f *runtimeRepositoryFake) ActivateIdentityRevision(_ context.Context, id string, expected int64, traceID, auditID string) (identity.Revision, bool, error) { f.activateCalled = true revision := f.revisions[id] if revision.Version != expected || revision.State != identity.RevisionValidated { @@ -56,16 +56,16 @@ func (f *runtimeRepositoryFake) ActivateIdentityRevision(_ context.Context, id s old.State = identity.RevisionSuperseded f.revisions[old.ID] = old } - revision.State, revision.Version = identity.RevisionActive, revision.Version+1 + revision.State, revision.Version, revision.LastTraceID, revision.LastAuditID = identity.RevisionActive, revision.Version+1, traceID, auditID f.active, f.revisions[id] = revision, revision return revision, true, nil } -func (f *runtimeRepositoryFake) DisableActiveIdentityRevision(_ context.Context, expected int64) (identity.Revision, error) { +func (f *runtimeRepositoryFake) DisableActiveIdentityRevision(_ context.Context, expected int64, traceID, auditID string) (identity.Revision, error) { if f.active.Version != expected { return identity.Revision{}, identity.ErrRevisionConflict } disabled := f.active - disabled.State, disabled.Version = identity.RevisionSuperseded, disabled.Version+1 + disabled.State, disabled.Version, disabled.LastTraceID, disabled.LastAuditID = identity.RevisionSuperseded, disabled.Version+1, traceID, auditID f.revisions[disabled.ID] = disabled f.active = identity.Revision{} return disabled, nil @@ -110,7 +110,7 @@ func TestActivationSwapsRuntimeOnlyAfterRepositoryActivation(t *testing.T) { manager := NewManager(repository, &runtimeBuilderFake{}) manager.current.Store(&Runtime{Revision: active}) - activated, err := manager.Activate(context.Background(), candidate.ID, candidate.Version) + activated, err := manager.Activate(context.Background(), candidate.ID, candidate.Version, "trace", "audit") if err != nil { t.Fatal(err) } @@ -118,3 +118,21 @@ func TestActivationSwapsRuntimeOnlyAfterRepositoryActivation(t *testing.T) { t.Fatalf("activation order failed: activated=%#v current=%#v", activated, manager.Current()) } } + +func TestRollbackValidatesAndAtomicallyReplacesActiveRuntime(t *testing.T) { + active := identity.Revision{ID: "current", State: identity.RevisionActive, Version: 5} + previous := identity.Revision{ID: "previous", State: identity.RevisionSuperseded, Version: 3} + repository := &runtimeRepositoryFake{ + revisions: map[string]identity.Revision{"current": active, "previous": previous}, active: active, + } + manager := NewManager(repository, &runtimeBuilderFake{}) + manager.current.Store(&Runtime{Revision: active}) + + rolledBack, err := manager.Rollback(context.Background(), previous.ID, previous.Version, "trace", "audit") + if err != nil { + t.Fatal(err) + } + if rolledBack.State != identity.RevisionActive || manager.Current().Revision.ID != previous.ID || repository.active.ID != previous.ID { + t.Fatalf("rollback did not replace active runtime: revision=%#v current=%#v", rolledBack, manager.Current()) + } +} diff --git a/apps/api/internal/store/identity_configurations.go b/apps/api/internal/store/identity_configurations.go index 1dcf285..0c66665 100644 --- a/apps/api/internal/store/identity_configurations.go +++ b/apps/api/internal/store/identity_configurations.go @@ -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 } diff --git a/apps/api/internal/store/identity_pairing.go b/apps/api/internal/store/identity_pairing.go index e37a215..6930ebc 100644 --- a/apps/api/internal/store/identity_pairing.go +++ b/apps/api/internal/store/identity_pairing.go @@ -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,''),