feat(identity): 实现统一认证运行时热切换

从 Active Revision 构造并验证 OIDC、BFF Session、Introspection 与 SSF Runtime，在数据库激活成功后原子替换内存引用。请求链路使用不可变快照，失败保留当前运行时，本地管理登录不受影响。\n\n验证：go test ./apps/api/...；go vet ./apps/api/...
2026-07-17 12:05:00 +08:00
parent 96bbd3a2f6
commit a9e23cb237
14 changed files with 797 additions and 152 deletions
@@ -0,0 +1,209 @@
+package identityruntime
+
+import (
+	"context"
+	"errors"
+	"net/url"
+	"slices"
+	"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/oidcsession"
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/securityevents"
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
+)
+
+type RuntimeBuilderConfig struct {
+	AppEnv            string
+	JWKSCacheTTL      time.Duration
+	HeartbeatInterval time.Duration
+	StaleAfter        time.Duration
+	ClockSkew         time.Duration
+}
+
+type preparedSecurityRuntime struct {
+	manager *securityevents.ConnectionManager
+	cancel  context.CancelFunc
+}
+
+type RuntimeBuilder struct {
+	ctx      context.Context
+	store    *store.Store
+	secrets  PairingSecretStore
+	config   RuntimeBuilderConfig
+	metrics  *securityevents.Metrics
+	mutex    sync.Mutex
+	prepared map[string]preparedSecurityRuntime
+}
+
+type PairingSecretStore interface {
+	Put(context.Context, string, []byte) error
+	Get(context.Context, string) ([]byte, error)
+	Delete(context.Context, string) error
+}
+
+func NewRuntimeBuilder(ctx context.Context, data *store.Store, secrets PairingSecretStore, config RuntimeBuilderConfig, metrics *securityevents.Metrics) *RuntimeBuilder {
+	if metrics == nil {
+		metrics = &securityevents.Metrics{}
+	}
+	if config.JWKSCacheTTL <= 0 {
+		config.JWKSCacheTTL = 5 * time.Minute
+	}
+	return &RuntimeBuilder{ctx: ctx, store: data, secrets: secrets, config: config, metrics: metrics, prepared: map[string]preparedSecurityRuntime{}}
+}
+
+func (builder *RuntimeBuilder) Build(ctx context.Context, revision identity.Revision) (*Runtime, error) {
+	if builder.store == nil || builder.secrets == nil || revision.Issuer == "" || revision.TenantID == "" ||
+		revision.Audience == "" || revision.RolePrefix == "" {
+		return nil, errors.New("identity runtime configuration is incomplete")
+	}
+	if exists, err := builder.store.HasActiveTenantKey(ctx, revision.LocalTenantKey); err != nil {
+		return nil, err
+	} else if !exists {
+		return nil, identity.ErrLocalTenantInvalid
+	}
+	runtimeCtx, cancel := context.WithCancel(builder.ctx)
+	runtime := &Runtime{Revision: revision, CookieSecure: secureCookieFor(revision.PublicBaseURL), close: cancel}
+
+	var securityManager *securityevents.ConnectionManager
+	if revision.SessionRevocation {
+		prepared := builder.takePreparedSecurityRuntime(revision.ID)
+		securityManager = prepared.manager
+		if prepared.cancel != nil {
+			runtime.close = func() {
+				cancel()
+				prepared.cancel()
+			}
+		}
+		if securityManager == nil {
+			var err error
+			securityManager, err = builder.newSecurityEventManager(runtimeCtx, revision)
+			if err != nil {
+				cancel()
+				return nil, err
+			}
+		}
+		runtime.SecurityEvents = securityManager
+	}
+
+	var evaluator func(context.Context, auth.OIDCSecurityEventIdentity) (auth.OIDCSecurityEventEvaluation, error)
+	if securityManager != nil {
+		evaluator = securityManager.Evaluate
+	}
+	credentialProvider := func(ctx context.Context) (string, []byte, error) {
+		if revision.MachineClientID == "" || revision.MachineCredentialRef == "" {
+			return "", nil, errors.New("managed machine credential is unavailable")
+		}
+		secret, err := builder.secrets.Get(ctx, revision.MachineCredentialRef)
+		return revision.MachineClientID, secret, err
+	}
+	verifier, err := auth.NewOIDCVerifier(auth.OIDCConfig{
+		Issuer: revision.Issuer, Audience: revision.Audience, TenantID: revision.TenantID,
+		RolePrefix: revision.RolePrefix, RequiredScopes: append([]string(nil), revision.Scopes...),
+		JWKSCacheTTL: builder.config.JWKSCacheTTL, IntrospectionEnabled: revision.TokenIntrospection,
+		IntrospectionCredentialProvider: credentialProvider, SecurityEventEvaluator: evaluator,
+		IntrospectionObserver:      builder.metrics.ObserveIntrospection,
+		JWKSRefreshFailureObserver: func() { builder.metrics.ObserveJWKSRefreshFailure("oidc") },
+	})
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+	if err := verifier.ValidateConfiguration(ctx); err != nil {
+		cancel()
+		return nil, err
+	}
+	runtime.Verifier = verifier
+
+	if slices.Contains(revision.Capabilities, "oidc_login") {
+		if revision.BrowserClientID == "" || revision.SessionEncryptionKeyRef == "" {
+			cancel()
+			return nil, errors.New("OIDC browser session configuration is incomplete")
+		}
+		key, err := builder.secrets.Get(ctx, revision.SessionEncryptionKeyRef)
+		if err != nil {
+			cancel()
+			return nil, err
+		}
+		cipher, err := oidcsession.NewCipher(key)
+		clear(key)
+		if err != nil {
+			cancel()
+			return nil, err
+		}
+		client, err := auth.NewOIDCPublicClient(auth.OIDCPublicClientConfig{
+			Issuer: revision.Issuer, ClientID: revision.BrowserClientID,
+			RedirectURI:           revision.PublicBaseURL + "/api/v1/auth/oidc/callback",
+			PostLogoutRedirectURI: revision.WebBaseURL + "/", Scopes: append([]string{"openid", "profile"}, revision.Scopes...),
+		})
+		if err != nil {
+			cancel()
+			return nil, err
+		}
+		if err := client.ValidateConfiguration(ctx); err != nil {
+			cancel()
+			return nil, err
+		}
+		sessions, err := oidcsession.NewService(builder.store, cipher, verifier, client, oidcsession.Config{
+			IdleTTL:       time.Duration(revision.SessionIdleSeconds) * time.Second,
+			AbsoluteTTL:   time.Duration(revision.SessionAbsoluteSeconds) * time.Second,
+			RefreshBefore: time.Duration(revision.SessionRefreshSeconds) * time.Second,
+		})
+		if err != nil {
+			cancel()
+			return nil, err
+		}
+		runtime.PublicClient, runtime.Sessions, runtime.SessionCipher = client, sessions, cipher
+	}
+	return runtime, nil
+}
+
+func (builder *RuntimeBuilder) PrepareSecurityEvents(ctx context.Context, revision identity.Revision, managementSecret []byte) error {
+	if !revision.SessionRevocation || revision.SecurityEventIssuer == "" || revision.MachineClientID == "" {
+		return errors.New("security event configuration is incomplete")
+	}
+	runtimeCtx, cancel := context.WithCancel(builder.ctx)
+	manager, err := builder.newSecurityEventManager(runtimeCtx, revision)
+	if err != nil {
+		cancel()
+		return err
+	}
+	secretCopy := append([]byte(nil), managementSecret...)
+	_, err = manager.Connect(ctx, revision.SecurityEventIssuer, revision.MachineClientID, secretCopy, "identity-pairing-ssf-"+revision.ID)
+	clear(secretCopy)
+	if err != nil {
+		cancel()
+		return err
+	}
+	builder.mutex.Lock()
+	previous := builder.prepared[revision.ID]
+	builder.prepared[revision.ID] = preparedSecurityRuntime{manager: manager, cancel: cancel}
+	builder.mutex.Unlock()
+	if previous.cancel != nil {
+		previous.cancel()
+	}
+	return nil
+}
+
+func (builder *RuntimeBuilder) newSecurityEventManager(ctx context.Context, revision identity.Revision) (*securityevents.ConnectionManager, error) {
+	return securityevents.NewConnectionManager(ctx, builder.store, builder.secrets, securityevents.ConnectionManagerConfig{
+		AppEnv: builder.config.AppEnv, OIDCEnabled: true, OIDCIssuer: revision.Issuer, OIDCTenantID: revision.TenantID,
+		ManagementClientID: revision.MachineClientID, PublicBaseURL: revision.PublicBaseURL,
+		HeartbeatInterval: builder.config.HeartbeatInterval, StaleAfter: builder.config.StaleAfter, ClockSkew: builder.config.ClockSkew,
+	}, builder.metrics)
+}
+
+func (builder *RuntimeBuilder) takePreparedSecurityRuntime(revisionID string) preparedSecurityRuntime {
+	builder.mutex.Lock()
+	defer builder.mutex.Unlock()
+	prepared := builder.prepared[revisionID]
+	delete(builder.prepared, revisionID)
+	return prepared
+}
+
+func secureCookieFor(baseURL string) bool {
+	parsed, err := url.Parse(baseURL)
+	return err == nil && parsed.Scheme == "https"
+}
@@ -0,0 +1,151 @@
+package identityruntime
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"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/oidcsession"
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/securityevents"
+)
+
+type Repository interface {
+	IdentityConfigurationRevision(context.Context, string) (identity.Revision, error)
+	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)
+}
+
+type Builder interface {
+	Build(context.Context, identity.Revision) (*Runtime, error)
+}
+
+type Runtime struct {
+	Revision       identity.Revision
+	Verifier       *auth.OIDCVerifier
+	PublicClient   *auth.OIDCPublicClient
+	Sessions       *oidcsession.Service
+	SessionCipher  *oidcsession.Cipher
+	SecurityEvents *securityevents.ConnectionManager
+	CookieSecure   bool
+	close          func()
+}
+
+func (runtime *Runtime) Close() {
+	if runtime != nil && runtime.close != nil {
+		runtime.close()
+	}
+}
+
+type Manager struct {
+	repository Repository
+	builder    Builder
+	operation  sync.Mutex
+	current    atomic.Pointer[Runtime]
+}
+
+func NewManager(repository Repository, builder Builder) *Manager {
+	return &Manager{repository: repository, builder: builder}
+}
+
+func (manager *Manager) Current() *Runtime {
+	return manager.current.Load()
+}
+
+func (manager *Manager) LoadActive(ctx context.Context) error {
+	manager.operation.Lock()
+	defer manager.operation.Unlock()
+	revision, err := manager.repository.ActiveIdentityConfigurationRevision(ctx)
+	if errors.Is(err, identity.ErrRevisionNotFound) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	runtime, err := manager.builder.Build(ctx, revision)
+	if err != nil {
+		return err
+	}
+	runtime.Revision = revision
+	manager.current.Store(runtime)
+	return nil
+}
+
+func (manager *Manager) Validate(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)
+	if err != nil {
+		return identity.Revision{}, err
+	}
+	if revision.Version != expectedVersion || revision.State != identity.RevisionDraft && revision.State != identity.RevisionSuperseded {
+		return identity.Revision{}, identity.ErrRevisionConflict
+	}
+	candidate, buildErr := manager.builder.Build(ctx, revision)
+	if buildErr != nil {
+		_, _ = manager.repository.MarkIdentityRevisionFailed(ctx, id, expectedVersion, "validation_failed", traceID, auditID)
+		return identity.Revision{}, buildErr
+	}
+	candidate.Close()
+	return manager.repository.MarkIdentityRevisionValidated(ctx, id, expectedVersion, traceID, auditID)
+}
+
+func (manager *Manager) Activate(ctx context.Context, id string, expectedVersion int64) (identity.Revision, error) {
+	manager.operation.Lock()
+	defer manager.operation.Unlock()
+	revision, err := manager.repository.IdentityConfigurationRevision(ctx, id)
+	if err != nil {
+		return identity.Revision{}, err
+	}
+	if revision.Version != expectedVersion || revision.State != identity.RevisionValidated {
+		return identity.Revision{}, identity.ErrRevisionConflict
+	}
+	candidate, err := manager.builder.Build(ctx, revision)
+	if err != nil {
+		return identity.Revision{}, err
+	}
+	activated, _, err := manager.repository.ActivateIdentityRevision(ctx, id, expectedVersion)
+	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) (identity.Revision, error) {
+	manager.operation.Lock()
+	defer manager.operation.Unlock()
+	disabled, err := manager.repository.DisableActiveIdentityRevision(ctx, expectedVersion)
+	if err != nil {
+		return identity.Revision{}, err
+	}
+	old := manager.current.Swap(nil)
+	retireRuntime(old)
+	return disabled, nil
+}
+
+func (manager *Manager) PrepareSecurityEvents(ctx context.Context, revision identity.Revision, secret []byte) error {
+	preparer, ok := manager.builder.(interface {
+		PrepareSecurityEvents(context.Context, identity.Revision, []byte) error
+	})
+	if !ok {
+		return errors.New("security event runtime preparation is unavailable")
+	}
+	return preparer.PrepareSecurityEvents(ctx, revision, secret)
+}
+
+func retireRuntime(runtime *Runtime) {
+	if runtime == nil || runtime.close == nil {
+		return
+	}
+	time.AfterFunc(30*time.Second, runtime.Close)
+}
@@ -0,0 +1,120 @@
+package identityruntime
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/identity"
+)
+
+type runtimeRepositoryFake struct {
+	revisions      map[string]identity.Revision
+	active         identity.Revision
+	activateCalled bool
+}
+
+func (f *runtimeRepositoryFake) IdentityConfigurationRevision(_ context.Context, id string) (identity.Revision, error) {
+	revision, ok := f.revisions[id]
+	if !ok {
+		return identity.Revision{}, identity.ErrRevisionNotFound
+	}
+	return revision, nil
+}
+func (f *runtimeRepositoryFake) ActiveIdentityConfigurationRevision(context.Context) (identity.Revision, error) {
+	if f.active.ID == "" {
+		return identity.Revision{}, identity.ErrRevisionNotFound
+	}
+	return f.active, nil
+}
+func (f *runtimeRepositoryFake) MarkIdentityRevisionValidated(_ context.Context, id string, expected int64, _, _ string) (identity.Revision, error) {
+	revision := f.revisions[id]
+	if revision.Version != expected {
+		return identity.Revision{}, identity.ErrRevisionConflict
+	}
+	revision.State, revision.Version = identity.RevisionValidated, revision.Version+1
+	f.revisions[id] = revision
+	return revision, nil
+}
+func (f *runtimeRepositoryFake) MarkIdentityRevisionFailed(_ context.Context, id string, expected int64, category, _, _ string) (identity.Revision, error) {
+	revision := f.revisions[id]
+	if revision.Version != expected {
+		return identity.Revision{}, identity.ErrRevisionConflict
+	}
+	revision.State, revision.Version, revision.LastErrorCategory = identity.RevisionFailed, revision.Version+1, category
+	f.revisions[id] = revision
+	return revision, nil
+}
+func (f *runtimeRepositoryFake) ActivateIdentityRevision(_ context.Context, id string, expected int64) (identity.Revision, bool, error) {
+	f.activateCalled = true
+	revision := f.revisions[id]
+	if revision.Version != expected || revision.State != identity.RevisionValidated {
+		return identity.Revision{}, false, identity.ErrRevisionConflict
+	}
+	if f.active.ID != "" {
+		old := f.active
+		old.State = identity.RevisionSuperseded
+		f.revisions[old.ID] = old
+	}
+	revision.State, revision.Version = identity.RevisionActive, revision.Version+1
+	f.active, f.revisions[id] = revision, revision
+	return revision, true, nil
+}
+func (f *runtimeRepositoryFake) DisableActiveIdentityRevision(_ context.Context, expected int64) (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
+	f.revisions[disabled.ID] = disabled
+	f.active = identity.Revision{}
+	return disabled, nil
+}
+
+type runtimeBuilderFake struct {
+	err      error
+	builtIDs []string
+}
+
+func (f *runtimeBuilderFake) Build(_ context.Context, revision identity.Revision) (*Runtime, error) {
+	f.builtIDs = append(f.builtIDs, revision.ID)
+	if f.err != nil {
+		return nil, f.err
+	}
+	return &Runtime{Revision: revision}, nil
+}
+
+func TestValidationFailureKeepsCurrentRuntimeAndDoesNotActivate(t *testing.T) {
+	active := identity.Revision{ID: "active", State: identity.RevisionActive, Version: 4}
+	draft := identity.Revision{ID: "draft", State: identity.RevisionDraft, Version: 1}
+	repository := &runtimeRepositoryFake{revisions: map[string]identity.Revision{"active": active, "draft": draft}, active: active}
+	builder := &runtimeBuilderFake{err: errors.New("discovery failed")}
+	manager := NewManager(repository, builder)
+	manager.current.Store(&Runtime{Revision: active})
+
+	if _, err := manager.Validate(context.Background(), draft.ID, draft.Version, "trace", "audit"); err == nil {
+		t.Fatal("validation failure was ignored")
+	}
+	if manager.Current().Revision.ID != active.ID || repository.activateCalled {
+		t.Fatal("validation failure changed current runtime or activated the draft")
+	}
+	if repository.revisions[draft.ID].State != identity.RevisionFailed {
+		t.Fatal("failed draft was not marked failed")
+	}
+}
+
+func TestActivationSwapsRuntimeOnlyAfterRepositoryActivation(t *testing.T) {
+	active := identity.Revision{ID: "old", State: identity.RevisionActive, Version: 2}
+	candidate := identity.Revision{ID: "new", State: identity.RevisionValidated, Version: 3}
+	repository := &runtimeRepositoryFake{revisions: map[string]identity.Revision{"old": active, "new": candidate}, active: active}
+	manager := NewManager(repository, &runtimeBuilderFake{})
+	manager.current.Store(&Runtime{Revision: active})
+
+	activated, err := manager.Activate(context.Background(), candidate.ID, candidate.Version)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !repository.activateCalled || activated.State != identity.RevisionActive || manager.Current().Revision.ID != candidate.ID {
+		t.Fatalf("activation order failed: activated=%#v current=%#v", activated, manager.Current())
+	}
+}