diff --git a/.env.example b/.env.example index 25fd408..5fe82c3 100644 --- a/.env.example +++ b/.env.example @@ -35,19 +35,16 @@ OIDC_ACCEPT_LEGACY_HS256=true OIDC_INTROSPECTION_ENABLED=false OIDC_INTROSPECTION_CLIENT_ID= OIDC_INTROSPECTION_CLIENT_SECRET= -# Optional SSF/CAEP real-time revocation receiver. Enabling it also requires -# the RFC 7662 credentials above so bootstrap and fail-closed fallback work. -OIDC_SECURITY_EVENTS_ENABLED=false -OIDC_SECURITY_EVENTS_TRANSMITTER_ISSUER=https://auth.51easyai.com/ssf -OIDC_SECURITY_EVENTS_RECEIVER_AUDIENCE= -OIDC_SECURITY_EVENTS_BEARER_SECRET= -# During zero-downtime rotation, set next before removing current. -OIDC_SECURITY_EVENTS_BEARER_SECRET_NEXT= -OIDC_SECURITY_EVENTS_PUBLIC_ENDPOINT=https://api.51easyai.com/api/v1/security-events/ssf -OIDC_SECURITY_EVENTS_STREAM_ID= -OIDC_SECURITY_EVENTS_MANAGEMENT_TOKEN_URL= -OIDC_SECURITY_EVENTS_MANAGEMENT_CLIENT_ID= -OIDC_SECURITY_EVENTS_MANAGEMENT_CLIENT_SECRET= +# SSF/CAEP is activated by the durable connection created in System Settings; +# there is no enable flag and no Push Bearer in environment variables. +AI_GATEWAY_PUBLIC_BASE_URL=http://localhost:8088 +OIDC_SECURITY_EVENTS_SECRET_STORE=file +OIDC_SECURITY_EVENTS_SECRET_DIR=.local-secrets/ssf +# Kubernetes uses one pre-provisioned empty Secret and narrowly scoped RBAC. +# OIDC_SECURITY_EVENTS_SECRET_STORE=kubernetes +# OIDC_SECURITY_EVENTS_KUBERNETES_NAMESPACE=easyai +# OIDC_SECURITY_EVENTS_KUBERNETES_SECRET_NAME=easyai-gateway-security-events +# OIDC_SECURITY_EVENTS_KUBERNETES_API_SERVER=https://kubernetes.default.svc OIDC_SECURITY_EVENTS_HEARTBEAT_INTERVAL_SECONDS=60 OIDC_SECURITY_EVENTS_STALE_AFTER_SECONDS=180 OIDC_SECURITY_EVENTS_CLOCK_SKEW_SECONDS=60 diff --git a/.gitignore b/.gitignore index 6ff9836..5121d4a 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,7 @@ node_modules/ .DS_Store .env .env.local +.local-secrets/ .gateway-local-password *.log diff --git a/README.md b/README.md index d4561c0..5e33e10 100644 --- a/README.md +++ b/README.md @@ -56,7 +56,7 @@ OIDC_SESSION_ENCRYPTION_KEY= Web Console 使用公共 Client + PKCE + Gateway BFF Session：Gateway 服务端换码并加密保存 Access/Refresh/ID Token，浏览器 JavaScript 只持有不可读的 HttpOnly 随机 Session Cookie。Access Token 仍为 5 分钟并由业务请求按需刷新；Session 闲置 30 分钟、最长 8 小时。Gateway 不使用 Client Secret，也不二次签发 JWT。完整行为、错误码和回滚方式见 [OIDC JIT 接入说明](docs/oidc-jit-provisioning.md)。 -可选的 SSF/CAEP 实时撤销接收器提供 `POST /api/v1/security-events/ssf`：收到并验证 `session-revoked` SET 后，以本地 PostgreSQL 水位立即拒绝旧 OIDC Token，并删除同一租户 Subject 的 BFF Session。推送不健康时自动切换 RFC 7662，内省也不可用时 OIDC Fail Closed；API Key 与 Legacy JWT 行为不变。配置、轮换、监控和回滚见 [SSF 会话撤销运行手册](docs/security/ssf-session-revocation.md)。 +可选的 SSF/CAEP 实时撤销接收器提供 `POST /api/v1/security-events/ssf`：收到并验证 `session-revoked` SET 后，以本地 PostgreSQL 水位立即拒绝旧 OIDC Token，并删除同一租户 Subject 的 BFF Session。功能由“系统设置”中的持久连接启用，不使用环境变量开关；Gateway 后端自动生成和托管 Push Bearer，并复用现有 RFC 7662 机器 Client。推送不健康时自动切换 RFC 7662，内省也不可用时 OIDC Fail Closed；API Key 与 Legacy JWT 行为不变。配置、轮换、监控和回滚见 [SSF 会话撤销运行手册](docs/security/ssf-session-revocation.md)。 `pnpm dev` 会先创建数据库并执行 migration，然后并行启动： diff --git a/apps/api/internal/auth/oidc.go b/apps/api/internal/auth/oidc.go index df62e60..39690d4 100644 --- a/apps/api/internal/auth/oidc.go +++ b/apps/api/internal/auth/oidc.go @@ -48,6 +48,7 @@ type OIDCSecurityEventIdentity struct { } type OIDCSecurityEventEvaluation struct { + Enabled bool Revoked bool RequireIntrospection bool } @@ -91,7 +92,7 @@ func NewOIDCVerifier(config OIDCConfig) (*OIDCVerifier, error) { if err := validatePublicURL(config.Issuer); err != nil || config.Audience == "" || config.TenantID == "" || config.RolePrefix == "" { return nil, errors.New("issuer, audience, tenant and role prefix are required") } - if (config.IntrospectionEnabled || config.SecurityEventEvaluator != nil) && (strings.TrimSpace(config.IntrospectionClientID) == "" || config.IntrospectionClientSecret == "") { + if config.IntrospectionEnabled && (strings.TrimSpace(config.IntrospectionClientID) == "" || config.IntrospectionClientSecret == "") { return nil, errors.New("introspection client credentials are required when introspection is enabled") } if config.JWKSCacheTTL <= 0 { @@ -145,9 +146,6 @@ func (v *OIDCVerifier) Verify(ctx context.Context, raw string) (*User, error) { return nil, oidcUnauthorized("nbf is missing", nil) } issuedAt, hasIssuedAt := numericDateClaim(claims["iat"]) - if v.config.SecurityEventEvaluator != nil && !hasIssuedAt { - return nil, oidcUnauthorized("iat is required when security events are enabled", nil) - } scopes := scopeClaims(claims) if !containsAll(scopes, v.config.RequiredScopes) { return nil, oidcUnauthorized("required scope is missing", nil) @@ -166,6 +164,9 @@ func (v *OIDCVerifier) Verify(ctx context.Context, raw string) (*User, error) { if evaluateErr != nil { return nil, NewRequestAuthError(http.StatusServiceUnavailable, "OIDC_SECURITY_EVENT_STATE_UNAVAILABLE", "认证撤销状态暂时不可用") } + if evaluation.Enabled && !hasIssuedAt { + return nil, oidcUnauthorized("iat is required when security events are enabled", nil) + } if evaluation.Revoked { return nil, oidcUnauthorized("token was issued before the revocation watermark", nil) } @@ -177,6 +178,11 @@ func (v *OIDCVerifier) Verify(ctx context.Context, raw string) (*User, error) { if !active { return nil, oidcUnauthorized("token is inactive", nil) } + } else if !evaluation.Enabled && v.config.IntrospectionEnabled { + active, introspectionErr := v.introspect(ctx, raw) + if introspectionErr != nil || !active { + return nil, oidcUnauthorized("token is inactive", introspectionErr) + } } } else if v.config.IntrospectionEnabled { active, err := v.introspect(ctx, raw) diff --git a/apps/api/internal/auth/oidc_test.go b/apps/api/internal/auth/oidc_test.go index 16762a1..36ed7fd 100644 --- a/apps/api/internal/auth/oidc_test.go +++ b/apps/api/internal/auth/oidc_test.go @@ -179,9 +179,10 @@ func TestOIDCSecurityEventEvaluationUsesWatermarkAndFallbackIntrospection(t *tes verifier, err := NewOIDCVerifier(OIDCConfig{ Issuer: issuer, Audience: "gateway-api", TenantID: "tenant-1", RolePrefix: "gateway.", RequiredScopes: []string{"gateway.access"}, HTTPClient: server.Client(), + IntrospectionEnabled: true, IntrospectionClientID: "gateway-introspection", IntrospectionClientSecret: "introspection-secret", SecurityEventEvaluator: func(_ context.Context, identity OIDCSecurityEventIdentity) (OIDCSecurityEventEvaluation, error) { - if identity.Subject != "platform-subject" || identity.IssuedAt.IsZero() { + if identity.Subject != "platform-subject" { t.Fatal("security event evaluator received incomplete identity") } return evaluation, nil @@ -194,6 +195,10 @@ func TestOIDCSecurityEventEvaluationUsesWatermarkAndFallbackIntrospection(t *tes if _, err := verifier.Verify(context.Background(), raw); err != nil || introspectionCalls != 1 { t.Fatalf("fallback token error=%v calls=%d", err, introspectionCalls) } + evaluation = OIDCSecurityEventEvaluation{Enabled: true} + if _, err := verifier.Verify(context.Background(), raw); err != nil || introspectionCalls != 1 { + t.Fatalf("healthy push token error=%v calls=%d", err, introspectionCalls) + } evaluation = OIDCSecurityEventEvaluation{Revoked: true} if _, err := verifier.Verify(context.Background(), raw); err == nil || introspectionCalls != 1 { t.Fatalf("revoked token error=%v calls=%d", err, introspectionCalls) @@ -205,6 +210,16 @@ func TestOIDCSecurityEventEvaluationUsesWatermarkAndFallbackIntrospection(t *tes if !errors.As(err, &requestError) || requestError.Status != http.StatusServiceUnavailable { t.Fatalf("introspection outage error=%T %v", err, err) } + withoutIssuedAt := signedOIDCToken(t, issuer, "ec-key", jwt.SigningMethodES256, key, func(claims jwt.MapClaims) { delete(claims, "iat") }) + evaluation = OIDCSecurityEventEvaluation{} + introspectionAvailable = true + if _, err := verifier.Verify(context.Background(), withoutIssuedAt); err != nil { + t.Fatalf("unconfigured security events unexpectedly required iat: %v", err) + } + evaluation = OIDCSecurityEventEvaluation{Enabled: true} + if _, err := verifier.Verify(context.Background(), withoutIssuedAt); err == nil { + t.Fatal("configured security events accepted a token without iat") + } } func signedOIDCToken(t *testing.T, issuer, kid string, method jwt.SigningMethod, key any, mutate func(jwt.MapClaims)) string { diff --git a/apps/api/internal/config/config.go b/apps/api/internal/config/config.go index 3819271..310291d 100644 --- a/apps/api/internal/config/config.go +++ b/apps/api/internal/config/config.go @@ -8,8 +8,6 @@ import ( "os" "strconv" "strings" - - "github.com/google/uuid" ) const ( @@ -38,16 +36,13 @@ type Config struct { OIDCIntrospectionEnabled bool OIDCIntrospectionClientID string OIDCIntrospectionClientSecret string - OIDCSecurityEventsEnabled bool - OIDCSecurityEventsTransmitterIssuer string - OIDCSecurityEventsReceiverAudience string - OIDCSecurityEventsBearerSecret string - OIDCSecurityEventsBearerSecretNext string - OIDCSecurityEventsPublicEndpoint string - OIDCSecurityEventsStreamID string - OIDCSecurityEventsManagementTokenURL string - OIDCSecurityEventsManagementClientID string - OIDCSecurityEventsManagementClientSecret string + OIDCSecurityEventsSecretStore string + OIDCSecurityEventsSecretDir string + OIDCSecurityEventsKubernetesNamespace string + OIDCSecurityEventsKubernetesSecretName string + OIDCSecurityEventsKubernetesAPIServer string + OIDCSecurityEventsKubernetesTokenFile string + OIDCSecurityEventsKubernetesCAFile string OIDCSecurityEventsHeartbeatIntervalSeconds int OIDCSecurityEventsStaleAfterSeconds int OIDCSecurityEventsClockSkewSeconds int @@ -80,6 +75,9 @@ type Config struct { func Load() Config { globalProxy := LoadGlobalHTTPProxyStatus() appEnv := env("APP_ENV", "development") + oidcIssuer := strings.TrimRight(env("OIDC_ISSUER", ""), "/") + introspectionClientID := env("OIDC_INTROSPECTION_CLIENT_ID", "") + introspectionClientSecret := env("OIDC_INTROSPECTION_CLIENT_SECRET", "") return Config{ AppEnv: appEnv, HTTPAddr: env("HTTP_ADDR", ":8088"), @@ -94,7 +92,7 @@ func Load() Config { ServerMainInternalKey: env("SERVER_MAIN_INTERNAL_KEY", "gateway"), ServerMainInternalSecret: env("SERVER_MAIN_INTERNAL_SECRET", env("SERVER_MAIN_INTERNAL_TOKEN", "")), OIDCEnabled: env("OIDC_ENABLED", "false") == "true", - OIDCIssuer: strings.TrimRight(env("OIDC_ISSUER", ""), "/"), + OIDCIssuer: oidcIssuer, OIDCAudience: env("OIDC_AUDIENCE", ""), OIDCTenantID: env("OIDC_TENANT_ID", ""), OIDCRolePrefix: env("OIDC_ROLE_PREFIX", "gateway."), @@ -102,18 +100,15 @@ func Load() Config { OIDCJWKSCacheTTLSeconds: envInt("OIDC_JWKS_CACHE_TTL_SECONDS", 300), OIDCAcceptLegacyHS256: env("OIDC_ACCEPT_LEGACY_HS256", "true") == "true", OIDCIntrospectionEnabled: env("OIDC_INTROSPECTION_ENABLED", "false") == "true", - OIDCIntrospectionClientID: env("OIDC_INTROSPECTION_CLIENT_ID", ""), - OIDCIntrospectionClientSecret: env("OIDC_INTROSPECTION_CLIENT_SECRET", ""), - OIDCSecurityEventsEnabled: env("OIDC_SECURITY_EVENTS_ENABLED", "false") == "true", - OIDCSecurityEventsTransmitterIssuer: strings.TrimRight(env("OIDC_SECURITY_EVENTS_TRANSMITTER_ISSUER", ""), "/"), - OIDCSecurityEventsReceiverAudience: env("OIDC_SECURITY_EVENTS_RECEIVER_AUDIENCE", ""), - OIDCSecurityEventsBearerSecret: env("OIDC_SECURITY_EVENTS_BEARER_SECRET", ""), - OIDCSecurityEventsBearerSecretNext: env("OIDC_SECURITY_EVENTS_BEARER_SECRET_NEXT", ""), - OIDCSecurityEventsPublicEndpoint: env("OIDC_SECURITY_EVENTS_PUBLIC_ENDPOINT", ""), - OIDCSecurityEventsStreamID: env("OIDC_SECURITY_EVENTS_STREAM_ID", ""), - OIDCSecurityEventsManagementTokenURL: env("OIDC_SECURITY_EVENTS_MANAGEMENT_TOKEN_URL", ""), - OIDCSecurityEventsManagementClientID: env("OIDC_SECURITY_EVENTS_MANAGEMENT_CLIENT_ID", ""), - OIDCSecurityEventsManagementClientSecret: env("OIDC_SECURITY_EVENTS_MANAGEMENT_CLIENT_SECRET", ""), + OIDCIntrospectionClientID: introspectionClientID, + OIDCIntrospectionClientSecret: introspectionClientSecret, + OIDCSecurityEventsSecretStore: env("OIDC_SECURITY_EVENTS_SECRET_STORE", "file"), + OIDCSecurityEventsSecretDir: env("OIDC_SECURITY_EVENTS_SECRET_DIR", ".local-secrets/ssf"), + OIDCSecurityEventsKubernetesNamespace: env("OIDC_SECURITY_EVENTS_KUBERNETES_NAMESPACE", env("POD_NAMESPACE", "")), + OIDCSecurityEventsKubernetesSecretName: env("OIDC_SECURITY_EVENTS_KUBERNETES_SECRET_NAME", "easyai-gateway-security-events"), + OIDCSecurityEventsKubernetesAPIServer: env("OIDC_SECURITY_EVENTS_KUBERNETES_API_SERVER", "https://kubernetes.default.svc"), + OIDCSecurityEventsKubernetesTokenFile: env("OIDC_SECURITY_EVENTS_KUBERNETES_TOKEN_FILE", "/var/run/secrets/kubernetes.io/serviceaccount/token"), + OIDCSecurityEventsKubernetesCAFile: env("OIDC_SECURITY_EVENTS_KUBERNETES_CA_FILE", "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"), OIDCSecurityEventsHeartbeatIntervalSeconds: envInt("OIDC_SECURITY_EVENTS_HEARTBEAT_INTERVAL_SECONDS", 60), OIDCSecurityEventsStaleAfterSeconds: envInt("OIDC_SECURITY_EVENTS_STALE_AFTER_SECONDS", 180), OIDCSecurityEventsClockSkewSeconds: envInt("OIDC_SECURITY_EVENTS_CLOCK_SKEW_SECONDS", 60), @@ -149,6 +144,26 @@ func Load() Config { } func (c Config) Validate() error { + switch strings.ToLower(strings.TrimSpace(c.OIDCSecurityEventsSecretStore)) { + case "": + case "file": + if strings.TrimSpace(c.OIDCSecurityEventsSecretDir) == "" { + return errors.New("OIDC_SECURITY_EVENTS_SECRET_DIR is required for the file SecretStore") + } + case "kubernetes": + if strings.TrimSpace(c.OIDCSecurityEventsKubernetesNamespace) == "" || strings.TrimSpace(c.OIDCSecurityEventsKubernetesSecretName) == "" { + return errors.New("Kubernetes security event SecretStore requires namespace and Secret name") + } + default: + return errors.New("OIDC_SECURITY_EVENTS_SECRET_STORE must be file or kubernetes") + } + if c.OIDCSecurityEventsHeartbeatIntervalSeconds != 0 || c.OIDCSecurityEventsStaleAfterSeconds != 0 || c.OIDCSecurityEventsClockSkewSeconds != 0 { + if c.OIDCSecurityEventsHeartbeatIntervalSeconds <= 0 || + c.OIDCSecurityEventsStaleAfterSeconds < 2*c.OIDCSecurityEventsHeartbeatIntervalSeconds || + c.OIDCSecurityEventsClockSkewSeconds < 0 || c.OIDCSecurityEventsClockSkewSeconds > 300 { + return errors.New("OIDC security event heartbeat, stale threshold, or clock skew is invalid") + } + } if c.OIDCJITProvisioningEnabled && strings.TrimSpace(c.OIDCGatewayTenantKey) == "" { return errors.New("OIDC_GATEWAY_TENANT_KEY is required when OIDC_JIT_PROVISIONING_ENABLED=true") } @@ -185,59 +200,9 @@ func (c Config) Validate() error { } } } - if c.OIDCSecurityEventsEnabled { - if !c.OIDCEnabled { - return errors.New("OIDC_ENABLED must be true when OIDC security events are enabled") - } - if !validServiceURL(c.OIDCSecurityEventsTransmitterIssuer, c.AppEnv) || - !validSecurityEventReceiverURL(c.OIDCSecurityEventsPublicEndpoint, c.AppEnv) || - !validServiceURL(c.OIDCSecurityEventsManagementTokenURL, c.AppEnv) { - return errors.New("OIDC security event issuer, public endpoint, and management token URL must be HTTPS URLs") - } - if _, err := uuid.Parse(c.OIDCTenantID); err != nil { - return errors.New("OIDC_TENANT_ID must be a UUID when OIDC security events are enabled") - } - applicationID := strings.TrimPrefix(c.OIDCSecurityEventsReceiverAudience, "urn:easyai:ssf:receiver:") - if applicationID == c.OIDCSecurityEventsReceiverAudience { - return errors.New("OIDC security event receiver audience and stream ID are required") - } - if _, err := uuid.Parse(applicationID); err != nil { - return errors.New("OIDC security event receiver audience must contain an application UUID") - } - if _, err := uuid.Parse(c.OIDCSecurityEventsStreamID); err != nil { - return errors.New("OIDC security event stream ID must be a UUID") - } - if !validBearerSecret(c.OIDCSecurityEventsBearerSecret) || - c.OIDCSecurityEventsBearerSecretNext != "" && !validBearerSecret(c.OIDCSecurityEventsBearerSecretNext) { - return errors.New("OIDC security event bearer secrets must be 16-4096 printable non-space ASCII characters") - } - if c.OIDCSecurityEventsManagementClientID == "" || c.OIDCSecurityEventsManagementClientSecret == "" { - return errors.New("OIDC security event management machine client credentials are required") - } - if c.OIDCIntrospectionClientID == "" || c.OIDCIntrospectionClientSecret == "" { - return errors.New("RFC 7662 client credentials are required when OIDC security events are enabled") - } - if c.OIDCSecurityEventsHeartbeatIntervalSeconds <= 0 || - c.OIDCSecurityEventsStaleAfterSeconds < 2*c.OIDCSecurityEventsHeartbeatIntervalSeconds || - c.OIDCSecurityEventsClockSkewSeconds < 0 || c.OIDCSecurityEventsClockSkewSeconds > 300 { - return errors.New("OIDC security event heartbeat, stale threshold, or clock skew is invalid") - } - } return nil } -func validBearerSecret(value string) bool { - if len(value) < 16 || len(value) > 4096 { - return false - } - for _, character := range []byte(value) { - if character < 0x21 || character > 0x7e { - return false - } - } - return true -} - func (c Config) OIDCSessionEncryptionKeyBytes() ([]byte, error) { raw := strings.TrimSpace(c.OIDCSessionEncryptionKey) for _, encoding := range []*base64.Encoding{base64.RawStdEncoding, base64.StdEncoding, base64.RawURLEncoding, base64.URLEncoding} { @@ -260,25 +225,6 @@ func validPublicRedirectURL(raw string) bool { return parsed.Scheme == "http" && (parsed.Hostname() == "localhost" || parsed.Hostname() == "127.0.0.1") } -func validServiceURL(raw, appEnv string) bool { - parsed, err := url.Parse(strings.TrimSpace(raw)) - if err != nil || parsed.Host == "" || parsed.User != nil || parsed.Fragment != "" { - return false - } - if parsed.Scheme == "https" { - return true - } - return isLocalEnvironment(appEnv) && parsed.Scheme == "http" && (parsed.Hostname() == "localhost" || parsed.Hostname() == "127.0.0.1") -} - -func validSecurityEventReceiverURL(raw, appEnv string) bool { - if !validServiceURL(raw, appEnv) { - return false - } - parsed, err := url.Parse(strings.TrimSpace(raw)) - return err == nil && parsed.EscapedPath() == "/api/v1/security-events/ssf" && parsed.RawQuery == "" -} - func isLocalEnvironment(value string) bool { switch strings.ToLower(strings.TrimSpace(value)) { case "development", "dev", "local", "test": diff --git a/apps/api/internal/config/config_test.go b/apps/api/internal/config/config_test.go index 6312c7e..989d054 100644 --- a/apps/api/internal/config/config_test.go +++ b/apps/api/internal/config/config_test.go @@ -137,33 +137,46 @@ func TestValidateRejectsOfflineAccessForBrowserSession(t *testing.T) { } } -func TestSecurityEventsAreOptionalButFailFastWhenPartiallyConfigured(t *testing.T) { +func TestSecurityEventsUseDurableConnectionInsteadOfAnEnableFlag(t *testing.T) { + t.Setenv("OIDC_SECURITY_EVENTS_ENABLED", "true") + t.Setenv("OIDC_SECURITY_EVENTS_TRANSMITTER_ISSUER", "https://untrusted.example/ssf") + t.Setenv("OIDC_SECURITY_EVENTS_BEARER_SECRET", "must-not-be-loaded") + t.Setenv("OIDC_SECURITY_EVENTS_SECRET_STORE", "file") + t.Setenv("OIDC_SECURITY_EVENTS_SECRET_DIR", ".test-secrets/ssf") + + cfg := Load() + if cfg.OIDCSecurityEventsSecretStore != "file" || cfg.OIDCSecurityEventsSecretDir != ".test-secrets/ssf" { + t.Fatalf("secret directory=%q", cfg.OIDCSecurityEventsSecretDir) + } if err := (Config{}).Validate(); err != nil { - t.Fatalf("disabled security events changed existing config: %v", err) - } - cfg := Config{AppEnv: "test", OIDCEnabled: true, OIDCSecurityEventsEnabled: true} - if err := cfg.Validate(); err == nil || !strings.Contains(err.Error(), "issuer") { - t.Fatalf("partial security event config error = %v", err) - } - cfg.OIDCSecurityEventsTransmitterIssuer = "http://localhost:8080/ssf" - cfg.OIDCSecurityEventsPublicEndpoint = "http://localhost:8088/api/v1/security-events/ssf" - cfg.OIDCSecurityEventsManagementTokenURL = "http://localhost:8080/oauth/token" - cfg.OIDCTenantID = "00000000-0000-4000-8000-000000000003" - cfg.OIDCSecurityEventsReceiverAudience = "urn:easyai:ssf:receiver:00000000-0000-4000-8000-000000000002" - cfg.OIDCSecurityEventsStreamID = "00000000-0000-4000-8000-000000000001" - cfg.OIDCSecurityEventsBearerSecret = "receiver-secret-at-least-16" - cfg.OIDCSecurityEventsManagementClientID = "gateway-ssf" - cfg.OIDCSecurityEventsManagementClientSecret = "management-secret" - cfg.OIDCIntrospectionClientID = "gateway-introspection" - cfg.OIDCIntrospectionClientSecret = "introspection-secret" - cfg.OIDCSecurityEventsHeartbeatIntervalSeconds = 60 - cfg.OIDCSecurityEventsStaleAfterSeconds = 180 - cfg.OIDCSecurityEventsClockSkewSeconds = 60 - if err := cfg.Validate(); err != nil { - t.Fatalf("valid security event config: %v", err) - } - cfg.OIDCSecurityEventsPublicEndpoint = "http://localhost:8088/wrong-path" - if err := cfg.Validate(); err == nil { - t.Fatal("non-canonical security event receiver path was accepted") + t.Fatalf("an absent security event connection changed existing config: %v", err) + } +} + +func TestValidateKubernetesSecurityEventSecretStore(t *testing.T) { + config := Config{OIDCSecurityEventsSecretStore: "kubernetes", OIDCSecurityEventsKubernetesSecretName: "easyai-gateway-security-events"} + if err := config.Validate(); err == nil || !strings.Contains(err.Error(), "namespace") { + t.Fatalf("Validate() error=%v, want missing namespace", err) + } + config.OIDCSecurityEventsKubernetesNamespace = "easyai" + if err := config.Validate(); err != nil { + t.Fatalf("valid Kubernetes SecretStore was rejected: %v", err) + } +} + +func TestValidateSecurityEventTiming(t *testing.T) { + config := Config{OIDCSecurityEventsHeartbeatIntervalSeconds: 60, OIDCSecurityEventsStaleAfterSeconds: 60, OIDCSecurityEventsClockSkewSeconds: 60} + if err := config.Validate(); err == nil || !strings.Contains(err.Error(), "heartbeat") { + t.Fatalf("Validate() error=%v, want invalid stale threshold", err) + } +} + +func TestLoadSecurityEventsReusesOnlyTheRFC7662MachineClient(t *testing.T) { + t.Setenv("OIDC_INTROSPECTION_CLIENT_ID", "gateway-service") + t.Setenv("OIDC_INTROSPECTION_CLIENT_SECRET", "service-secret") + + cfg := Load() + if cfg.OIDCIntrospectionClientID != "gateway-service" || cfg.OIDCIntrospectionClientSecret != "service-secret" { + t.Fatal("RFC 7662 machine credentials were not preserved") } } diff --git a/apps/api/internal/httpapi/security_event_connection_handlers.go b/apps/api/internal/httpapi/security_event_connection_handlers.go new file mode 100644 index 0000000..bbd179a --- /dev/null +++ b/apps/api/internal/httpapi/security_event_connection_handlers.go @@ -0,0 +1,234 @@ +package httpapi + +import ( + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "net/http" + "strconv" + "strings" + + ssfreceiver "github.com/easyai/easyai-ai-gateway/apps/api/internal/securityevents" + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +type securityEventConnectionRequest struct { + TransmitterIssuer string `json:"transmitter_issuer"` +} + +type securityEventConnectionResponse struct { + Connected bool `json:"connected"` + Connection ssfreceiver.ConnectionView `json:"connection"` +} + +func (s *Server) getSecurityEventConnection(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-store") + if s.securityEventManager == nil { + writeJSON(w, http.StatusOK, map[string]any{"connected": false, "lifecycleStatus": "disconnected", "prerequisites": s.securityEventPrerequisites()}) + return + } + connection, err := s.securityEventManager.Get(r.Context()) + if errors.Is(err, store.ErrSecurityEventConnectionNotFound) { + writeJSON(w, http.StatusOK, map[string]any{"connected": false, "lifecycleStatus": "disconnected", "prerequisites": s.securityEventPrerequisites()}) + return + } + if err != nil { + writeError(w, http.StatusServiceUnavailable, "security event connection state is unavailable", "security_event_state_unavailable") + return + } + w.Header().Set("ETag", fmt.Sprintf(`W/"%d"`, connection.Version)) + writeJSON(w, http.StatusOK, map[string]any{"connected": true, "connection": connection, "prerequisites": s.securityEventPrerequisites()}) +} + +func (s *Server) putSecurityEventConnection(w http.ResponseWriter, r *http.Request) { + if s.securityEventManager == nil { + writeError(w, http.StatusConflict, "OIDC must be configured before connecting security events", "security_event_prerequisite_missing") + return + } + idempotencyKey, ok := requiredConnectionIdempotencyKey(w, r) + if !ok { + return + } + var request securityEventConnectionRequest + r.Body = http.MaxBytesReader(w, r.Body, 16*1024) + decoder := json.NewDecoder(r.Body) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&request); err != nil || strings.TrimSpace(request.TransmitterIssuer) == "" { + writeError(w, http.StatusBadRequest, "invalid security event connection request", "invalid_request") + return + } + request.TransmitterIssuer = strings.TrimRight(strings.TrimSpace(request.TransmitterIssuer), "/") + requestHash := securityEventOperationHash("connect", request.TransmitterIssuer) + if s.replaySecurityEventOperation(w, r, "connect", idempotencyKey, requestHash) { + return + } + if current, err := s.securityEventManager.Get(r.Context()); err == nil && !matchConnectionVersion(w, r, current.Version) { + return + } + connection, err := s.securityEventManager.Connect(r.Context(), request.TransmitterIssuer, idempotencyKey) + if err != nil { + writeSecurityEventConnectionError(w, err) + return + } + s.writeSecurityEventOperation(w, r, "connect", idempotencyKey, requestHash, connection) +} + +func (s *Server) verifySecurityEventConnection(w http.ResponseWriter, r *http.Request) { + idempotencyKey, requestHash, ok := s.beginSecurityEventOperation(w, r, "verify") + if !ok { + return + } + connection, err := s.securityEventManager.Verify(r.Context()) + if err != nil { + writeSecurityEventConnectionError(w, err) + return + } + s.writeSecurityEventOperation(w, r, "verify", idempotencyKey, requestHash, connection) +} + +func (s *Server) rotateSecurityEventConnectionCredential(w http.ResponseWriter, r *http.Request) { + idempotencyKey, requestHash, ok := s.beginSecurityEventOperation(w, r, "rotate") + if !ok { + return + } + connection, err := s.securityEventManager.RotateCredential(r.Context()) + if err != nil { + writeSecurityEventConnectionError(w, err) + return + } + s.writeSecurityEventOperation(w, r, "rotate", idempotencyKey, requestHash, connection) +} + +func (s *Server) deleteSecurityEventConnection(w http.ResponseWriter, r *http.Request) { + idempotencyKey, requestHash, ok := s.beginSecurityEventOperation(w, r, "disconnect") + if !ok { + return + } + connection, err := s.securityEventManager.Disconnect(r.Context()) + if err != nil { + writeSecurityEventConnectionError(w, err) + return + } + s.writeSecurityEventOperation(w, r, "disconnect", idempotencyKey, requestHash, connection) +} + +func (s *Server) securityEventPrerequisites() map[string]any { + return map[string]any{ + "oidcConfigured": s.cfg.OIDCEnabled && s.cfg.OIDCIssuer != "" && s.cfg.OIDCTenantID != "", + "introspectionClientConfigured": s.cfg.OIDCIntrospectionClientID != "" && s.cfg.OIDCIntrospectionClientSecret != "", + "publicBaseUrlConfigured": s.cfg.PublicBaseURL != "", + "managementClientId": s.cfg.OIDCIntrospectionClientID, + } +} + +func requiredConnectionIdempotencyKey(w http.ResponseWriter, r *http.Request) (string, bool) { + value := strings.TrimSpace(r.Header.Get("Idempotency-Key")) + if value == "" || len(value) > 255 { + writeError(w, http.StatusBadRequest, "Idempotency-Key is required", "idempotency_key_required") + return "", false + } + return value, true +} + +func (s *Server) beginSecurityEventOperation(w http.ResponseWriter, r *http.Request, operation string) (string, string, bool) { + if s.securityEventManager == nil { + writeError(w, http.StatusNotFound, "security event connection does not exist", "security_event_connection_not_found") + return "", "", false + } + idempotencyKey, ok := requiredConnectionIdempotencyKey(w, r) + if !ok { + return "", "", false + } + requestHash := securityEventOperationHash(operation, normalizedConnectionETag(r.Header.Get("If-Match"))) + if s.replaySecurityEventOperation(w, r, operation, idempotencyKey, requestHash) { + return "", "", false + } + connection, err := s.securityEventManager.Get(r.Context()) + if err != nil { + writeSecurityEventConnectionError(w, err) + return "", "", false + } + return idempotencyKey, requestHash, matchConnectionVersion(w, r, connection.Version) +} + +func (s *Server) replaySecurityEventOperation(w http.ResponseWriter, r *http.Request, operation, idempotencyKey, requestHash string) bool { + if s.store == nil { + return false + } + record, err := s.store.SecurityEventConnectionIdempotency(r.Context(), operation, idempotencyKey) + if errors.Is(err, store.ErrSecurityEventConnectionNotFound) { + return false + } + if err != nil { + writeError(w, http.StatusServiceUnavailable, "security event idempotency state is unavailable", "security_event_state_unavailable") + return true + } + if record.RequestHash != requestHash { + writeError(w, http.StatusConflict, "Idempotency-Key was already used for a different request", "idempotency_key_reused") + return true + } + var payload securityEventConnectionResponse + if json.Unmarshal(record.Response, &payload) != nil { + writeError(w, http.StatusServiceUnavailable, "security event idempotency response is unavailable", "security_event_state_unavailable") + return true + } + w.Header().Set("Cache-Control", "no-store") + w.Header().Set("Idempotent-Replayed", "true") + w.Header().Set("ETag", fmt.Sprintf(`W/"%d"`, payload.Connection.Version)) + writeJSON(w, http.StatusAccepted, payload) + return true +} + +func (s *Server) writeSecurityEventOperation(w http.ResponseWriter, r *http.Request, operation, idempotencyKey, requestHash string, connection ssfreceiver.ConnectionView) { + payload := securityEventConnectionResponse{Connected: true, Connection: connection} + encoded, _ := json.Marshal(payload) + if s.store != nil { + if err := s.store.RecordSecurityEventConnectionIdempotency(r.Context(), operation, idempotencyKey, requestHash, encoded); err != nil && s.logger != nil { + s.logger.Error("security event idempotency result could not be recorded", "error_category", "idempotency_store_failed", "operation", operation) + } + } + w.Header().Set("Cache-Control", "no-store") + w.Header().Set("ETag", fmt.Sprintf(`W/"%d"`, connection.Version)) + writeJSON(w, http.StatusAccepted, payload) +} + +func securityEventOperationHash(operation, canonicalRequest string) string { + digest := sha256.Sum256([]byte(operation + "\x00" + canonicalRequest)) + return fmt.Sprintf("%x", digest[:]) +} + +func normalizedConnectionETag(value string) string { + value = strings.TrimSpace(value) + value = strings.TrimPrefix(value, "W/") + return strings.Trim(value, `"`) +} + +func matchConnectionVersion(w http.ResponseWriter, r *http.Request, expected int64) bool { + value := strings.TrimSpace(r.Header.Get("If-Match")) + value = strings.TrimPrefix(value, "W/") + value = strings.Trim(value, `"`) + version, err := strconv.ParseInt(value, 10, 64) + if err != nil { + writeError(w, http.StatusPreconditionRequired, "If-Match is required", "if_match_required") + return false + } + if version != expected { + writeError(w, http.StatusPreconditionFailed, "security event connection version changed", "version_conflict") + return false + } + return true +} + +func writeSecurityEventConnectionError(w http.ResponseWriter, err error) { + switch { + case errors.Is(err, store.ErrSecurityEventConnectionNotFound): + writeError(w, http.StatusNotFound, "security event connection does not exist", "security_event_connection_not_found") + case errors.Is(err, store.ErrSecurityEventConnectionConflict): + writeError(w, http.StatusConflict, "security event connection conflicts with current state", "security_event_connection_conflict") + case strings.Contains(err.Error(), "configured"), strings.Contains(err.Error(), "invalid"), strings.Contains(err.Error(), "HTTPS"): + writeError(w, http.StatusConflict, "security event connection prerequisites are incomplete", "security_event_prerequisite_missing") + default: + writeError(w, http.StatusBadGateway, "authentication center security event service is unavailable", "security_event_transmitter_unavailable") + } +} diff --git a/apps/api/internal/httpapi/security_event_connection_handlers_test.go b/apps/api/internal/httpapi/security_event_connection_handlers_test.go new file mode 100644 index 0000000..f92ade8 --- /dev/null +++ b/apps/api/internal/httpapi/security_event_connection_handlers_test.go @@ -0,0 +1,51 @@ +package httpapi + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/config" +) + +func TestSecurityEventConnectionWriteHeaders(t *testing.T) { + request := httptest.NewRequest(http.MethodPost, "/connection/verify", nil) + recorder := httptest.NewRecorder() + if _, ok := requiredConnectionIdempotencyKey(recorder, request); ok || recorder.Code != http.StatusBadRequest { + t.Fatalf("missing Idempotency-Key status=%d", recorder.Code) + } + + request = httptest.NewRequest(http.MethodPost, "/connection/verify", nil) + request.Header.Set("If-Match", `W/"7"`) + recorder = httptest.NewRecorder() + if !matchConnectionVersion(recorder, request, 7) { + t.Fatalf("valid weak ETag was rejected: status=%d", recorder.Code) + } + + request.Header.Set("If-Match", `"6"`) + recorder = httptest.NewRecorder() + if matchConnectionVersion(recorder, request, 7) || recorder.Code != http.StatusPreconditionFailed { + t.Fatalf("stale ETag status=%d", recorder.Code) + } + if normalizedConnectionETag(`W/"7"`) != "7" || + securityEventOperationHash("verify", "7") == securityEventOperationHash("verify", "8") { + t.Fatal("security event idempotency request fingerprint is not stable") + } +} + +func TestSecurityEventPrerequisitesNeverExposeSecrets(t *testing.T) { + server := &Server{cfg: config.Config{ + OIDCEnabled: true, OIDCIssuer: "https://auth.example/issuer/shared", OIDCTenantID: "stable-tenant", + OIDCIntrospectionClientID: "gateway-machine", OIDCIntrospectionClientSecret: "must-not-escape", + PublicBaseURL: "https://gateway.example", + }} + payload, err := json.Marshal(server.securityEventPrerequisites()) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(payload), "must-not-escape") || strings.Contains(strings.ToLower(string(payload)), "secret") { + t.Fatalf("secret escaped in prerequisites: %s", payload) + } +} diff --git a/apps/api/internal/httpapi/security_events.go b/apps/api/internal/httpapi/security_events.go index bbe48a8..addf64e 100644 --- a/apps/api/internal/httpapi/security_events.go +++ b/apps/api/internal/httpapi/security_events.go @@ -18,5 +18,9 @@ import "net/http" // @Failure 503 {object} map[string]string // @Router /api/v1/security-events/ssf [post] func (s *Server) receiveSecurityEvent(w http.ResponseWriter, r *http.Request) { + if s.securityEventReceiver == nil { + http.NotFound(w, r) + return + } s.securityEventReceiver.ServeHTTP(w, r) } diff --git a/apps/api/internal/httpapi/server.go b/apps/api/internal/httpapi/server.go index 2ab5fc1..1b03891 100644 --- a/apps/api/internal/httpapi/server.go +++ b/apps/api/internal/httpapi/server.go @@ -30,6 +30,7 @@ type Server struct { logger *slog.Logger geminiUploadSessions sync.Map securityEventReceiver http.Handler + securityEventManager *ssfreceiver.ConnectionManager } type oidcPublicClient interface { @@ -66,46 +67,29 @@ func NewServerWithContext(ctx context.Context, cfg config.Config, db *store.Stor server.auth.ServerMainInternalKey = cfg.ServerMainInternalKey server.auth.ServerMainInternalSecret = cfg.ServerMainInternalSecret securityEventMetrics := &ssfreceiver.Metrics{} - var securityEventService *ssfreceiver.Service - if cfg.OIDCSecurityEventsEnabled { - ssfVerifier, err := ssfreceiver.NewVerifier(ssfreceiver.VerifierConfig{ - TransmitterIssuer: cfg.OIDCSecurityEventsTransmitterIssuer, - Audience: cfg.OIDCSecurityEventsReceiverAudience, SubjectIssuer: cfg.OIDCIssuer, - TenantID: cfg.OIDCTenantID, StreamID: cfg.OIDCSecurityEventsStreamID, - ClockSkew: time.Duration(cfg.OIDCSecurityEventsClockSkewSeconds) * time.Second, - }) + if cfg.OIDCEnabled { + secretStore, err := securityEventSecretStore(cfg) if err != nil { - panic("invalid OIDC security event verifier configuration: " + err.Error()) + panic("invalid OIDC security event secret store: " + err.Error()) } - ssfVerifier.SetMetrics(securityEventMetrics) - securityEventService, err = ssfreceiver.NewService(db, ssfreceiver.ServiceConfig{ - Issuer: cfg.OIDCSecurityEventsTransmitterIssuer, Audience: cfg.OIDCSecurityEventsReceiverAudience, - StreamID: cfg.OIDCSecurityEventsStreamID, ManagementTokenURL: cfg.OIDCSecurityEventsManagementTokenURL, - ManagementClientID: cfg.OIDCSecurityEventsManagementClientID, ManagementSecret: cfg.OIDCSecurityEventsManagementClientSecret, + manager, err := ssfreceiver.NewConnectionManager(ctx, db, secretStore, ssfreceiver.ConnectionManagerConfig{ + AppEnv: cfg.AppEnv, OIDCEnabled: cfg.OIDCEnabled, OIDCIssuer: cfg.OIDCIssuer, OIDCTenantID: cfg.OIDCTenantID, + ManagementClientID: cfg.OIDCIntrospectionClientID, ManagementClientSecret: cfg.OIDCIntrospectionClientSecret, + PublicBaseURL: cfg.PublicBaseURL, HeartbeatInterval: time.Duration(cfg.OIDCSecurityEventsHeartbeatIntervalSeconds) * time.Second, StaleAfter: time.Duration(cfg.OIDCSecurityEventsStaleAfterSeconds) * time.Second, - }) + ClockSkew: time.Duration(cfg.OIDCSecurityEventsClockSkewSeconds) * time.Second, + }, securityEventMetrics) if err != nil { - panic("invalid OIDC security event heartbeat configuration: " + err.Error()) + panic("initialize OIDC security event connection manager: " + err.Error()) } - securityEventService.SetMetrics(securityEventMetrics) - if err := securityEventService.Start(ctx); err != nil { - panic("OIDC security event state initialization failed") - } - receiver, err := ssfreceiver.NewHandler( - ssfVerifier, db, cfg.OIDCSecurityEventsTransmitterIssuer, cfg.OIDCSecurityEventsReceiverAudience, - cfg.OIDCSecurityEventsStreamID, cfg.OIDCSecurityEventsBearerSecret, cfg.OIDCSecurityEventsBearerSecretNext, - ) - if err != nil { - panic("invalid OIDC security event receiver configuration: " + err.Error()) - } - receiver.SetMetrics(securityEventMetrics) - server.securityEventReceiver = receiver + server.securityEventManager = manager + server.securityEventReceiver = manager } if cfg.OIDCEnabled { var evaluator func(context.Context, auth.OIDCSecurityEventIdentity) (auth.OIDCSecurityEventEvaluation, error) - if securityEventService != nil { - evaluator = securityEventService.Evaluate + if server.securityEventManager != nil { + evaluator = server.securityEventManager.Evaluate } verifier, err := auth.NewOIDCVerifier(auth.OIDCConfig{ Issuer: cfg.OIDCIssuer, Audience: cfg.OIDCAudience, TenantID: cfg.OIDCTenantID, @@ -164,7 +148,7 @@ func NewServerWithContext(ctx context.Context, cfg config.Config, db *store.Stor mux := http.NewServeMux() mux.HandleFunc("GET /healthz", server.health) mux.HandleFunc("GET /readyz", server.ready) - mux.Handle("GET /metrics", securityEventMetrics.Handler(db, cfg.OIDCSecurityEventsTransmitterIssuer, cfg.OIDCSecurityEventsReceiverAudience, cfg.OIDCSecurityEventsEnabled)) + mux.Handle("GET /metrics", securityEventMetrics.DynamicHandler(db)) mux.HandleFunc("GET /static/simulation/{asset}", serveSimulationAsset) mux.HandleFunc("GET /static/generated/{asset}", server.serveGeneratedStaticAsset) mux.HandleFunc("GET /static/uploaded/{asset}", server.serveUploadedStaticAsset) @@ -175,9 +159,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) - if server.securityEventReceiver != nil { - mux.HandleFunc("POST /api/v1/security-events/ssf", server.receiveSecurityEvent) - } + 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))) mux.Handle("GET /api/v1/public/catalog/base-models", server.auth.Require(auth.PermissionPublic, http.HandlerFunc(server.listBaseModels))) @@ -247,6 +229,11 @@ 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/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))) + mux.Handle("POST /api/admin/system/identity/security-events/connection/rotate-credential", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.rotateSecurityEventConnectionCredential))) + mux.Handle("DELETE /api/admin/system/identity/security-events/connection", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.deleteSecurityEventConnection))) mux.Handle("GET /api/admin/system/file-storage/channels", server.requireAdmin(auth.PermissionPower, http.HandlerFunc(server.listFileStorageChannels))) mux.Handle("POST /api/admin/system/file-storage/channels", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.createFileStorageChannel))) mux.Handle("PATCH /api/admin/system/file-storage/channels/{channelID}", server.requireAdmin(auth.PermissionManager, http.HandlerFunc(server.updateFileStorageChannel))) @@ -324,6 +311,25 @@ func NewServerWithContext(ctx context.Context, cfg config.Config, db *store.Stor return server.recover(server.cors(server.protectOIDCSessionCookie(mux))) } +func securityEventSecretStore(cfg config.Config) (ssfreceiver.SecretStore, error) { + switch strings.ToLower(strings.TrimSpace(cfg.OIDCSecurityEventsSecretStore)) { + case "", "file": + directory := cfg.OIDCSecurityEventsSecretDir + if directory == "" { + directory = ".local-secrets/ssf" + } + return ssfreceiver.NewFileSecretStore(directory) + case "kubernetes": + return ssfreceiver.NewKubernetesSecretStore(ssfreceiver.KubernetesSecretStoreConfig{ + Namespace: cfg.OIDCSecurityEventsKubernetesNamespace, SecretName: cfg.OIDCSecurityEventsKubernetesSecretName, + APIServer: cfg.OIDCSecurityEventsKubernetesAPIServer, TokenFile: cfg.OIDCSecurityEventsKubernetesTokenFile, + CAFile: cfg.OIDCSecurityEventsKubernetesCAFile, + }) + default: + return nil, errors.New("unsupported OIDC security event SecretStore") + } +} + func oidcSessionRequestError(err error) error { switch { case err == nil: @@ -361,7 +367,7 @@ func (s *Server) cors(next http.Handler) http.Handler { w.Header().Set("Access-Control-Allow-Origin", origin) w.Header().Set("Vary", "Origin") w.Header().Set("Access-Control-Allow-Credentials", "true") - w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type, X-Comfy-Api-Key, X-Goog-Api-Key, X-Goog-Upload-Protocol, X-Goog-Upload-Command, X-Goog-Upload-Header-Content-Length, X-Goog-Upload-Header-Content-Type, X-Goog-Upload-Offset, X-Async, X-EasyAI-Conversation-ID") + w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type, Idempotency-Key, If-Match, X-Comfy-Api-Key, X-Goog-Api-Key, X-Goog-Upload-Protocol, X-Goog-Upload-Command, X-Goog-Upload-Header-Content-Length, X-Goog-Upload-Header-Content-Type, X-Goog-Upload-Offset, X-Async, X-EasyAI-Conversation-ID") w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, PATCH, DELETE, OPTIONS") } if r.Method == http.MethodOptions { diff --git a/apps/api/internal/securityevents/connection_manager.go b/apps/api/internal/securityevents/connection_manager.go new file mode 100644 index 0000000..1a02463 --- /dev/null +++ b/apps/api/internal/securityevents/connection_manager.go @@ -0,0 +1,940 @@ +package securityevents + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/tls" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/netip" + "net/url" + "strings" + "sync" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/auth" + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" + "github.com/google/uuid" +) + +const ( + sessionRevokedEvent = "https://schemas.openid.net/secevent/caep/event-type/session-revoked" + pushDeliveryMethod = "urn:ietf:rfc:8935" +) + +type ConnectionRepository interface { + Repository + StateRepository + CreateSecurityEventConnection(context.Context, store.CreateSecurityEventConnectionInput) (store.SecurityEventConnection, error) + SecurityEventConnection(context.Context) (store.SecurityEventConnection, error) + BindSecurityEventConnection(context.Context, string, string, string, string, int64) (store.SecurityEventConnection, error) + UpdateSecurityEventConnectionLifecycle(context.Context, string, string, *string) (store.SecurityEventConnection, error) + SetSecurityEventNextCredential(context.Context, string, string) (store.SecurityEventConnection, error) + PromoteSecurityEventCredential(context.Context, string, string) (store.SecurityEventConnection, error) + DeleteSecurityEventConnection(context.Context, string) error + SecurityEventMetrics(context.Context, string, string) (string, *time.Time, error) +} + +type ConnectionManagerConfig struct { + AppEnv string + OIDCEnabled bool + OIDCIssuer string + OIDCTenantID string + ManagementClientID string + ManagementClientSecret string + PublicBaseURL string + HeartbeatInterval time.Duration + StaleAfter time.Duration + ClockSkew time.Duration + BootstrapDuration time.Duration + CredentialOverlapDuration time.Duration + RetirementDuration time.Duration + HTTPClient *http.Client +} + +type ConnectionView struct { + ConnectionID string `json:"connectionId"` + TransmitterIssuer string `json:"transmitterIssuer"` + ReceiverEndpoint string `json:"receiverEndpoint"` + Audience *string `json:"audience,omitempty"` + StreamID *string `json:"streamId,omitempty"` + LifecycleStatus string `json:"lifecycleStatus"` + HealthMode string `json:"healthMode"` + ManagementClientID string `json:"managementClientId"` + LastVerificationAt *time.Time `json:"lastVerificationAt,omitempty"` + LastErrorCategory *string `json:"lastErrorCategory,omitempty"` + Version int64 `json:"version"` +} + +type connectionRuntime struct { + handler *Handler + service *Service + cancel context.CancelFunc +} + +type ConnectionManager struct { + ctx context.Context + repository ConnectionRepository + secrets SecretStore + config ConnectionManagerConfig + metrics *Metrics + mutex sync.RWMutex + operation sync.Mutex + runtime *connectionRuntime +} + +type transmitterConfiguration struct { + SpecVersion string `json:"spec_version"` + Issuer string `json:"issuer"` + JWKSURI string `json:"jwks_uri"` + ConfigurationEndpoint string `json:"configuration_endpoint"` + StatusEndpoint string `json:"status_endpoint"` + VerificationEndpoint string `json:"verification_endpoint"` + DeliveryMethodsSupported []string `json:"delivery_methods_supported"` +} + +type streamConfiguration struct { + StreamID string `json:"stream_id"` + Issuer string `json:"iss"` + Audience string `json:"aud"` + EventsRequested []string `json:"events_requested"` + Delivery struct { + Method string `json:"method"` + EndpointURL string `json:"endpoint_url"` + } `json:"delivery"` + Description *string `json:"description,omitempty"` +} + +type remoteHTTPError struct{ status int } + +func (e remoteHTTPError) Error() string { + return fmt.Sprintf("SSF endpoint returned HTTP %d", e.status) +} + +func NewConnectionManager(ctx context.Context, repository ConnectionRepository, secrets SecretStore, config ConnectionManagerConfig, metrics *Metrics) (*ConnectionManager, error) { + if repository == nil || secrets == nil || !config.OIDCEnabled || config.OIDCIssuer == "" || config.OIDCTenantID == "" { + return nil, errors.New("security event connection prerequisites are incomplete") + } + if config.HeartbeatInterval <= 0 { + config.HeartbeatInterval = time.Minute + } + if config.StaleAfter < 2*config.HeartbeatInterval { + config.StaleAfter = 3 * time.Minute + } + if config.ClockSkew == 0 { + config.ClockSkew = time.Minute + } + if config.BootstrapDuration == 0 { + config.BootstrapDuration = 6 * time.Minute + } + if config.CredentialOverlapDuration == 0 { + config.CredentialOverlapDuration = 3 * time.Minute + } + if config.RetirementDuration == 0 { + config.RetirementDuration = 6 * time.Minute + } + manager := &ConnectionManager{ctx: ctx, repository: repository, secrets: secrets, config: config, metrics: metrics} + connection, err := repository.SecurityEventConnection(ctx) + if errors.Is(err, store.ErrSecurityEventConnectionNotFound) { + return manager, nil + } + if err != nil { + return nil, fmt.Errorf("load security event connection: %w", err) + } + if connection.LifecycleStatus == "retiring" { + if connection.StreamID != nil && connection.Audience != nil { + _ = manager.activate(ctx, connection) + } + go manager.finishRetirement(connection) + return manager, nil + } + if connection.LifecycleStatus == "disconnect_pending" { + if connection.StreamID != nil && connection.Audience != nil { + _ = manager.activate(ctx, connection) + } + go manager.retryDisconnect(connection.ConnectionID) + return manager, nil + } + if connection.StreamID != nil && connection.Audience != nil { + if err := manager.activate(ctx, connection); err != nil { + category := "credential_unavailable" + _, _ = repository.UpdateSecurityEventConnectionLifecycle(ctx, connection.ConnectionID, "degraded", &category) + } + } + return manager, nil +} + +func (m *ConnectionManager) Get(ctx context.Context) (ConnectionView, error) { + connection, err := m.repository.SecurityEventConnection(ctx) + if err != nil { + return ConnectionView{}, err + } + return m.view(connection), nil +} + +func (m *ConnectionManager) Connect(ctx context.Context, issuer, idempotencyKey string) (ConnectionView, error) { + m.operation.Lock() + defer m.operation.Unlock() + issuer = strings.TrimRight(strings.TrimSpace(issuer), "/") + if idempotencyKey == "" { + return ConnectionView{}, errors.New("idempotency key is required") + } + if err := m.validatePrerequisites(issuer); err != nil { + return ConnectionView{}, err + } + existing, err := m.repository.SecurityEventConnection(ctx) + if err == nil { + if existing.TransmitterIssuer != issuer { + return ConnectionView{}, store.ErrSecurityEventConnectionConflict + } + if existing.StreamID != nil { + return m.view(existing), nil + } + return m.resumeConnecting(ctx, existing) + } + if !errors.Is(err, store.ErrSecurityEventConnectionNotFound) { + return ConnectionView{}, err + } + connectionID := uuid.NewString() + reference := "ssf-push-" + connectionID + secret, err := randomPushBearer() + if err != nil { + return ConnectionView{}, err + } + defer clear(secret) + if err := m.secrets.Put(ctx, reference, secret); err != nil { + return ConnectionView{}, err + } + endpoint := strings.TrimRight(m.config.PublicBaseURL, "/") + "/api/v1/security-events/ssf" + connection, err := m.repository.CreateSecurityEventConnection(ctx, store.CreateSecurityEventConnectionInput{ + ConnectionID: connectionID, TransmitterIssuer: issuer, EndpointURL: endpoint, + CredentialRef: reference, IdempotencyKey: idempotencyKey, + }) + if err != nil { + _ = m.secrets.Delete(ctx, reference) + return ConnectionView{}, err + } + return m.resumeConnecting(ctx, connection) +} + +func (m *ConnectionManager) resumeConnecting(ctx context.Context, connection store.SecurityEventConnection) (ConnectionView, error) { + secret, err := m.secrets.Get(ctx, connection.CredentialRef) + if err != nil { + return ConnectionView{}, err + } + defer clear(secret) + configuration, client, err := m.discover(ctx, connection.TransmitterIssuer) + if err != nil { + return ConnectionView{}, m.connectionError(ctx, connection, "discovery_failed", err) + } + token, err := m.managementToken(ctx, client, "ssf.stream.read ssf.stream.verify ssf.stream.manage") + if err != nil { + return ConnectionView{}, m.connectionError(ctx, connection, "management_token_failed", err) + } + description := "easyai-gateway:" + connection.ConnectionID + stream, err := m.findStream(ctx, client, configuration, token, description, connection.EndpointURL) + if errors.Is(err, store.ErrSecurityEventConnectionNotFound) { + stream, err = m.createStream(ctx, client, configuration, token, connection.EndpointURL, string(secret), description) + } + if err != nil { + return ConnectionView{}, m.connectionError(ctx, connection, "stream_create_failed", err) + } + if err := validateCreatedStream(stream, connection.TransmitterIssuer, connection.EndpointURL); err != nil { + return ConnectionView{}, m.connectionError(ctx, connection, "stream_response_invalid", err) + } + connection, err = m.repository.BindSecurityEventConnection(ctx, connection.ConnectionID, stream.Audience, stream.StreamID, "verifying", connection.Version) + if err != nil { + return ConnectionView{}, err + } + started := time.Now().UTC() + if err := m.activate(ctx, connection); err != nil { + return ConnectionView{}, m.connectionError(ctx, connection, "receiver_activation_failed", err) + } + go m.finishAutomaticVerification(connection.ConnectionID, configuration, client, token, started, false) + return m.view(connection), nil +} + +func (m *ConnectionManager) Verify(ctx context.Context) (ConnectionView, error) { + connection, err := m.repository.SecurityEventConnection(ctx) + if err != nil { + return ConnectionView{}, err + } + m.mutex.RLock() + runtime := m.runtime + m.mutex.RUnlock() + if runtime == nil || runtime.service == nil { + return ConnectionView{}, errors.New("security event receiver is unavailable") + } + started := time.Now().UTC() + runtime.service.RequestVerification(ctx) + if connection.LifecycleStatus == "verifying" || connection.LifecycleStatus == "rotating" { + configuration, client, discoverErr := m.discover(ctx, connection.TransmitterIssuer) + if discoverErr != nil { + return ConnectionView{}, discoverErr + } + token, tokenErr := m.managementToken(ctx, client, "ssf.stream.read ssf.stream.verify ssf.stream.manage") + if tokenErr != nil { + return ConnectionView{}, tokenErr + } + go m.finishAutomaticVerification(connection.ConnectionID, configuration, client, token, started, connection.LifecycleStatus == "rotating") + } + return m.view(connection), nil +} + +func (m *ConnectionManager) RotateCredential(ctx context.Context) (ConnectionView, error) { + m.operation.Lock() + defer m.operation.Unlock() + connection, err := m.repository.SecurityEventConnection(ctx) + if err != nil { + return ConnectionView{}, err + } + if connection.StreamID == nil || connection.Audience == nil { + return ConnectionView{}, store.ErrSecurityEventConnectionConflict + } + var nextReference string + var next []byte + if connection.NextCredentialRef == nil { + nextReference = "ssf-push-" + connection.ConnectionID + "-next-" + uuid.NewString() + next, err = randomPushBearer() + if err != nil { + return ConnectionView{}, err + } + if err := m.secrets.Put(ctx, nextReference, next); err != nil { + clear(next) + return ConnectionView{}, err + } + connection, err = m.repository.SetSecurityEventNextCredential(ctx, connection.ConnectionID, nextReference) + if err != nil { + clear(next) + _ = m.secrets.Delete(ctx, nextReference) + return ConnectionView{}, err + } + } else { + nextReference = *connection.NextCredentialRef + next, err = m.secrets.Get(ctx, nextReference) + if err != nil { + return ConnectionView{}, err + } + } + defer clear(next) + configuration, client, err := m.discover(ctx, connection.TransmitterIssuer) + if err != nil { + return ConnectionView{}, err + } + token, err := m.managementToken(ctx, client, "ssf.stream.read ssf.stream.verify ssf.stream.manage") + if err != nil { + return ConnectionView{}, err + } + if err := m.setStreamStatus(ctx, client, configuration, token, *connection.StreamID, "paused", "credential_rotation"); err != nil { + return ConnectionView{}, err + } + if err := m.patchStreamCredential(ctx, client, configuration, token, *connection.StreamID, connection.EndpointURL, string(next)); err != nil { + return ConnectionView{}, err + } + started := time.Now().UTC() + if err := m.activate(ctx, connection); err != nil { + return ConnectionView{}, err + } + go m.finishAutomaticVerification(connection.ConnectionID, configuration, client, token, started, true) + return m.view(connection), nil +} + +func (m *ConnectionManager) Disconnect(ctx context.Context) (ConnectionView, error) { + m.operation.Lock() + defer m.operation.Unlock() + connection, err := m.repository.SecurityEventConnection(ctx) + if err != nil { + return ConnectionView{}, err + } + if connection.StreamID != nil { + configuration, client, discoverErr := m.discover(ctx, connection.TransmitterIssuer) + if discoverErr == nil { + token, tokenErr := m.managementToken(ctx, client, "ssf.stream.read ssf.stream.verify ssf.stream.manage") + if tokenErr == nil { + discoverErr = m.deleteStream(ctx, client, configuration, token, *connection.StreamID) + if isRemoteHTTPStatus(discoverErr, http.StatusNotFound) { + discoverErr = nil + } + } + } + if discoverErr != nil { + category := "remote_disconnect_failed" + connection, _ = m.repository.UpdateSecurityEventConnectionLifecycle(ctx, connection.ConnectionID, "disconnect_pending", &category) + go m.retryDisconnect(connection.ConnectionID) + return m.view(connection), nil + } + } + connection, err = m.repository.UpdateSecurityEventConnectionLifecycle(ctx, connection.ConnectionID, "retiring", nil) + if err != nil { + return ConnectionView{}, err + } + go m.finishRetirement(connection) + return m.view(connection), nil +} + +func (m *ConnectionManager) retryDisconnect(connectionID string) { + delay := time.Second + for { + select { + case <-m.ctx.Done(): + return + case <-time.After(delay): + } + connection, err := m.repository.SecurityEventConnection(m.ctx) + if err != nil || connection.ConnectionID != connectionID || connection.LifecycleStatus != "disconnect_pending" || connection.StreamID == nil { + return + } + configuration, client, err := m.discover(m.ctx, connection.TransmitterIssuer) + if err == nil { + var token string + token, err = m.managementToken(m.ctx, client, "ssf.stream.read ssf.stream.verify ssf.stream.manage") + if err == nil { + err = m.deleteStream(m.ctx, client, configuration, token, *connection.StreamID) + if isRemoteHTTPStatus(err, http.StatusNotFound) { + err = nil + } + } + } + if err == nil { + connection, err = m.repository.UpdateSecurityEventConnectionLifecycle(m.ctx, connectionID, "retiring", nil) + if err == nil { + go m.finishRetirement(connection) + } + return + } + if delay < time.Minute { + delay *= 2 + } + } +} + +func (m *ConnectionManager) ServeHTTP(w http.ResponseWriter, r *http.Request) { + m.mutex.RLock() + runtime := m.runtime + m.mutex.RUnlock() + if runtime == nil || runtime.handler == nil { + w.Header().Set("Cache-Control", "no-store") + if _, err := m.repository.SecurityEventConnection(r.Context()); errors.Is(err, store.ErrSecurityEventConnectionNotFound) { + http.NotFound(w, r) + return + } + w.Header().Set("Retry-After", "1") + w.WriteHeader(http.StatusServiceUnavailable) + return + } + runtime.handler.ServeHTTP(w, r) +} + +func (m *ConnectionManager) Evaluate(ctx context.Context, identity auth.OIDCSecurityEventIdentity) (auth.OIDCSecurityEventEvaluation, error) { + connection, err := m.repository.SecurityEventConnection(ctx) + if errors.Is(err, store.ErrSecurityEventConnectionNotFound) { + return auth.OIDCSecurityEventEvaluation{}, nil + } + if err != nil { + return auth.OIDCSecurityEventEvaluation{Enabled: true, RequireIntrospection: true}, err + } + m.mutex.RLock() + runtime := m.runtime + m.mutex.RUnlock() + if runtime == nil || runtime.service == nil { + if connection.Audience == nil { + return auth.OIDCSecurityEventEvaluation{Enabled: true, RequireIntrospection: true}, nil + } + result, evaluateErr := m.repository.EvaluateOIDCSecurityEvent( + ctx, connection.TransmitterIssuer, *connection.Audience, identity.Issuer, identity.TenantID, identity.Subject, + identity.IssuedAt, time.Now().UTC(), m.config.StaleAfter, + ) + if result.Revoked && m.metrics != nil { + m.metrics.ObserveWatermarkRejection() + } + return auth.OIDCSecurityEventEvaluation{Enabled: true, Revoked: result.Revoked, RequireIntrospection: true}, evaluateErr + } + evaluation, err := runtime.service.Evaluate(ctx, identity) + evaluation.Enabled = true + if connection.LifecycleStatus != "enabled" && connection.LifecycleStatus != "bootstrap" { + evaluation.RequireIntrospection = true + } + return evaluation, err +} + +func (m *ConnectionManager) activate(ctx context.Context, connection store.SecurityEventConnection) error { + if connection.StreamID == nil || connection.Audience == nil { + return errors.New("security event connection binding is incomplete") + } + current, err := m.secrets.Get(ctx, connection.CredentialRef) + if err != nil { + return err + } + defer clear(current) + var next []byte + if connection.NextCredentialRef != nil { + next, err = m.secrets.Get(ctx, *connection.NextCredentialRef) + if err != nil { + return err + } + defer clear(next) + } + issuerURL, err := validatedIssuerURL(connection.TransmitterIssuer, m.config.AppEnv) + if err != nil { + return err + } + safeClient := m.config.HTTPClient + if safeClient == nil { + safeClient = safeHTTPClient(issuerURL, m.config.AppEnv) + } + verifier, err := NewVerifier(VerifierConfig{ + TransmitterIssuer: connection.TransmitterIssuer, Audience: *connection.Audience, + SubjectIssuer: m.config.OIDCIssuer, TenantID: m.config.OIDCTenantID, StreamID: *connection.StreamID, + ClockSkew: m.config.ClockSkew, HTTPClient: safeClient, + }) + if err != nil { + return err + } + verifier.SetMetrics(m.metrics) + handler, err := NewHandler(verifier, m.repository, connection.TransmitterIssuer, *connection.Audience, *connection.StreamID, string(current), string(next)) + if err != nil { + return err + } + handler.SetMetrics(m.metrics) + service, err := NewService(m.repository, ServiceConfig{ + Issuer: connection.TransmitterIssuer, Audience: *connection.Audience, StreamID: *connection.StreamID, + ManagementTokenURL: strings.TrimRight(m.config.OIDCIssuer, "/") + "/token", + ManagementClientID: m.config.ManagementClientID, ManagementSecret: m.config.ManagementClientSecret, + HeartbeatInterval: m.config.HeartbeatInterval, StaleAfter: m.config.StaleAfter, + HTTPClient: safeClient, + }) + if err != nil { + return err + } + service.SetMetrics(m.metrics) + runtimeContext, cancel := context.WithCancel(m.ctx) + if err := service.Initialize(runtimeContext); err != nil { + cancel() + return err + } + m.mutex.Lock() + previous := m.runtime + m.runtime = &connectionRuntime{handler: handler, service: service, cancel: cancel} + m.mutex.Unlock() + if previous != nil && previous.cancel != nil { + previous.cancel() + } + go service.Run(runtimeContext) + return nil +} + +func (m *ConnectionManager) stopRuntime() { + m.mutex.Lock() + previous := m.runtime + m.runtime = nil + m.mutex.Unlock() + if previous != nil && previous.cancel != nil { + previous.cancel() + } +} + +func (m *ConnectionManager) finishAutomaticVerification(connectionID string, configuration transmitterConfiguration, client *http.Client, token string, started time.Time, rotating bool) { + ctx, cancel := context.WithTimeout(m.ctx, 90*time.Second) + defer cancel() + ticker := time.NewTicker(250 * time.Millisecond) + defer ticker.Stop() + for { + connection, err := m.repository.SecurityEventConnection(ctx) + if err != nil || connection.ConnectionID != connectionID || connection.StreamID == nil { + return + } + _, verifiedAt, _ := m.repository.SecurityEventMetrics(ctx, connection.TransmitterIssuer, valueOrEmpty(connection.Audience)) + if verifiedAt != nil && !verifiedAt.Before(started) { + if err := m.setStreamStatus(ctx, client, configuration, token, *connection.StreamID, "enabled", "verification_succeeded"); err != nil { + category := "stream_enable_failed" + lifecycle := "verifying" + if rotating { + lifecycle = "rotating" + } + _, _ = m.repository.UpdateSecurityEventConnectionLifecycle(ctx, connectionID, lifecycle, &category) + return + } + if rotating && connection.NextCredentialRef != nil { + oldReference, nextReference := connection.CredentialRef, *connection.NextCredentialRef + select { + case <-m.ctx.Done(): + return + case <-time.After(m.config.CredentialOverlapDuration): + } + connection, err = m.repository.PromoteSecurityEventCredential(m.ctx, connectionID, nextReference) + if err == nil { + _ = m.secrets.Delete(m.ctx, oldReference) + _ = m.activate(m.ctx, connection) + go m.finishBootstrap(connectionID) + } + return + } + connection, _ = m.repository.UpdateSecurityEventConnectionLifecycle(ctx, connectionID, "bootstrap", nil) + go m.finishBootstrap(connectionID) + return + } + select { + case <-ctx.Done(): + category := "verification_timeout" + lifecycle := "verifying" + if rotating { + lifecycle = "rotating" + } + _, _ = m.repository.UpdateSecurityEventConnectionLifecycle(m.ctx, connectionID, lifecycle, &category) + return + case <-ticker.C: + } + } +} + +func (m *ConnectionManager) finishBootstrap(connectionID string) { + select { + case <-m.ctx.Done(): + return + case <-time.After(m.config.BootstrapDuration): + } + connection, err := m.repository.SecurityEventConnection(m.ctx) + if err == nil && connection.ConnectionID == connectionID && connection.LifecycleStatus == "bootstrap" { + mode, _, metricsErr := m.repository.SecurityEventMetrics(m.ctx, connection.TransmitterIssuer, valueOrEmpty(connection.Audience)) + if metricsErr == nil && mode == "push_healthy" { + _, _ = m.repository.UpdateSecurityEventConnectionLifecycle(m.ctx, connectionID, "enabled", nil) + return + } + category := "bootstrap_verification_stale" + _, _ = m.repository.UpdateSecurityEventConnectionLifecycle(m.ctx, connectionID, "degraded", &category) + } +} + +func (m *ConnectionManager) finishRetirement(connection store.SecurityEventConnection) { + elapsed := time.Since(connection.UpdatedAt) + delay := m.config.RetirementDuration - elapsed + if delay > 0 { + select { + case <-m.ctx.Done(): + return + case <-time.After(delay): + } + } + delay = time.Second + for { + if err := m.repository.DeleteSecurityEventConnection(m.ctx, connection.ConnectionID); err == nil || errors.Is(err, store.ErrSecurityEventConnectionNotFound) { + break + } + select { + case <-m.ctx.Done(): + return + case <-time.After(delay): + } + if delay < time.Minute { + delay *= 2 + } + } + m.stopRuntime() + _ = m.secrets.Delete(m.ctx, connection.CredentialRef) + if connection.NextCredentialRef != nil { + _ = m.secrets.Delete(m.ctx, *connection.NextCredentialRef) + } +} + +func (m *ConnectionManager) validatePrerequisites(issuer string) error { + if !m.config.OIDCEnabled || m.config.ManagementClientID == "" || m.config.ManagementClientSecret == "" { + return errors.New("OIDC and RFC 7662 machine client must be configured") + } + if _, err := uuid.Parse(m.config.OIDCTenantID); err != nil { + return errors.New("OIDC tenant id is invalid") + } + if _, err := validatedIssuerURL(issuer, m.config.AppEnv); err != nil { + return err + } + endpoint, err := url.Parse(strings.TrimRight(m.config.PublicBaseURL, "/") + "/api/v1/security-events/ssf") + if err != nil || endpoint.Host == "" || endpoint.User != nil || endpoint.RawQuery != "" || endpoint.Fragment != "" { + return errors.New("AI_GATEWAY_PUBLIC_BASE_URL is invalid") + } + if endpoint.Scheme != "https" && !(isLocalEnvironment(m.config.AppEnv) && endpoint.Scheme == "http" && isLocalHostname(endpoint.Hostname())) { + return errors.New("AI_GATEWAY_PUBLIC_BASE_URL must use HTTPS") + } + return nil +} + +func (m *ConnectionManager) discover(ctx context.Context, issuer string) (transmitterConfiguration, *http.Client, error) { + issuerURL, err := validatedIssuerURL(issuer, m.config.AppEnv) + if err != nil { + return transmitterConfiguration{}, nil, err + } + client := m.config.HTTPClient + if client == nil { + client = safeHTTPClient(issuerURL, m.config.AppEnv) + } + discoveryURL := *issuerURL + discoveryURL.Path = "/.well-known/ssf-configuration" + strings.TrimRight(issuerURL.EscapedPath(), "/") + var configuration transmitterConfiguration + if err := requestJSON(ctx, client, http.MethodGet, discoveryURL.String(), "", nil, &configuration); err != nil { + return transmitterConfiguration{}, nil, err + } + if configuration.SpecVersion != "1_0" || strings.TrimRight(configuration.Issuer, "/") != strings.TrimRight(issuer, "/") || + !contains(configuration.DeliveryMethodsSupported, pushDeliveryMethod) { + return transmitterConfiguration{}, nil, errors.New("SSF discovery metadata is incompatible") + } + for _, endpoint := range []string{configuration.JWKSURI, configuration.ConfigurationEndpoint, configuration.StatusEndpoint, configuration.VerificationEndpoint} { + parsed, parseErr := url.Parse(endpoint) + if parseErr != nil || parsed.Scheme != issuerURL.Scheme || !strings.EqualFold(parsed.Host, issuerURL.Host) || parsed.User != nil || parsed.Fragment != "" { + return transmitterConfiguration{}, nil, errors.New("SSF discovery endpoint is not trusted") + } + } + return configuration, client, nil +} + +func (m *ConnectionManager) managementToken(ctx context.Context, client *http.Client, scopes string) (string, error) { + form := url.Values{"grant_type": {"client_credentials"}, "scope": {scopes}} + request, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(m.config.OIDCIssuer, "/")+"/token", strings.NewReader(form.Encode())) + if err != nil { + return "", err + } + request.SetBasicAuth(m.config.ManagementClientID, m.config.ManagementClientSecret) + request.Header.Set("Content-Type", "application/x-www-form-urlencoded") + response, err := client.Do(request) + if err != nil { + return "", err + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 64*1024)) + return "", fmt.Errorf("machine token request returned HTTP %d", response.StatusCode) + } + var payload struct { + AccessToken string `json:"access_token"` + TokenType string `json:"token_type"` + } + if err := decodeLimitedJSON(response.Body, &payload); err != nil || payload.AccessToken == "" || !strings.EqualFold(payload.TokenType, "Bearer") { + return "", errors.New("machine token response is invalid") + } + return payload.AccessToken, nil +} + +func (m *ConnectionManager) findStream(ctx context.Context, client *http.Client, configuration transmitterConfiguration, token, description, endpoint string) (streamConfiguration, error) { + var streams []streamConfiguration + if err := requestJSON(ctx, client, http.MethodGet, configuration.ConfigurationEndpoint, token, nil, &streams); err != nil { + return streamConfiguration{}, err + } + for _, stream := range streams { + if stream.Description != nil && *stream.Description == description && stream.Delivery.EndpointURL == endpoint { + return stream, nil + } + } + return streamConfiguration{}, store.ErrSecurityEventConnectionNotFound +} + +func (m *ConnectionManager) createStream(ctx context.Context, client *http.Client, configuration transmitterConfiguration, token, endpoint, secret, description string) (streamConfiguration, error) { + body := map[string]any{ + "delivery": map[string]any{"method": pushDeliveryMethod, "endpoint_url": endpoint, "authorization_header": "Bearer " + secret}, + "events_requested": []string{sessionRevokedEvent}, "description": description, + } + var stream streamConfiguration + err := requestJSON(ctx, client, http.MethodPost, configuration.ConfigurationEndpoint, token, body, &stream) + return stream, err +} + +func (m *ConnectionManager) patchStreamCredential(ctx context.Context, client *http.Client, configuration transmitterConfiguration, token, streamID, endpoint, secret string) error { + body := map[string]any{"stream_id": streamID, "delivery": map[string]any{ + "method": pushDeliveryMethod, "endpoint_url": endpoint, "authorization_header": "Bearer " + secret, + }} + return requestJSON(ctx, client, http.MethodPatch, configuration.ConfigurationEndpoint, token, body, &streamConfiguration{}) +} + +func (m *ConnectionManager) setStreamStatus(ctx context.Context, client *http.Client, configuration transmitterConfiguration, token, streamID, status, reason string) error { + body := map[string]any{"stream_id": streamID, "status": status, "reason": reason} + return requestJSON(ctx, client, http.MethodPost, configuration.StatusEndpoint, token, body, &map[string]any{}) +} + +func (m *ConnectionManager) deleteStream(ctx context.Context, client *http.Client, configuration transmitterConfiguration, token, streamID string) error { + endpoint, _ := url.Parse(configuration.ConfigurationEndpoint) + query := endpoint.Query() + query.Set("stream_id", streamID) + endpoint.RawQuery = query.Encode() + return requestJSON(ctx, client, http.MethodDelete, endpoint.String(), token, nil, nil) +} + +func (m *ConnectionManager) connectionError(ctx context.Context, connection store.SecurityEventConnection, category string, cause error) error { + _, _ = m.repository.UpdateSecurityEventConnectionLifecycle(ctx, connection.ConnectionID, "error", &category) + return cause +} + +func (m *ConnectionManager) view(connection store.SecurityEventConnection) ConnectionView { + return ConnectionView{ + ConnectionID: connection.ConnectionID, TransmitterIssuer: connection.TransmitterIssuer, + ReceiverEndpoint: connection.EndpointURL, Audience: connection.Audience, StreamID: connection.StreamID, + LifecycleStatus: connection.LifecycleStatus, HealthMode: connection.HealthMode, + ManagementClientID: m.config.ManagementClientID, LastVerificationAt: connection.LastVerificationAt, + LastErrorCategory: connection.LastErrorCategory, Version: connection.Version, + } +} + +func validateCreatedStream(stream streamConfiguration, issuer, endpoint string) error { + if _, err := uuid.Parse(stream.StreamID); err != nil || stream.Issuer != issuer || stream.Audience == "" || + stream.Delivery.Method != pushDeliveryMethod || stream.Delivery.EndpointURL != endpoint || + len(stream.EventsRequested) != 1 || stream.EventsRequested[0] != sessionRevokedEvent { + return errors.New("created SSF stream does not match the requested receiver") + } + return nil +} + +func randomPushBearer() ([]byte, error) { + raw := make([]byte, 32) + if _, err := rand.Read(raw); err != nil { + return nil, err + } + encoded := make([]byte, base64.RawURLEncoding.EncodedLen(len(raw))) + base64.RawURLEncoding.Encode(encoded, raw) + clear(raw) + return encoded, nil +} + +func requestJSON(ctx context.Context, client *http.Client, method, endpoint, token string, body, output any) error { + var reader io.Reader + if body != nil { + payload, err := json.Marshal(body) + if err != nil { + return err + } + reader = bytes.NewReader(payload) + } + request, err := http.NewRequestWithContext(ctx, method, endpoint, reader) + if err != nil { + return err + } + request.Header.Set("Accept", "application/json") + if token != "" { + request.Header.Set("Authorization", "Bearer "+token) + } + if body != nil { + request.Header.Set("Content-Type", "application/json") + } + response, err := client.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode < 200 || response.StatusCode >= 300 { + _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 64*1024)) + return remoteHTTPError{status: response.StatusCode} + } + if output == nil || response.StatusCode == http.StatusNoContent { + _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 64*1024)) + return nil + } + return decodeLimitedJSON(response.Body, output) +} + +func isRemoteHTTPStatus(err error, status int) bool { + var remoteError remoteHTTPError + return errors.As(err, &remoteError) && remoteError.status == status +} + +func decodeLimitedJSON(reader io.Reader, output any) error { + payload, err := io.ReadAll(io.LimitReader(reader, 64*1024+1)) + if err != nil || len(payload) > 64*1024 { + return errors.New("security event response is too large") + } + if len(payload) == 0 { + return nil + } + return json.Unmarshal(payload, output) +} + +func validatedIssuerURL(raw, appEnv string) (*url.URL, error) { + parsed, err := url.Parse(strings.TrimSpace(raw)) + if err != nil || parsed.Host == "" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" { + return nil, errors.New("transmitter issuer is invalid") + } + if parsed.Scheme != "https" && !(isLocalEnvironment(appEnv) && parsed.Scheme == "http" && isLocalHostname(parsed.Hostname())) { + return nil, errors.New("transmitter issuer must use HTTPS") + } + return parsed, nil +} + +func safeHTTPClient(issuer *url.URL, appEnv string) *http.Client { + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.Proxy = nil + transport.DisableCompression = true + transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12} + transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { + host, port, err := net.SplitHostPort(address) + if err != nil { + return nil, err + } + addresses, err := net.DefaultResolver.LookupIPAddr(ctx, host) + if err != nil || len(addresses) == 0 { + return nil, errors.New("transmitter DNS resolution failed") + } + allowLocal := isLocalEnvironment(appEnv) && isLocalHostname(issuer.Hostname()) + for _, address := range addresses { + if blockedAddress(address.IP) && !allowLocal { + return nil, errors.New("transmitter resolved to a blocked network") + } + } + dialer := &net.Dialer{Timeout: 5 * time.Second} + return dialer.DialContext(ctx, network, net.JoinHostPort(addresses[0].IP.String(), port)) + } + return &http.Client{Timeout: 10 * time.Second, Transport: transport, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }} +} + +func blockedAddress(ip net.IP) bool { + if ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || + ip.IsUnspecified() || ip.IsMulticast() { + return true + } + address, ok := netip.AddrFromSlice(ip) + if !ok { + return true + } + address = address.Unmap() + for _, prefix := range blockedNetworkPrefixes { + if prefix.Contains(address) { + return true + } + } + return false +} + +var blockedNetworkPrefixes = []netip.Prefix{ + netip.MustParsePrefix("0.0.0.0/8"), netip.MustParsePrefix("100.64.0.0/10"), + netip.MustParsePrefix("192.0.0.0/24"), netip.MustParsePrefix("192.0.2.0/24"), + netip.MustParsePrefix("198.18.0.0/15"), netip.MustParsePrefix("198.51.100.0/24"), + netip.MustParsePrefix("203.0.113.0/24"), netip.MustParsePrefix("240.0.0.0/4"), + netip.MustParsePrefix("100::/64"), netip.MustParsePrefix("2001:db8::/32"), +} + +func isLocalHostname(value string) bool { + return value == "localhost" || value == "127.0.0.1" || value == "::1" +} + +func isLocalEnvironment(value string) bool { + switch strings.ToLower(strings.TrimSpace(value)) { + case "development", "dev", "local", "test": + return true + default: + return false + } +} + +func contains(values []string, expected string) bool { + for _, value := range values { + if value == expected { + return true + } + } + return false +} + +func valueOrEmpty(value *string) string { + if value == nil { + return "" + } + return *value +} diff --git a/apps/api/internal/securityevents/connection_manager_test.go b/apps/api/internal/securityevents/connection_manager_test.go new file mode 100644 index 0000000..4d40535 --- /dev/null +++ b/apps/api/internal/securityevents/connection_manager_test.go @@ -0,0 +1,293 @@ +package securityevents + +import ( + "context" + "encoding/json" + "errors" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/auth" + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" + "github.com/google/uuid" +) + +type memoryConnectionRepository struct { + mutex sync.Mutex + connection *store.SecurityEventConnection + mode string + verifiedAt *time.Time + evaluation store.SecurityEventEvaluation +} + +func (*memoryConnectionRepository) ApplySessionRevoked(context.Context, store.ApplySessionRevokedInput) (store.ApplySecurityEventResult, error) { + return store.ApplySecurityEventResult{}, nil +} +func (*memoryConnectionRepository) ApplySecurityEventStreamUpdated(context.Context, string, string, string, string, string, string, time.Time) (bool, error) { + return true, nil +} +func (*memoryConnectionRepository) ConfirmSecurityEventVerification(context.Context, string, string, string, string, []byte, time.Time) (bool, error) { + return true, nil +} +func (*memoryConnectionRepository) RecordSecurityEventReceipt(context.Context, string, string, string, string, string) (bool, error) { + return true, nil +} +func (r *memoryConnectionRepository) EnsureSecurityEventStreamState(context.Context, string, string, string) error { + r.mode = "bootstrap" + return nil +} +func (*memoryConnectionRepository) BeginSecurityEventVerification(context.Context, string, string, []byte, time.Time) error { + return nil +} +func (*memoryConnectionRepository) RecordSecurityEventHeartbeatFailure(context.Context, string, string, string) error { + return nil +} +func (*memoryConnectionRepository) AdvanceSecurityEventStreamState(context.Context, string, string, time.Time, time.Duration) error { + return nil +} +func (r *memoryConnectionRepository) EvaluateOIDCSecurityEvent(context.Context, string, string, string, string, string, time.Time, time.Time, time.Duration) (store.SecurityEventEvaluation, error) { + if r.evaluation.Mode == "" { + return store.SecurityEventEvaluation{Mode: "bootstrap", RequireIntrospection: true}, nil + } + return r.evaluation, nil +} +func (r *memoryConnectionRepository) CreateSecurityEventConnection(_ context.Context, input store.CreateSecurityEventConnectionInput) (store.SecurityEventConnection, error) { + r.mutex.Lock() + defer r.mutex.Unlock() + if r.connection != nil { + return store.SecurityEventConnection{}, store.ErrSecurityEventConnectionConflict + } + now := time.Now().UTC() + value := store.SecurityEventConnection{ConnectionID: input.ConnectionID, TransmitterIssuer: input.TransmitterIssuer, + EndpointURL: input.EndpointURL, CredentialRef: input.CredentialRef, LifecycleStatus: "connecting", Version: 1, + IdempotencyKey: input.IdempotencyKey, CreatedAt: now, UpdatedAt: now, HealthMode: "disabled"} + r.connection = &value + return value, nil +} +func (r *memoryConnectionRepository) SecurityEventConnection(context.Context) (store.SecurityEventConnection, error) { + r.mutex.Lock() + defer r.mutex.Unlock() + if r.connection == nil { + return store.SecurityEventConnection{}, store.ErrSecurityEventConnectionNotFound + } + value := *r.connection + value.HealthMode, value.LastVerificationAt = r.mode, r.verifiedAt + return value, nil +} +func (r *memoryConnectionRepository) BindSecurityEventConnection(_ context.Context, id, audience, streamID, lifecycle string, expected int64) (store.SecurityEventConnection, error) { + r.mutex.Lock() + defer r.mutex.Unlock() + if r.connection == nil || r.connection.ConnectionID != id || r.connection.Version != expected { + return store.SecurityEventConnection{}, store.ErrSecurityEventConnectionConflict + } + r.connection.Audience, r.connection.StreamID = &audience, &streamID + r.connection.LifecycleStatus, r.connection.Version = lifecycle, r.connection.Version+1 + return *r.connection, nil +} +func (r *memoryConnectionRepository) UpdateSecurityEventConnectionLifecycle(_ context.Context, id, lifecycle string, category *string) (store.SecurityEventConnection, error) { + r.mutex.Lock() + defer r.mutex.Unlock() + if r.connection == nil || r.connection.ConnectionID != id { + return store.SecurityEventConnection{}, store.ErrSecurityEventConnectionNotFound + } + r.connection.LifecycleStatus, r.connection.LastErrorCategory = lifecycle, category + r.connection.Version++ + r.connection.UpdatedAt = time.Now().UTC() + return *r.connection, nil +} +func (r *memoryConnectionRepository) SetSecurityEventNextCredential(context.Context, string, string) (store.SecurityEventConnection, error) { + return store.SecurityEventConnection{}, errors.New("not used") +} + +func TestConnectionManagerBootstrapOnlyBecomesEnabledWhenPushIsHealthy(t *testing.T) { + for _, test := range []struct { + mode, wantLifecycle string + }{ + {mode: "push_healthy", wantLifecycle: "enabled"}, + {mode: "introspection_fallback", wantLifecycle: "degraded"}, + } { + t.Run(test.mode, func(t *testing.T) { + audience := "urn:easyai:ssf:receiver:" + uuid.NewString() + repository := &memoryConnectionRepository{mode: test.mode, connection: &store.SecurityEventConnection{ + ConnectionID: uuid.NewString(), TransmitterIssuer: "https://auth.example/ssf", Audience: &audience, + LifecycleStatus: "bootstrap", Version: 1, UpdatedAt: time.Now().UTC(), + }} + manager := &ConnectionManager{ctx: context.Background(), repository: repository, + config: ConnectionManagerConfig{BootstrapDuration: time.Millisecond}} + manager.finishBootstrap(repository.connection.ConnectionID) + connection, err := repository.SecurityEventConnection(context.Background()) + if err != nil || connection.LifecycleStatus != test.wantLifecycle { + t.Fatalf("connection=%#v err=%v", connection, err) + } + }) + } +} + +func TestRemoteNotFoundIsRecognizedForIdempotentDisconnect(t *testing.T) { + if !isRemoteHTTPStatus(remoteHTTPError{status: http.StatusNotFound}, http.StatusNotFound) { + t.Fatal("remote 404 was not recognized as an already-deleted Stream") + } +} + +func TestTransmitterSSRFProtectionBlocksReservedNetworks(t *testing.T) { + for _, address := range []string{ + "127.0.0.1", "10.0.0.1", "169.254.169.254", "100.64.0.1", "192.0.2.1", "198.51.100.1", "203.0.113.1", "2001:db8::1", + } { + if !blockedAddress(net.ParseIP(address)) { + t.Fatalf("reserved address %s was allowed", address) + } + } + if blockedAddress(net.ParseIP("8.8.8.8")) { + t.Fatal("public address was blocked") + } +} + +func TestRetiringConnectionStillAppliesRevocationWatermark(t *testing.T) { + audience := "urn:easyai:ssf:receiver:" + uuid.NewString() + repository := &memoryConnectionRepository{ + evaluation: store.SecurityEventEvaluation{Mode: "introspection_fallback", Revoked: true, RequireIntrospection: true}, + connection: &store.SecurityEventConnection{ + ConnectionID: uuid.NewString(), TransmitterIssuer: "https://auth.example/ssf", Audience: &audience, + LifecycleStatus: "retiring", Version: 1, + }, + } + manager := &ConnectionManager{ctx: context.Background(), repository: repository, config: ConnectionManagerConfig{StaleAfter: 3 * time.Minute}} + evaluation, err := manager.Evaluate(context.Background(), auth.OIDCSecurityEventIdentity{IssuedAt: time.Now().Add(-time.Minute)}) + if err != nil || !evaluation.Enabled || !evaluation.Revoked || !evaluation.RequireIntrospection { + t.Fatalf("retiring evaluation=%#v err=%v", evaluation, err) + } +} +func (r *memoryConnectionRepository) PromoteSecurityEventCredential(context.Context, string, string) (store.SecurityEventConnection, error) { + return store.SecurityEventConnection{}, errors.New("not used") +} +func (r *memoryConnectionRepository) DeleteSecurityEventConnection(context.Context, string) error { + r.connection = nil + return nil +} +func (r *memoryConnectionRepository) SecurityEventMetrics(context.Context, string, string) (string, *time.Time, error) { + return r.mode, r.verifiedAt, nil +} + +type memorySecretStore struct { + values map[string][]byte +} + +func (s *memorySecretStore) Put(_ context.Context, reference string, value []byte) error { + if s.values == nil { + s.values = map[string][]byte{} + } + s.values[reference] = append([]byte(nil), value...) + return nil +} +func (s *memorySecretStore) Get(_ context.Context, reference string) ([]byte, error) { + value, ok := s.values[reference] + if !ok { + return nil, ErrSecretNotFound + } + return append([]byte(nil), value...), nil +} +func (s *memorySecretStore) Delete(_ context.Context, reference string) error { + delete(s.values, reference) + return nil +} + +func TestConnectionManagerCreatesPausedStreamWithBackendGeneratedBearer(t *testing.T) { + repository := &memoryConnectionRepository{} + secrets := &memorySecretStore{} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var transmitter *httptest.Server + verificationRequested := make(chan struct{}, 1) + var managementScopeRequested atomic.Bool + transmitter = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/.well-known/ssf-configuration/ssf": + _ = json.NewEncoder(w).Encode(map[string]any{ + "spec_version": "1_0", "issuer": transmitter.URL + "/ssf", "jwks_uri": transmitter.URL + "/ssf/jwks.json", + "configuration_endpoint": transmitter.URL + "/ssf/v1/stream", "status_endpoint": transmitter.URL + "/ssf/v1/status", + "verification_endpoint": transmitter.URL + "/ssf/v1/verify", "delivery_methods_supported": []string{pushDeliveryMethod}, + }) + case "/oidc/token": + clientID, secret, ok := r.BasicAuth() + if !ok || clientID != "gateway-machine" || secret != "machine-secret" { + t.Fatal("RFC 7662 machine client was not reused") + } + _ = r.ParseForm() + if strings.Contains(r.Form.Get("scope"), "ssf.stream.manage") { + managementScopeRequested.Store(true) + } + _ = json.NewEncoder(w).Encode(map[string]any{"access_token": "machine-token", "token_type": "Bearer"}) + case "/ssf/v1/stream": + if r.Method == http.MethodGet { + _, _ = w.Write([]byte(`[]`)) + return + } + var request struct { + Delivery struct { + AuthorizationHeader string `json:"authorization_header"` + EndpointURL string `json:"endpoint_url"` + } `json:"delivery"` + Description string `json:"description"` + } + _ = json.NewDecoder(r.Body).Decode(&request) + if len(strings.TrimPrefix(request.Delivery.AuthorizationHeader, "Bearer ")) < 43 { + t.Fatal("generated Push Bearer has less than 256 bits") + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "stream_id": uuid.NewString(), "iss": transmitter.URL + "/ssf", + "aud": "urn:easyai:ssf:receiver:" + uuid.NewString(), "events_requested": []string{sessionRevokedEvent}, + "delivery": map[string]any{"method": pushDeliveryMethod, "endpoint_url": request.Delivery.EndpointURL}, + "description": request.Description, + }) + case "/ssf/v1/verify": + w.WriteHeader(http.StatusNoContent) + select { + case verificationRequested <- struct{}{}: + default: + } + default: + http.NotFound(w, r) + } + })) + defer transmitter.Close() + manager, err := NewConnectionManager(ctx, repository, secrets, ConnectionManagerConfig{ + AppEnv: "test", OIDCEnabled: true, OIDCIssuer: transmitter.URL + "/oidc", OIDCTenantID: uuid.NewString(), + ManagementClientID: "gateway-machine", ManagementClientSecret: "machine-secret", + PublicBaseURL: transmitter.URL, HeartbeatInterval: time.Hour, StaleAfter: 2 * time.Hour, HTTPClient: transmitter.Client(), + }, &Metrics{}) + if err != nil { + t.Fatal(err) + } + evaluation, err := manager.Evaluate(ctx, auth.OIDCSecurityEventIdentity{}) + if err != nil || evaluation.Enabled { + t.Fatalf("unconfigured manager evaluation=%#v err=%v", evaluation, err) + } + view, err := manager.Connect(ctx, transmitter.URL+"/ssf", "connect-once") + if err != nil { + t.Fatal(err) + } + if view.LifecycleStatus != "verifying" || view.StreamID == nil || view.Audience == nil { + t.Fatalf("connection view=%#v", view) + } + encoded, _ := json.Marshal(view) + if strings.Contains(string(encoded), "credential") || strings.Contains(string(encoded), "machine-secret") || strings.Contains(string(encoded), "ssf-push-") { + t.Fatalf("secret metadata escaped in response: %s", encoded) + } + if len(secrets.values) != 1 { + t.Fatalf("secret count=%d", len(secrets.values)) + } + if !managementScopeRequested.Load() { + t.Fatal("management scopes were not requested automatically") + } + select { + case <-verificationRequested: + case <-time.After(2 * time.Second): + t.Fatal("verification was not started without a restart") + } +} diff --git a/apps/api/internal/securityevents/kubernetes_secret_store.go b/apps/api/internal/securityevents/kubernetes_secret_store.go new file mode 100644 index 0000000..34bf582 --- /dev/null +++ b/apps/api/internal/securityevents/kubernetes_secret_store.go @@ -0,0 +1,186 @@ +package securityevents + +import ( + "bytes" + "context" + "crypto/tls" + "crypto/x509" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "regexp" + "strings" + "time" +) + +const ( + defaultKubernetesAPIServer = "https://kubernetes.default.svc" + defaultServiceAccountToken = "/var/run/secrets/kubernetes.io/serviceaccount/token" + defaultServiceAccountCA = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt" +) + +var kubernetesDNSNamePattern = regexp.MustCompile(`^[a-z0-9]([-a-z0-9.]*[a-z0-9])?$`) + +type KubernetesSecretStoreConfig struct { + Namespace string + SecretName string + APIServer string + TokenFile string + CAFile string + HTTPClient *http.Client +} + +// KubernetesSecretStore keeps all generated Push Bearers as keys in one +// pre-provisioned Secret. It intentionally has no permission to create or +// delete Kubernetes Secret resources. +type KubernetesSecretStore struct { + endpoint string + tokenFile string + client *http.Client +} + +func NewKubernetesSecretStore(config KubernetesSecretStoreConfig) (*KubernetesSecretStore, error) { + if !validKubernetesDNSName(config.Namespace) || !validKubernetesDNSName(config.SecretName) { + return nil, errors.New("Kubernetes security event Secret namespace or name is invalid") + } + if config.APIServer == "" { + config.APIServer = defaultKubernetesAPIServer + } + if config.TokenFile == "" { + config.TokenFile = defaultServiceAccountToken + } + parsed, err := url.Parse(strings.TrimRight(config.APIServer, "/")) + if err != nil || parsed.Host == "" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" || + (parsed.Scheme != "https" && config.HTTPClient == nil) { + return nil, errors.New("Kubernetes API server URL is invalid") + } + client := config.HTTPClient + if client == nil { + if config.CAFile == "" { + config.CAFile = defaultServiceAccountCA + } + certificate, readErr := os.ReadFile(config.CAFile) + if readErr != nil { + return nil, fmt.Errorf("read Kubernetes service account CA: %w", readErr) + } + roots := x509.NewCertPool() + if !roots.AppendCertsFromPEM(certificate) { + return nil, errors.New("Kubernetes service account CA is invalid") + } + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.Proxy = nil + transport.DisableCompression = true + transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12, RootCAs: roots} + client = &http.Client{Timeout: 10 * time.Second, Transport: transport, + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }} + } + endpoint := fmt.Sprintf("%s/api/v1/namespaces/%s/secrets/%s", strings.TrimRight(config.APIServer, "/"), config.Namespace, config.SecretName) + return &KubernetesSecretStore{endpoint: endpoint, tokenFile: config.TokenFile, client: client}, nil +} + +func (s *KubernetesSecretStore) Put(ctx context.Context, reference string, value []byte) error { + if !secretReferencePattern.MatchString(reference) { + return errors.New("security event secret reference is invalid") + } + if len(value) < 32 || len(value) > 4096 { + return errors.New("security event secret length is invalid") + } + return s.patch(ctx, map[string]any{reference: base64.StdEncoding.EncodeToString(value)}) +} + +func (s *KubernetesSecretStore) Get(ctx context.Context, reference string) ([]byte, error) { + if !secretReferencePattern.MatchString(reference) { + return nil, errors.New("security event secret reference is invalid") + } + request, err := s.request(ctx, http.MethodGet, nil) + if err != nil { + return nil, err + } + response, err := s.client.Do(request) + if err != nil { + return nil, fmt.Errorf("read Kubernetes security event Secret: %w", err) + } + defer response.Body.Close() + if response.StatusCode == http.StatusNotFound { + _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 64*1024)) + return nil, ErrSecretNotFound + } + if response.StatusCode != http.StatusOK { + _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 64*1024)) + return nil, fmt.Errorf("Kubernetes security event Secret returned HTTP %d", response.StatusCode) + } + var payload struct { + Data map[string]string `json:"data"` + } + if err := decodeLimitedJSON(response.Body, &payload); err != nil { + return nil, err + } + encoded, ok := payload.Data[reference] + if !ok { + return nil, ErrSecretNotFound + } + value, err := base64.StdEncoding.DecodeString(encoded) + if err != nil || len(value) < 32 || len(value) > 4096 { + clear(value) + return nil, errors.New("Kubernetes security event Secret value is invalid") + } + return value, nil +} + +func (s *KubernetesSecretStore) Delete(ctx context.Context, reference string) error { + if !secretReferencePattern.MatchString(reference) { + return errors.New("security event secret reference is invalid") + } + return s.patch(ctx, map[string]any{reference: nil}) +} + +func (s *KubernetesSecretStore) patch(ctx context.Context, data map[string]any) error { + payload, err := json.Marshal(map[string]any{"data": data}) + if err != nil { + return err + } + request, err := s.request(ctx, http.MethodPatch, bytes.NewReader(payload)) + if err != nil { + return err + } + request.Header.Set("Content-Type", "application/merge-patch+json") + response, err := s.client.Do(request) + if err != nil { + return fmt.Errorf("update Kubernetes security event Secret: %w", err) + } + defer response.Body.Close() + _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 64*1024)) + if response.StatusCode == http.StatusNotFound { + return errors.New("Kubernetes security event Secret is not provisioned") + } + if response.StatusCode < 200 || response.StatusCode >= 300 { + return fmt.Errorf("Kubernetes security event Secret returned HTTP %d", response.StatusCode) + } + return nil +} + +func (s *KubernetesSecretStore) request(ctx context.Context, method string, body io.Reader) (*http.Request, error) { + token, err := os.ReadFile(s.tokenFile) + if err != nil || strings.TrimSpace(string(token)) == "" { + clear(token) + return nil, errors.New("Kubernetes service account token is unavailable") + } + request, err := http.NewRequestWithContext(ctx, method, s.endpoint, body) + if err != nil { + clear(token) + return nil, err + } + request.Header.Set("Authorization", "Bearer "+strings.TrimSpace(string(token))) + request.Header.Set("Accept", "application/json") + clear(token) + return request, nil +} + +func validKubernetesDNSName(value string) bool { + return len(value) > 0 && len(value) <= 253 && kubernetesDNSNamePattern.MatchString(value) +} diff --git a/apps/api/internal/securityevents/kubernetes_secret_store_test.go b/apps/api/internal/securityevents/kubernetes_secret_store_test.go new file mode 100644 index 0000000..fadbc31 --- /dev/null +++ b/apps/api/internal/securityevents/kubernetes_secret_store_test.go @@ -0,0 +1,95 @@ +package securityevents + +import ( + "context" + "encoding/base64" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync" + "testing" +) + +func TestKubernetesSecretStoreUsesOnePreprovisionedSecret(t *testing.T) { + var mutex sync.Mutex + data := map[string]string{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/v1/namespaces/easyai/secrets/easyai-gateway-security-events" || r.Header.Get("Authorization") != "Bearer service-account-token" { + t.Fatalf("unexpected Kubernetes request: %s authorization=%t", r.URL.Path, r.Header.Get("Authorization") != "") + } + mutex.Lock() + defer mutex.Unlock() + switch r.Method { + case http.MethodGet: + _ = json.NewEncoder(w).Encode(map[string]any{"data": data}) + case http.MethodPatch: + var patch struct { + Data map[string]*string `json:"data"` + } + if err := json.NewDecoder(r.Body).Decode(&patch); err != nil { + t.Fatal(err) + } + for key, value := range patch.Data { + if value == nil { + delete(data, key) + } else { + data[key] = *value + } + } + _ = json.NewEncoder(w).Encode(map[string]any{"data": data}) + default: + http.Error(w, "unsupported", http.StatusMethodNotAllowed) + } + })) + defer server.Close() + tokenFile := filepath.Join(t.TempDir(), "token") + if err := os.WriteFile(tokenFile, []byte("service-account-token\n"), 0o600); err != nil { + t.Fatal(err) + } + secrets, err := NewKubernetesSecretStore(KubernetesSecretStoreConfig{ + Namespace: "easyai", SecretName: "easyai-gateway-security-events", APIServer: server.URL, + TokenFile: tokenFile, HTTPClient: server.Client(), + }) + if err != nil { + t.Fatal(err) + } + value := []byte("0123456789abcdef0123456789abcdef") + if err := secrets.Put(context.Background(), "ssf-push-connection", value); err != nil { + t.Fatal(err) + } + if data["ssf-push-connection"] != base64.StdEncoding.EncodeToString(value) { + t.Fatal("Push Bearer was not stored in Kubernetes Secret data") + } + loaded, err := secrets.Get(context.Background(), "ssf-push-connection") + if err != nil || string(loaded) != string(value) { + t.Fatalf("loaded secret mismatch: err=%v", err) + } + clear(loaded) + if err := secrets.Delete(context.Background(), "ssf-push-connection"); err != nil { + t.Fatal(err) + } + if _, ok := data["ssf-push-connection"]; ok { + t.Fatal("Push Bearer key was not removed") + } +} + +func TestKubernetesSecretStoreRejectsUnprovisionedSecret(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { http.NotFound(w, nil) })) + defer server.Close() + tokenFile := filepath.Join(t.TempDir(), "token") + if err := os.WriteFile(tokenFile, []byte("token"), 0o600); err != nil { + t.Fatal(err) + } + secrets, err := NewKubernetesSecretStore(KubernetesSecretStoreConfig{ + Namespace: "easyai", SecretName: "easyai-gateway-security-events", APIServer: server.URL, + TokenFile: tokenFile, HTTPClient: server.Client(), + }) + if err != nil { + t.Fatal(err) + } + if err := secrets.Put(context.Background(), "ssf-push-connection", []byte("0123456789abcdef0123456789abcdef")); err == nil { + t.Fatal("missing pre-provisioned Kubernetes Secret was accepted") + } +} diff --git a/apps/api/internal/securityevents/metrics.go b/apps/api/internal/securityevents/metrics.go index 39a5fc3..b7a8e88 100644 --- a/apps/api/internal/securityevents/metrics.go +++ b/apps/api/internal/securityevents/metrics.go @@ -2,16 +2,24 @@ package securityevents import ( "context" + "errors" "fmt" "net/http" "sync/atomic" "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" ) type MetricsSnapshotProvider interface { SecurityEventMetrics(context.Context, string, string) (string, *time.Time, error) } +type DynamicMetricsSnapshotProvider interface { + MetricsSnapshotProvider + SecurityEventConnection(context.Context) (store.SecurityEventConnection, error) +} + type Metrics struct { accepted atomic.Uint64 rejected atomic.Uint64 @@ -151,6 +159,25 @@ func (m *Metrics) Handler(provider MetricsSnapshotProvider, issuer, audience str }) } +func (m *Metrics) DynamicHandler(provider DynamicMetricsSnapshotProvider) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + connection, err := provider.SecurityEventConnection(r.Context()) + if errors.Is(err, store.ErrSecurityEventConnectionNotFound) { + m.Handler(provider, "", "", false).ServeHTTP(w, r) + return + } + if err != nil { + http.Error(w, "metrics unavailable", http.StatusServiceUnavailable) + return + } + if connection.Audience == nil { + m.Handler(provider, "", "", false).ServeHTTP(w, r) + return + } + m.Handler(provider, connection.TransmitterIssuer, *connection.Audience, true).ServeHTTP(w, r) + }) +} + type outcomeValue struct { name string value uint64 diff --git a/apps/api/internal/securityevents/secret_store.go b/apps/api/internal/securityevents/secret_store.go new file mode 100644 index 0000000..2b39bd4 --- /dev/null +++ b/apps/api/internal/securityevents/secret_store.go @@ -0,0 +1,112 @@ +package securityevents + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "regexp" +) + +var secretReferencePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,127}$`) + +var ErrSecretNotFound = errors.New("security event secret not found") + +type SecretStore interface { + Put(context.Context, string, []byte) error + Get(context.Context, string) ([]byte, error) + Delete(context.Context, string) error +} + +type FileSecretStore struct{ directory string } + +func NewFileSecretStore(directory string) (*FileSecretStore, error) { + if directory == "" { + return nil, errors.New("security event secret directory is required") + } + if err := os.MkdirAll(directory, 0o700); err != nil { + return nil, fmt.Errorf("create security event secret directory: %w", err) + } + if err := os.Chmod(directory, 0o700); err != nil { + return nil, fmt.Errorf("protect security event secret directory: %w", err) + } + return &FileSecretStore{directory: directory}, nil +} + +func (s *FileSecretStore) Put(_ context.Context, reference string, value []byte) error { + path, err := s.path(reference) + if err != nil { + return err + } + if len(value) < 32 || len(value) > 4096 { + return errors.New("security event secret length is invalid") + } + temporary, err := os.CreateTemp(s.directory, ".ssf-secret-*") + if err != nil { + return fmt.Errorf("create temporary security event secret: %w", err) + } + temporaryName := temporary.Name() + defer func() { _ = os.Remove(temporaryName) }() + if err := temporary.Chmod(0o600); err != nil { + _ = temporary.Close() + return err + } + if _, err := temporary.Write(value); err != nil { + _ = temporary.Close() + return fmt.Errorf("write security event secret: %w", err) + } + if err := temporary.Sync(); err != nil { + _ = temporary.Close() + return err + } + if err := temporary.Close(); err != nil { + return err + } + if err := os.Rename(temporaryName, path); err != nil { + return fmt.Errorf("publish security event secret: %w", err) + } + return os.Chmod(path, 0o600) +} + +func (s *FileSecretStore) Get(_ context.Context, reference string) ([]byte, error) { + path, err := s.path(reference) + if err != nil { + return nil, err + } + info, err := os.Stat(path) + if errors.Is(err, os.ErrNotExist) { + return nil, ErrSecretNotFound + } + if err != nil || !info.Mode().IsRegular() || info.Mode().Perm()&0o077 != 0 { + return nil, errors.New("security event secret file is not protected") + } + value, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read security event secret: %w", err) + } + if len(value) < 32 || len(value) > 4096 { + clear(value) + return nil, errors.New("security event secret length is invalid") + } + return value, nil +} + +func (s *FileSecretStore) Delete(_ context.Context, reference string) error { + path, err := s.path(reference) + if err != nil { + return err + } + err = os.Remove(path) + if errors.Is(err, os.ErrNotExist) { + return nil + } + return err +} + +func (s *FileSecretStore) path(reference string) (string, error) { + if !secretReferencePattern.MatchString(reference) { + return "", errors.New("security event secret reference is invalid") + } + return filepath.Join(s.directory, reference), nil +} diff --git a/apps/api/internal/securityevents/secret_store_test.go b/apps/api/internal/securityevents/secret_store_test.go new file mode 100644 index 0000000..d94934c --- /dev/null +++ b/apps/api/internal/securityevents/secret_store_test.go @@ -0,0 +1,40 @@ +package securityevents + +import ( + "bytes" + "context" + "errors" + "os" + "path/filepath" + "testing" +) + +func TestFileSecretStoreCreatesProtectedFilesAndNeverAcceptsTraversal(t *testing.T) { + directory := filepath.Join(t.TempDir(), "ssf") + secrets, err := NewFileSecretStore(directory) + if err != nil { + t.Fatal(err) + } + value := bytes.Repeat([]byte{0x41}, 32) + if err := secrets.Put(context.Background(), "ssf-push-connection", value); err != nil { + t.Fatal(err) + } + info, err := os.Stat(filepath.Join(directory, "ssf-push-connection")) + if err != nil || info.Mode().Perm() != 0o600 { + t.Fatalf("secret mode=%v err=%v", info.Mode().Perm(), err) + } + loaded, err := secrets.Get(context.Background(), "ssf-push-connection") + if err != nil || !bytes.Equal(loaded, value) { + t.Fatalf("secret mismatch err=%v", err) + } + clear(loaded) + if err := secrets.Put(context.Background(), "../escape", value); err == nil { + t.Fatal("path traversal reference was accepted") + } + if err := secrets.Delete(context.Background(), "ssf-push-connection"); err != nil { + t.Fatal(err) + } + if _, err := secrets.Get(context.Background(), "ssf-push-connection"); !errors.Is(err, ErrSecretNotFound) { + t.Fatalf("deleted secret error=%v", err) + } +} diff --git a/apps/api/internal/securityevents/service.go b/apps/api/internal/securityevents/service.go index 6814161..053d60f 100644 --- a/apps/api/internal/securityevents/service.go +++ b/apps/api/internal/securityevents/service.go @@ -75,13 +75,24 @@ func NewService(repository StateRepository, config ServiceConfig) (*Service, err } func (s *Service) Start(ctx context.Context) error { - if err := s.repository.EnsureSecurityEventStreamState(ctx, s.config.Issuer, s.config.Audience, s.config.StreamID); err != nil { + if err := s.Initialize(ctx); err != nil { return err } - go s.run(ctx) + go s.Run(ctx) return nil } +func (s *Service) Initialize(ctx context.Context) error { + return s.repository.EnsureSecurityEventStreamState(ctx, s.config.Issuer, s.config.Audience, s.config.StreamID) +} + +func (s *Service) Run(ctx context.Context) { s.run(ctx) } + +// RequestVerification asks for an immediate verification without changing the +// remote stream status. In particular, this never re-enables a stream paused +// by an authentication-center administrator. +func (s *Service) RequestVerification(ctx context.Context) { s.sendVerification(ctx) } + func (s *Service) Evaluate(ctx context.Context, identity auth.OIDCSecurityEventIdentity) (auth.OIDCSecurityEventEvaluation, error) { result, err := s.repository.EvaluateOIDCSecurityEvent( ctx, s.config.Issuer, s.config.Audience, identity.Issuer, identity.TenantID, identity.Subject, diff --git a/apps/api/internal/store/security_event_connections.go b/apps/api/internal/store/security_event_connections.go new file mode 100644 index 0000000..fb0e198 --- /dev/null +++ b/apps/api/internal/store/security_event_connections.go @@ -0,0 +1,175 @@ +package store + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +var ( + ErrSecurityEventConnectionNotFound = errors.New("security event connection not found") + ErrSecurityEventConnectionConflict = errors.New("security event connection conflicts with current state") +) + +type SecurityEventConnection struct { + ConnectionID string `json:"connectionId"` + TransmitterIssuer string `json:"transmitterIssuer"` + EndpointURL string `json:"endpointUrl"` + Audience *string `json:"audience,omitempty"` + StreamID *string `json:"streamId,omitempty"` + CredentialRef string `json:"-"` + NextCredentialRef *string `json:"-"` + LifecycleStatus string `json:"lifecycleStatus"` + Version int64 `json:"version"` + LastErrorCategory *string `json:"lastErrorCategory,omitempty"` + IdempotencyKey string `json:"-"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` + LastVerificationAt *time.Time `json:"lastVerificationAt,omitempty"` + HealthMode string `json:"healthMode"` +} + +type CreateSecurityEventConnectionInput struct { + ConnectionID, TransmitterIssuer, EndpointURL, CredentialRef, IdempotencyKey string +} + +type SecurityEventConnectionIdempotency struct { + RequestHash string + Response json.RawMessage +} + +func (s *Store) SecurityEventConnectionIdempotency(ctx context.Context, operation, key string) (SecurityEventConnectionIdempotency, error) { + var value SecurityEventConnectionIdempotency + err := s.pool.QueryRow(ctx, `SELECT request_hash,response FROM gateway_security_event_connection_idempotency +WHERE operation=$1 AND idempotency_key=$2`, operation, key).Scan(&value.RequestHash, &value.Response) + if errors.Is(err, pgx.ErrNoRows) { + return SecurityEventConnectionIdempotency{}, ErrSecurityEventConnectionNotFound + } + return value, err +} + +func (s *Store) RecordSecurityEventConnectionIdempotency(ctx context.Context, operation, key, requestHash string, response []byte) error { + if !json.Valid(response) { + return errors.New("security event idempotency response is invalid") + } + _, err := s.pool.Exec(ctx, `INSERT INTO gateway_security_event_connection_idempotency(operation,idempotency_key,request_hash,response) +VALUES($1,$2,$3,$4::jsonb) ON CONFLICT(operation,idempotency_key) DO NOTHING`, operation, key, requestHash, response) + return err +} + +func (s *Store) CreateSecurityEventConnection(ctx context.Context, input CreateSecurityEventConnectionInput) (SecurityEventConnection, error) { + var value SecurityEventConnection + err := s.pool.QueryRow(ctx, ` +INSERT INTO gateway_security_event_connections( + connection_id,transmitter_issuer,endpoint_url,credential_ref,lifecycle_status,idempotency_key +) VALUES($1::uuid,$2,$3,$4,'connecting',$5) +RETURNING connection_id::text,transmitter_issuer,endpoint_url,audience,stream_id::text,credential_ref, + next_credential_ref,lifecycle_status,version,last_error_category,idempotency_key,created_at,updated_at`, + input.ConnectionID, input.TransmitterIssuer, input.EndpointURL, input.CredentialRef, input.IdempotencyKey, + ).Scan(&value.ConnectionID, &value.TransmitterIssuer, &value.EndpointURL, &value.Audience, &value.StreamID, + &value.CredentialRef, &value.NextCredentialRef, &value.LifecycleStatus, &value.Version, + &value.LastErrorCategory, &value.IdempotencyKey, &value.CreatedAt, &value.UpdatedAt) + if err != nil { + if isUniqueViolation(err) { + existing, getErr := s.SecurityEventConnection(ctx) + if getErr == nil && existing.IdempotencyKey == input.IdempotencyKey && existing.TransmitterIssuer == input.TransmitterIssuer { + return existing, nil + } + return SecurityEventConnection{}, ErrSecurityEventConnectionConflict + } + return SecurityEventConnection{}, err + } + return value, nil +} + +func (s *Store) SecurityEventConnection(ctx context.Context) (SecurityEventConnection, error) { + var value SecurityEventConnection + err := s.pool.QueryRow(ctx, ` +SELECT connection.connection_id::text,connection.transmitter_issuer,connection.endpoint_url,connection.audience, + connection.stream_id::text,connection.credential_ref,connection.next_credential_ref,connection.lifecycle_status, + connection.version,connection.last_error_category,connection.idempotency_key,connection.created_at,connection.updated_at, + state.last_verification_at,COALESCE(state.mode,'disabled') +FROM gateway_security_event_connections connection +LEFT JOIN gateway_security_event_stream_state state + ON state.issuer=connection.transmitter_issuer AND state.audience=connection.audience +WHERE connection.singleton=true`).Scan( + &value.ConnectionID, &value.TransmitterIssuer, &value.EndpointURL, &value.Audience, &value.StreamID, + &value.CredentialRef, &value.NextCredentialRef, &value.LifecycleStatus, &value.Version, + &value.LastErrorCategory, &value.IdempotencyKey, &value.CreatedAt, &value.UpdatedAt, + &value.LastVerificationAt, &value.HealthMode, + ) + if errors.Is(err, pgx.ErrNoRows) { + return SecurityEventConnection{}, ErrSecurityEventConnectionNotFound + } + return value, err +} + +func (s *Store) BindSecurityEventConnection(ctx context.Context, connectionID, audience, streamID, lifecycle string, expectedVersion int64) (SecurityEventConnection, error) { + tag, err := s.pool.Exec(ctx, `UPDATE gateway_security_event_connections +SET audience=$2,stream_id=$3::uuid,lifecycle_status=$4,last_error_category=NULL,version=version+1,updated_at=now() +WHERE connection_id=$1::uuid AND version=$5`, connectionID, audience, streamID, lifecycle, expectedVersion) + if err != nil { + return SecurityEventConnection{}, err + } + if tag.RowsAffected() == 0 { + return SecurityEventConnection{}, ErrSecurityEventConnectionConflict + } + return s.SecurityEventConnection(ctx) +} + +func (s *Store) UpdateSecurityEventConnectionLifecycle(ctx context.Context, connectionID, lifecycle string, errorCategory *string) (SecurityEventConnection, error) { + tag, err := s.pool.Exec(ctx, `UPDATE gateway_security_event_connections +SET lifecycle_status=$2,last_error_category=$3,version=version+1,updated_at=now() WHERE connection_id=$1::uuid`, connectionID, lifecycle, errorCategory) + if err != nil { + return SecurityEventConnection{}, err + } + if tag.RowsAffected() == 0 { + return SecurityEventConnection{}, ErrSecurityEventConnectionNotFound + } + return s.SecurityEventConnection(ctx) +} + +func (s *Store) SetSecurityEventNextCredential(ctx context.Context, connectionID, reference string) (SecurityEventConnection, error) { + tag, err := s.pool.Exec(ctx, `UPDATE gateway_security_event_connections +SET next_credential_ref=$2,lifecycle_status='rotating',last_error_category=NULL,version=version+1,updated_at=now() +WHERE connection_id=$1::uuid`, connectionID, reference) + if err != nil { + return SecurityEventConnection{}, err + } + if tag.RowsAffected() == 0 { + return SecurityEventConnection{}, ErrSecurityEventConnectionNotFound + } + return s.SecurityEventConnection(ctx) +} + +func (s *Store) PromoteSecurityEventCredential(ctx context.Context, connectionID, nextReference string) (SecurityEventConnection, error) { + tag, err := s.pool.Exec(ctx, `UPDATE gateway_security_event_connections +SET credential_ref=$2,next_credential_ref=NULL,lifecycle_status='bootstrap',last_error_category=NULL, + version=version+1,updated_at=now() +WHERE connection_id=$1::uuid AND next_credential_ref=$2`, connectionID, nextReference) + if err != nil { + return SecurityEventConnection{}, err + } + if tag.RowsAffected() == 0 { + return SecurityEventConnection{}, ErrSecurityEventConnectionConflict + } + return s.SecurityEventConnection(ctx) +} + +func (s *Store) DeleteSecurityEventConnection(ctx context.Context, connectionID string) error { + if _, err := uuid.Parse(connectionID); err != nil { + return ErrSecurityEventConnectionNotFound + } + tag, err := s.pool.Exec(ctx, `DELETE FROM gateway_security_event_connections WHERE connection_id=$1::uuid`, connectionID) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrSecurityEventConnectionNotFound + } + return nil +} diff --git a/apps/api/migrations/0063_security_event_connections.sql b/apps/api/migrations/0063_security_event_connections.sql new file mode 100644 index 0000000..21171b8 --- /dev/null +++ b/apps/api/migrations/0063_security_event_connections.sql @@ -0,0 +1,40 @@ +CREATE TABLE IF NOT EXISTS gateway_security_event_connections ( + connection_id uuid PRIMARY KEY, + singleton boolean NOT NULL DEFAULT true UNIQUE CHECK (singleton), + transmitter_issuer text NOT NULL, + endpoint_url text NOT NULL, + audience text, + stream_id uuid, + credential_ref text NOT NULL, + next_credential_ref text, + lifecycle_status text NOT NULL, + version bigint NOT NULL DEFAULT 1, + last_error_category text, + idempotency_key text NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + CONSTRAINT gateway_security_event_connection_status CHECK (lifecycle_status IN ( + 'connecting','verifying','bootstrap','enabled','degraded','rotating', + 'disconnect_pending','retiring','error' + )), + CONSTRAINT gateway_security_event_connection_stream_pair CHECK ( + (audience IS NULL AND stream_id IS NULL) OR (audience IS NOT NULL AND stream_id IS NOT NULL) + ) +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_gateway_security_event_connections_idempotency + ON gateway_security_event_connections(idempotency_key); + +CREATE TABLE IF NOT EXISTS gateway_security_event_connection_idempotency ( + operation text NOT NULL, + idempotency_key text NOT NULL, + request_hash text NOT NULL, + response jsonb NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (operation, idempotency_key), + CONSTRAINT gateway_security_event_connection_idempotency_operation + CHECK (operation IN ('connect','verify','rotate','disconnect')) +); + +CREATE INDEX IF NOT EXISTS idx_gateway_security_event_connection_idempotency_created + ON gateway_security_event_connection_idempotency(created_at); diff --git a/apps/web/src/App.tsx b/apps/web/src/App.tsx index d115dd0..9792c91 100644 --- a/apps/web/src/App.tsx +++ b/apps/web/src/App.tsx @@ -32,6 +32,7 @@ import type { PricingRuleSet, RateLimitWindow, RuntimePolicySet, + SecurityEventConnectionResponse, UserGroupUpsertRequest, UserGroup, WalletRechargeRequest, @@ -46,6 +47,7 @@ import { createPlatform, createTenant, createUserGroup, + connectSecurityEventTransmitter, deleteAccessRule, deleteApiKey, deleteFileStorageChannel, @@ -53,6 +55,7 @@ import { deletePlatform, deleteTenant, deleteUserGroup, + disconnectSecurityEventConnection, GatewayApiError, getClientCustomizationSettings, getHealth, @@ -61,6 +64,7 @@ import { getFileStorageSettings, getNetworkProxyConfig, getRunnerPolicy, + getSecurityEventConnection, getWalletSummary, listAccessRules, listAuditLogs, @@ -93,6 +97,7 @@ import { rechargeUserWalletBalance, replacePlatformModels, restoreModelRuntimeStatus, + rotateSecurityEventConnectionCredential, type HealthResponse, updateAccessRule, updateApiKeyScopes, @@ -104,6 +109,7 @@ import { updatePlatformDynamicPriority, updateTenant, updateUserGroup, + verifySecurityEventConnection, } from './api'; import type { ConsoleData, StatItem } from './app-state'; import { AppShell } from './components/layout/AppShell'; @@ -173,6 +179,7 @@ type DataKey = | 'clientCustomizationSettings' | 'fileStorageChannels' | 'fileStorageSettings' + | 'securityEventConnection' | 'platforms' | 'models' | 'providers' @@ -222,6 +229,7 @@ export function App() { const [clientCustomizationSettings, setClientCustomizationSettings] = useState(null); const [fileStorageChannels, setFileStorageChannels] = useState([]); const [fileStorageSettings, setFileStorageSettings] = useState(null); + const [securityEventConnection, setSecurityEventConnection] = useState(null); const [providers, setProviders] = useState([]); const [baseModels, setBaseModels] = useState([]); const [pricingRules, setPricingRules] = useState([]); @@ -383,6 +391,7 @@ export function App() { modelRateLimits, modelRateLimitsUpdatedAt, runtimePolicySets, + securityEventConnection, taskResult, tasks, tenants, @@ -390,7 +399,7 @@ export function App() { users, walletAccounts, walletTransactions, - }), [accessRules, apiKeys, auditLogs, baseModels, clientCustomizationSettings, currentUser, currentUserGroups, fileStorageChannels, fileStorageSettings, modelCatalog, modelRateLimits, modelRateLimitsUpdatedAt, models, networkProxyConfig, platforms, pricingRuleSets, pricingRules, providers, rateLimitWindows, runnerPolicy, runtimePolicySets, taskResult, tasks, tenants, userGroups, users, walletAccounts, walletTransactions]); + }), [accessRules, apiKeys, auditLogs, baseModels, clientCustomizationSettings, currentUser, currentUserGroups, fileStorageChannels, fileStorageSettings, modelCatalog, modelRateLimits, modelRateLimitsUpdatedAt, models, networkProxyConfig, platforms, pricingRuleSets, pricingRules, providers, rateLimitWindows, runnerPolicy, runtimePolicySets, securityEventConnection, taskResult, tasks, tenants, userGroups, users, walletAccounts, walletTransactions]); async function refresh(nextToken = token) { await ensureRouteData(nextToken, true); @@ -480,6 +489,9 @@ export function App() { case 'fileStorageSettings': setFileStorageSettings(await getFileStorageSettings(nextToken)); return; + case 'securityEventConnection': + setSecurityEventConnection(await getSecurityEventConnection(nextToken)); + return; case 'playgroundModels': setPlaygroundModels((await listPlayableModels(nextToken)).items); return; @@ -1017,6 +1029,44 @@ export function App() { } } + async function connectSecurityEvents(transmitterIssuer: string) { + setCoreState('loading'); + setCoreMessage(''); + try { + const connection = await connectSecurityEventTransmitter(token, transmitterIssuer); + setSecurityEventConnection(connection); + setCoreState('ready'); + setCoreMessage('认证中心安全事件连接正在验证。'); + } catch (err) { + setCoreState('error'); + setCoreMessage(err instanceof Error ? err.message : '连接认证中心失败'); + throw err; + } + } + + async function refreshSecurityEvents() { + const connection = await getSecurityEventConnection(token); + setSecurityEventConnection(connection); + } + + async function verifySecurityEvents() { + const connection = securityEventConnection?.connection; + if (!connection) return; + setSecurityEventConnection(await verifySecurityEventConnection(token, connection.version)); + } + + async function rotateSecurityEventsCredential() { + const connection = securityEventConnection?.connection; + if (!connection) return; + setSecurityEventConnection(await rotateSecurityEventConnectionCredential(token, connection.version)); + } + + async function disconnectSecurityEvents() { + const connection = securityEventConnection?.connection; + if (!connection) return; + setSecurityEventConnection(await disconnectSecurityEventConnection(token, connection.version)); + } + async function removeFileStorageChannel(channelId: string) { setCoreState('loading'); setCoreMessage(''); @@ -1095,6 +1145,7 @@ export function App() { setPlaygroundModels([]); setNetworkProxyConfig(null); setFileStorageChannels([]); + setSecurityEventConnection(null); setProviders([]); setBaseModels([]); setPricingRules([]); @@ -1332,6 +1383,11 @@ export function App() { onSaveFileStorageChannel={saveFileStorageChannel} onSaveFileStorageSettings={saveFileStorageSettings} onSaveClientCustomizationSettings={saveClientCustomizationSettings} + onConnectSecurityEvents={connectSecurityEvents} + onDisconnectSecurityEvents={disconnectSecurityEvents} + onRefreshSecurityEvents={refreshSecurityEvents} + onRotateSecurityEventsCredential={rotateSecurityEventsCredential} + onVerifySecurityEvents={verifySecurityEvents} onSaveTenant={saveTenant} onSaveUser={saveUser} onRechargeUserWalletBalance={rechargeUserWallet} @@ -1553,7 +1609,7 @@ function dataKeysForRoute( case 'accessRules': return ['accessRules', 'userGroups', 'platforms', 'models']; case 'systemSettings': - return ['clientCustomizationSettings', 'fileStorageSettings', 'fileStorageChannels']; + return ['clientCustomizationSettings', 'fileStorageSettings', 'fileStorageChannels', 'securityEventConnection']; default: return []; } diff --git a/apps/web/src/api.test.ts b/apps/web/src/api.test.ts index 3e50a58..ddaa1f1 100644 --- a/apps/web/src/api.test.ts +++ b/apps/web/src/api.test.ts @@ -1,5 +1,6 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import { + connectSecurityEventTransmitter, deleteOIDCBrowserSession, GatewayApiError, gatewayErrorMessage, @@ -34,6 +35,29 @@ describe('Gateway provisioning errors', () => { }); }); +describe('security event connection transport', () => { + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it('sends only the public transmitter issuer and concurrency headers', async () => { + const fetchMock = vi.fn().mockResolvedValue(new Response(JSON.stringify({ connected: true }), { + status: 202, + headers: { 'Content-Type': 'application/json' }, + })); + vi.stubGlobal('fetch', fetchMock); + vi.stubGlobal('crypto', { randomUUID: () => '00000000-0000-4000-8000-000000000000' }); + + await connectSecurityEventTransmitter('manager-token', 'https://auth.example/ssf'); + + const [, init] = fetchMock.mock.calls[0] as [string, RequestInit]; + expect(init.method).toBe('PUT'); + expect(JSON.parse(String(init.body))).toEqual({ transmitter_issuer: 'https://auth.example/ssf' }); + expect(String(init.body)).not.toMatch(/secret|bearer|scope|credential/i); + expect(new Headers(init.headers).get('Idempotency-Key')).toContain('ssf-connect-'); + }); +}); + describe('OIDC browser session transport', () => { afterEach(() => { vi.unstubAllGlobals(); diff --git a/apps/web/src/api.ts b/apps/web/src/api.ts index a548d99..defcd37 100644 --- a/apps/web/src/api.ts +++ b/apps/web/src/api.ts @@ -43,6 +43,7 @@ import type { RateLimitWindow, RuntimePolicySet, RuntimePolicySetUpsertRequest, + SecurityEventConnectionResponse, UserGroup, UserGroupUpsertRequest, WalletAdjustmentResponse, @@ -1000,6 +1001,40 @@ export async function updateClientCustomizationSettings( }); } +const securityEventConnectionPath = '/api/admin/system/identity/security-events/connection'; + +export async function getSecurityEventConnection(token: string): Promise { + return request(securityEventConnectionPath, { token }); +} + +export async function connectSecurityEventTransmitter(token: string, transmitterIssuer: string): Promise { + return request(securityEventConnectionPath, { + body: { transmitter_issuer: transmitterIssuer }, method: 'PUT', token, + headers: { 'Idempotency-Key': `ssf-connect-${crypto.randomUUID()}` }, + }); +} + +export async function verifySecurityEventConnection(token: string, version: number): Promise { + return securityEventConnectionWrite(token, '/verify', version, 'POST', 'verify'); +} + +export async function rotateSecurityEventConnectionCredential(token: string, version: number): Promise { + return securityEventConnectionWrite(token, '/rotate-credential', version, 'POST', 'rotate'); +} + +export async function disconnectSecurityEventConnection(token: string, version: number): Promise { + return securityEventConnectionWrite(token, '', version, 'DELETE', 'disconnect'); +} + +function securityEventConnectionWrite(token: string, suffix: string, version: number, method: string, action: string) { + return request(securityEventConnectionPath + suffix, { + method, token, headers: { + 'Idempotency-Key': `ssf-${action}-${crypto.randomUUID()}`, + 'If-Match': `W/"${version}"`, + }, + }); +} + export async function getPublicClientCustomization(): Promise { return request('/api/v1/public/client-customization', { auth: false }); } diff --git a/apps/web/src/app-state.ts b/apps/web/src/app-state.ts index a45c190..9ef928f 100644 --- a/apps/web/src/app-state.ts +++ b/apps/web/src/app-state.ts @@ -23,6 +23,7 @@ import type { PricingRuleSet, RateLimitWindow, RuntimePolicySet, + SecurityEventConnectionResponse, UserGroup, } from '@easyai-ai-gateway/contracts'; @@ -48,6 +49,7 @@ export interface ConsoleData { modelRateLimits: ModelRateLimitStatus[]; modelRateLimitsUpdatedAt: number | null; runtimePolicySets: RuntimePolicySet[]; + securityEventConnection: SecurityEventConnectionResponse | null; taskResult: GatewayTask | null; tasks: GatewayTask[]; tenants: GatewayTenant[]; diff --git a/apps/web/src/pages/AdminPage.tsx b/apps/web/src/pages/AdminPage.tsx index 829fe3e..e406a37 100644 --- a/apps/web/src/pages/AdminPage.tsx +++ b/apps/web/src/pages/AdminPage.tsx @@ -82,6 +82,11 @@ export function AdminPage(props: { onSaveClientCustomizationSettings: (input: ClientCustomizationSettingsUpdateRequest) => Promise; onSaveFileStorageChannel: (input: FileStorageChannelUpsertRequest, channelId?: string) => Promise; onSaveFileStorageSettings: (input: FileStorageSettingsUpdateRequest) => Promise; + onConnectSecurityEvents: (transmitterIssuer: string) => Promise; + onDisconnectSecurityEvents: () => Promise; + onRefreshSecurityEvents: () => Promise; + onRotateSecurityEventsCredential: () => Promise; + onVerifySecurityEvents: () => Promise; onSaveTenant: (input: GatewayTenantUpsertRequest, tenantId?: string) => Promise; onSaveUser: (input: GatewayUserUpsertRequest, userId?: string) => Promise; onRechargeUserWalletBalance: (userId: string, input: WalletRechargeRequest) => Promise; @@ -189,12 +194,18 @@ export function AdminPage(props: { channels={props.data.fileStorageChannels} clientCustomizationSettings={props.data.clientCustomizationSettings} settings={props.data.fileStorageSettings} + securityEventConnection={props.data.securityEventConnection} message={props.operationMessage} state={props.state} onDeleteFileStorageChannel={props.onDeleteFileStorageChannel} onSaveFileStorageChannel={props.onSaveFileStorageChannel} onSaveFileStorageSettings={props.onSaveFileStorageSettings} onSaveClientCustomizationSettings={props.onSaveClientCustomizationSettings} + onConnectSecurityEvents={props.onConnectSecurityEvents} + onDisconnectSecurityEvents={props.onDisconnectSecurityEvents} + onRefreshSecurityEvents={props.onRefreshSecurityEvents} + onRotateSecurityEventsCredential={props.onRotateSecurityEventsCredential} + onVerifySecurityEvents={props.onVerifySecurityEvents} /> )} diff --git a/apps/web/src/pages/admin/SystemSettingsPanel.tsx b/apps/web/src/pages/admin/SystemSettingsPanel.tsx index 93bc7dc..0367987 100644 --- a/apps/web/src/pages/admin/SystemSettingsPanel.tsx +++ b/apps/web/src/pages/admin/SystemSettingsPanel.tsx @@ -1,10 +1,10 @@ import { useEffect, useState, type FormEvent } from 'react'; -import { Database, Pencil, Plus, RotateCcw, Save, ServerCog, Settings2, Trash2 } from 'lucide-react'; -import type { ClientCustomizationSettings, ClientCustomizationSettingsUpdateRequest, FileStorageChannel, FileStorageChannelUpsertRequest, FileStorageSettings, FileStorageSettingsUpdateRequest } from '@easyai-ai-gateway/contracts'; +import { Database, Link2, Pencil, Plus, RefreshCw, RotateCcw, Save, ServerCog, Settings2, ShieldCheck, Trash2, Unplug } from 'lucide-react'; +import type { ClientCustomizationSettings, ClientCustomizationSettingsUpdateRequest, FileStorageChannel, FileStorageChannelUpsertRequest, FileStorageSettings, FileStorageSettingsUpdateRequest, SecurityEventConnectionResponse } from '@easyai-ai-gateway/contracts'; import { Badge, Button, Card, CardContent, CardHeader, CardTitle, ConfirmDialog, FormDialog, Input, Label, Select, Tabs, Textarea } from '../../components/ui'; import type { LoadState } from '../../types'; -type SystemSettingsTab = 'fileStorage' | 'clientCustomization'; +type SystemSettingsTab = 'fileStorage' | 'clientCustomization' | 'securityEvents'; type ClientCustomizationForm = { clientEnglishName: string; @@ -58,11 +58,17 @@ export function SystemSettingsPanel(props: { clientCustomizationSettings: ClientCustomizationSettings | null; message: string; settings: FileStorageSettings | null; + securityEventConnection: SecurityEventConnectionResponse | null; state: LoadState; onDeleteFileStorageChannel: (channelId: string) => Promise; onSaveClientCustomizationSettings: (input: ClientCustomizationSettingsUpdateRequest) => Promise; onSaveFileStorageChannel: (input: FileStorageChannelUpsertRequest, channelId?: string) => Promise; onSaveFileStorageSettings: (input: FileStorageSettingsUpdateRequest) => Promise; + onConnectSecurityEvents: (transmitterIssuer: string) => Promise; + onDisconnectSecurityEvents: () => Promise; + onRefreshSecurityEvents: () => Promise; + onRotateSecurityEventsCredential: () => Promise; + onVerifySecurityEvents: () => Promise; }) { const [activeTab, setActiveTab] = useState('fileStorage'); const [dialogOpen, setDialogOpen] = useState(false); @@ -72,6 +78,8 @@ export function SystemSettingsPanel(props: { const [clientCustomizationForm, setClientCustomizationForm] = useState(() => clientCustomizationSettingsToForm(props.clientCustomizationSettings)); const [settingsPolicy, setSettingsPolicy] = useState(() => normalizeResultUploadPolicy(props.settings?.resultUploadPolicy)); const [localError, setLocalError] = useState(''); + const [transmitterIssuer, setTransmitterIssuer] = useState('https://auth.51easyai.com/ssf'); + const [disconnectOpen, setDisconnectOpen] = useState(false); useEffect(() => { setSettingsPolicy(normalizeResultUploadPolicy(props.settings?.resultUploadPolicy)); @@ -81,6 +89,13 @@ export function SystemSettingsPanel(props: { setClientCustomizationForm(clientCustomizationSettingsToForm(props.clientCustomizationSettings)); }, [props.clientCustomizationSettings]); + useEffect(() => { + const lifecycle = props.securityEventConnection?.connection?.lifecycleStatus; + if (!['connecting', 'verifying', 'bootstrap', 'rotating', 'disconnect_pending', 'retiring'].includes(lifecycle ?? '')) return undefined; + const timer = window.setInterval(() => { void props.onRefreshSecurityEvents(); }, 2000); + return () => window.clearInterval(timer); + }, [props.securityEventConnection?.connection?.lifecycleStatus, props.onRefreshSecurityEvents]); + function openCreateDialog() { setEditingChannel(null); setForm(defaultChannelForm(`server-main-${Date.now().toString(36)}`)); @@ -162,6 +177,7 @@ export function SystemSettingsPanel(props: { tabs={[ { value: 'fileStorage', label: '文件存储', icon: }, { value: 'clientCustomization', label: '客户端自定义', icon: }, + { value: 'securityEvents', label: '认证中心安全事件', icon: }, ]} onValueChange={setActiveTab} /> @@ -275,6 +291,20 @@ export function SystemSettingsPanel(props: { )} + {activeTab === 'securityEvents' && ( + setDisconnectOpen(true)} + onRefresh={props.onRefreshSecurityEvents} + onRotate={props.onRotateSecurityEventsCredential} + onVerify={props.onVerifySecurityEvents} + /> + )} + setPendingDeleteChannel(null)} onConfirm={() => pendingDeleteChannel ? deleteChannel(pendingDeleteChannel) : undefined} /> + setDisconnectOpen(false)} + onConfirm={async () => { await props.onDisconnectSecurityEvents(); setDisconnectOpen(false); }} + /> ); } +function SecurityEventConnectionPanel(props: { + connection: SecurityEventConnectionResponse | null; + issuer: string; + loading: boolean; + onIssuerChange(value: string): void; + onConnect(issuer: string): Promise; + onDisconnect(): void; + onRefresh(): Promise; + onRotate(): Promise; + onVerify(): Promise; +}) { + const connection = props.connection?.connection; + const prerequisites = props.connection?.prerequisites; + const missing = [ + !prerequisites?.oidcConfigured ? '先完成 OIDC Issuer、Tenant 和 Audience 配置' : '', + !prerequisites?.introspectionClientConfigured ? '配置 RFC 7662 机器 Client ID / Secret' : '', + !prerequisites?.publicBaseUrlConfigured ? '配置 AI_GATEWAY_PUBLIC_BASE_URL' : '', + ].filter(Boolean); + return ( +
+
+
+ SSF / CAEP 实时会话撤销 + 只需连接认证中心。Gateway 后端自动生成并托管 Push Bearer，复用现有 RFC 7662 机器 Client，不需要重启。 +
+ + {securityEventLifecycleLabel(connection?.lifecycleStatus)} + + +
+ {!connection && ( + + +
+ 连接认证中心 +

请先在认证中心 Application 的“安全事件流”中完成“准备 Gateway 接入”。

+
+ {missing.length > 0 &&
{missing.join('；')}
} +
+ 复用机器 Client: {prerequisites?.managementClientId || '未配置'} +
+ + + + + )} + {connection && ( + + +
+ 机器 Client: {connection.managementClientId} + Receiver Endpoint: {connection.receiverEndpoint} + Issuer: {connection.transmitterIssuer} + Audience: {connection.audience ?? '正在获取'} + Stream ID: {connection.streamId ?? '正在创建'} + 健康模式: {securityEventHealthLabel(connection.healthMode)} + 最近 Verification: {connection.lastVerificationAt ? new Date(connection.lastVerificationAt).toLocaleString() : '尚未成功'} + {connection.lastErrorCategory && 最近错误: {connection.lastErrorCategory}} +
+
+ + + +
+ + + )} +
+ ); +} + +function securityEventLifecycleLabel(status?: string) { + return ({ connecting: '连接中', verifying: '验证中', bootstrap: '内省保护期', enabled: '推送健康', degraded: '已降级', rotating: '轮换中', disconnect_pending: '等待断开', retiring: '安全退役中', error: '连接异常' } as Record)[status ?? ''] ?? '未连接'; +} + +function securityEventHealthLabel(mode: string) { + return ({ disabled: '未启用', bootstrap: 'RFC 7662 启动保护', push_healthy: 'Push 健康', introspection_fallback: 'RFC 7662 降级' } as Record)[mode] ?? mode; +} + function defaultClientCustomizationForm(): ClientCustomizationForm { return { clientEnglishName: 'EasyAI AI Gateway', diff --git a/deploy/kubernetes/security-events-secret-rbac.yaml b/deploy/kubernetes/security-events-secret-rbac.yaml new file mode 100644 index 0000000..ad40b69 --- /dev/null +++ b/deploy/kubernetes/security-events-secret-rbac.yaml @@ -0,0 +1,34 @@ +# Reference overlay for installations that run the Gateway in Kubernetes. +# Keep the Secret empty: the Gateway writes only generated Push Bearers. +apiVersion: v1 +kind: Secret +metadata: + name: easyai-gateway-security-events + namespace: easyai +type: Opaque +data: {} +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: easyai-gateway-security-events + namespace: easyai +rules: + - apiGroups: [""] + resources: ["secrets"] + resourceNames: ["easyai-gateway-security-events"] + verbs: ["get", "update", "patch"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: easyai-gateway-security-events + namespace: easyai +subjects: + - kind: ServiceAccount + name: easyai-ai-gateway + namespace: easyai +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: easyai-gateway-security-events diff --git a/docker-compose.yml b/docker-compose.yml index eda6f9b..3e0852f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -17,16 +17,8 @@ x-api-environment: &api-environment OIDC_INTROSPECTION_ENABLED: ${OIDC_INTROSPECTION_ENABLED:-false} OIDC_INTROSPECTION_CLIENT_ID: ${OIDC_INTROSPECTION_CLIENT_ID:-} OIDC_INTROSPECTION_CLIENT_SECRET: ${OIDC_INTROSPECTION_CLIENT_SECRET:-} - OIDC_SECURITY_EVENTS_ENABLED: ${OIDC_SECURITY_EVENTS_ENABLED:-false} - OIDC_SECURITY_EVENTS_TRANSMITTER_ISSUER: ${OIDC_SECURITY_EVENTS_TRANSMITTER_ISSUER:-} - OIDC_SECURITY_EVENTS_RECEIVER_AUDIENCE: ${OIDC_SECURITY_EVENTS_RECEIVER_AUDIENCE:-} - OIDC_SECURITY_EVENTS_BEARER_SECRET: ${OIDC_SECURITY_EVENTS_BEARER_SECRET:-} - OIDC_SECURITY_EVENTS_BEARER_SECRET_NEXT: ${OIDC_SECURITY_EVENTS_BEARER_SECRET_NEXT:-} - OIDC_SECURITY_EVENTS_PUBLIC_ENDPOINT: ${OIDC_SECURITY_EVENTS_PUBLIC_ENDPOINT:-} - OIDC_SECURITY_EVENTS_STREAM_ID: ${OIDC_SECURITY_EVENTS_STREAM_ID:-} - OIDC_SECURITY_EVENTS_MANAGEMENT_TOKEN_URL: ${OIDC_SECURITY_EVENTS_MANAGEMENT_TOKEN_URL:-} - OIDC_SECURITY_EVENTS_MANAGEMENT_CLIENT_ID: ${OIDC_SECURITY_EVENTS_MANAGEMENT_CLIENT_ID:-} - OIDC_SECURITY_EVENTS_MANAGEMENT_CLIENT_SECRET: ${OIDC_SECURITY_EVENTS_MANAGEMENT_CLIENT_SECRET:-} + OIDC_SECURITY_EVENTS_SECRET_STORE: file + OIDC_SECURITY_EVENTS_SECRET_DIR: /app/data/security-events/secrets OIDC_SECURITY_EVENTS_HEARTBEAT_INTERVAL_SECONDS: ${OIDC_SECURITY_EVENTS_HEARTBEAT_INTERVAL_SECONDS:-60} OIDC_SECURITY_EVENTS_STALE_AFTER_SECONDS: ${OIDC_SECURITY_EVENTS_STALE_AFTER_SECONDS:-180} OIDC_SECURITY_EVENTS_CLOCK_SKEW_SECONDS: ${OIDC_SECURITY_EVENTS_CLOCK_SKEW_SECONDS:-60} diff --git a/docs/security/ssf-session-revocation.md b/docs/security/ssf-session-revocation.md index f2cc589..aee2578 100644 --- a/docs/security/ssf-session-revocation.md +++ b/docs/security/ssf-session-revocation.md @@ -1,56 +1,67 @@ # SSF/CAEP 实时会话撤销运行手册 -Gateway 可选接收 Auth Center 通过 RFC 8935 Push 投递的 RFC 8417 Security Event Token，并处理 OpenID CAEP `session-revoked`。默认关闭；关闭时不会注册接收路由或启动状态机，也不会改变 OIDC、BFF、API Key 或 Legacy JWT 行为。 +Gateway 可选接收 Auth Center 通过 RFC 8935 Push 投递的 RFC 8417 Security Event Token，并处理 OpenID CAEP `session-revoked`。功能由数据库中的连接资源控制：没有连接时保持原有 OIDC、BFF、API Key 和 Legacy JWT 行为；不再使用 `OIDC_SECURITY_EVENTS_ENABLED`。 -## 启用前提 +## 用户接入流程 -- Auth Center 已创建属于当前 Application/租户的 paused Stream，Audience 为 `urn:easyai:ssf:receiver:{application-public-id}`。 -- Gateway OIDC `iss`、`tid` 与 Stream 绑定完全一致；Stream ID 为公开 UUID，不使用认证内核 ID。 -- 接收 Bearer、Management Client Secret、RFC 7662 Client Secret 仅从 Git 忽略的 `.env`、Kubernetes Secret 或 Secret Manager 注入。 -- `OIDC_SECURITY_EVENTS_PUBLIC_ENDPOINT` 必须是外部可访问的精确 HTTPS 地址；本地测试环境才允许 localhost HTTP。 -- Management Client 至少有 `ssf.stream.read ssf.stream.verify`，不能使用浏览器 PKCE Client。 +用户只执行两项操作： -完整配置键见仓库根目录 `.env.example`。启用时 `OIDC_SECURITY_EVENTS_ENABLED=true`，且 RFC 7662 Client ID/Secret 必须同时存在，否则启动失败。 +1. 在认证中心 Application 的“安全事件流”页面选择现有 RFC 7662 机器 Client，点击“准备 Gateway 接入”。认证中心自动合并 SSF 管理 Scope；不会创建第二个 Client。 +2. 在 Gateway“系统设置 → 认证中心安全事件”中填写认证中心公开 SSF Issuer，点击“连接认证中心”。 -## 启用顺序 +Gateway 后端随后自动读取 Discovery、生成并托管 256 bit Push Bearer、创建 paused Stream、完成 Verification、首次启用 Stream，并进入 360 秒 RFC 7662 bootstrap。浏览器看不到 Push Bearer、OAuth Secret 或 Secret 引用，管理员不编辑 Secret 文件，也不需要重启 Gateway。 -1. 运行数据库迁移，部署仍保持功能关闭。 -2. 在 Secret Store 创建接收 Bearer、Management Client Secret 和内省 Client Secret；不要把值放进日志或工单。 -3. 启用 Gateway。启动模式为 `bootstrap`，所有 OIDC Token 都先内省。 -4. 在 Auth Center 对 paused Stream 发起 Verification，确认 Gateway 返回空 Body 的 `202`。 -5. 启用 Stream。首次 Verification 成功后仍内省 360 秒，覆盖启用前已签发的最长 300 秒 Token。 -6. 观察 `easyai_gateway_ssf_mode{mode="push_healthy"}` 变为 1，再执行测试用户撤销。 +前置配置只有既有 OIDC/RFC 7662 配置和 Gateway 公共地址： -健康时 Gateway 每 60 秒请求一次 Verification，不按业务 QPS 请求 Auth Center。180 秒没有匹配的有效 Verification 会进入 `introspection_fallback`；只有最近收到的 Stream 状态为 `enabled` 时，连续两次 Verification 成功才会恢复，`paused`/`disabled` 状态下心跳不会错误恢复推送信任。降级期间内省不可用会对 OIDC 返回可审计的 503，API Key 继续工作。 +- `OIDC_ISSUER`、`OIDC_TENANT_ID`； +- `OIDC_INTROSPECTION_ENABLED=true`； +- `OIDC_INTROSPECTION_CLIENT_ID/SECRET`，与认证中心准备 Receiver 时选择的机器 Client 相同； +- `AI_GATEWAY_PUBLIC_BASE_URL`，生产必须是认证中心可访问的 HTTPS 地址。 -## 零停机 Bearer 轮换 +第一版一个 Gateway 部署只允许一个认证中心连接。下游业务服务无需接入 SSF。 -1. 在 Gateway 写入 `OIDC_SECURITY_EVENTS_BEARER_SECRET_NEXT`，同时接受 current/next。 -2. 更新 Auth Center 的 Stream `credential_ref` 指向 next。 -3. 发起 Verification，并确认投递和状态机健康。 -4. 等待至少 180 秒后，将 next 提升为 current 并清空旧值。 +## SecretStore -Gateway 使用常量时间比较两个值。响应、日志、审计和指标都不得包含 Authorization Header 或原始 SET。 +本地和 Docker 使用 `file` 驱动。服务自动创建目录并强制目录 `0700`、文件 `0600`；Docker Compose 将目录放在持久化 `api_data` 卷中。用户不写入任何 `ssf-delivery-*` 文件。 -## 监控与告警 +Kubernetes 使用 `kubernetes` 驱动和一个部署时预创建的空 Secret： -`GET /metrics` 输出以下低基数指标；生产入口应仅允许监控网络访问该路径： +```text +OIDC_SECURITY_EVENTS_SECRET_STORE=kubernetes +OIDC_SECURITY_EVENTS_KUBERNETES_NAMESPACE=easyai +OIDC_SECURITY_EVENTS_KUBERNETES_SECRET_NAME=easyai-gateway-security-events +``` -- `easyai_gateway_ssf_receipts_total{outcome}`：accepted/rejected/duplicate。 -- `easyai_gateway_ssf_sessions_deleted_total`：被撤销删除的 BFF Session。 -- `easyai_gateway_ssf_watermark_rejections_total`：本地水位拒绝的 Token。 -- `easyai_gateway_ssf_processing_duration_seconds`：Receiver 数据库事务处理直方图，用于 P99 告警。 -- `easyai_gateway_ssf_verification_age_seconds` 和 `easyai_gateway_ssf_mode{mode}`。 -- `easyai_gateway_ssf_heartbeat_requests_total{outcome}`。 -- `easyai_gateway_oidc_introspection_total{outcome}`。 -- `easyai_gateway_jwks_refresh_failures_total{source}`：SSF/OIDC JWKS 刷新失败。 +参考清单见 `deploy/kubernetes/security-events-secret-rbac.yaml`。ServiceAccount 只能对这个固定 Secret 执行 `get/update/patch`，不能创建、删除或读取其他 Secret。数据库、API、浏览器、日志和审计只保存或展示非敏感连接元数据。 -至少配置以下告警：Verification age 超过 120 秒、fallback 超过 5 分钟、内省 failed 增长、SET rejected 增长、撤销端到端 P99 超过 3 秒。日志只能携带脱敏 Trace ID、Audit ID 和错误类别。 +## 动态运行与状态 -## 回滚 +Receiver 路由始终注册：没有连接时返回无敏感信息的 `404`；存在连接但凭据丢失时返回 `503` 并进入 `introspection_fallback`。OIDC Evaluator 始终挂载但在无连接时无副作用。 -先在 Auth Center 暂停 Stream，再保持 Gateway 启用但持续使用 RFC 7662，确认没有遗漏撤销后才关闭接收功能。保留迁移表和 Secret，不做破坏性数据库回滚。Legacy HS256 Token 和 API Key 不在本协议撤销范围内。 +连接生命周期包括 `connecting`、`verifying`、`bootstrap`、`enabled`、`degraded`、`rotating`、`disconnect_pending` 和 `retiring`。健康时每 60 秒一次 Verification，不随业务请求 QPS 增长；180 秒无匹配 Verification 时切换 RFC 7662。内省也不可用时 OIDC Fail Closed，API Key 继续工作。 -## 验收证据 +Verification 成功后仅在首次连接和主动轮换时自动启用 Stream。认证中心管理员之后手工暂停或禁用 Stream，Gateway 的“重新验证”只检查链路，不会擅自恢复远端状态。 -真实链路验收只能使用明确标记为测试用途的 Application，且必须在自动化测试通过后执行。报告需包含脱敏 Trace、Claims 结构、截图、Audit ID、投递延迟、fallback 与恢复证据。涉及人工、真实设备或第三方操作时记录 `SKIPPED_HUMAN_OR_THIRD_PARTY_REQUIRED`，Mock、契约和安全测试不得跳过。 +## 轮换与断开 + +“轮换凭据”由 Gateway 自动完成：生成 next、Receiver 同时接受 current/next、暂停并更新远端 Stream、Verification、重新启用、保留旧值至少 180 秒，最后提升 next 并删除旧值。中途失败保留两个凭据和 `rotating` 状态，可安全重试。 + +“断开连接”先删除远端 Stream，再进入至少 360 秒 `retiring`，期间继续应用撤销水位和 RFC 7662，避免旧 Token 因关闭功能重新有效。远端不可用时保持 `disconnect_pending` 并指数退避重试；不会提前删除 Push Secret。收据、水位和脱敏审计数据不会清表。 + +所有写管理接口要求 Gateway `Manager` 权限、`Idempotency-Key` 和 `If-Match`： + +```text +GET /api/admin/system/identity/security-events/connection +PUT /api/admin/system/identity/security-events/connection +POST /api/admin/system/identity/security-events/connection/verify +POST /api/admin/system/identity/security-events/connection/rotate-credential +DELETE /api/admin/system/identity/security-events/connection +``` + +## 监控、回滚与验收 + +`GET /metrics` 输出接收结果、删除 Session 数、水位拒绝数、Verification 年龄、健康模式、内省结果、JWKS 失败和处理延迟。至少告警 Verification age 超过 120 秒、fallback 超过 5 分钟、内省失败、SET 拒绝增长以及撤销 P99 超过 3 秒。日志只记录脱敏 Trace/Audit ID 和错误类别。 + +回滚时先在认证中心暂停 Stream，再通过 Gateway 安全断开；保留迁移表、水位和审计。Legacy HS256 Token 和 API Key 不在本协议撤销范围内。 + +真实链路只使用明确标记为测试用途的 Application，并且必须在自动化测试后执行。报告包含脱敏 Trace、Claims、截图、Audit ID、投递延迟、fallback 和恢复证据；需要人工或第三方操作时记录 `SKIPPED_HUMAN_OR_THIRD_PARTY_REQUIRED`。 diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index e4677e0..d229616 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -957,6 +957,34 @@ export interface ClientCustomizationSettingsUpdateRequest { iconPath?: string; } +export interface SecurityEventConnection { + connectionId: string; + transmitterIssuer: string; + receiverEndpoint: string; + audience?: string; + streamId?: string; + lifecycleStatus: 'connecting' | 'verifying' | 'bootstrap' | 'enabled' | 'degraded' | 'rotating' | 'disconnect_pending' | 'retiring' | 'error' | string; + healthMode: 'disabled' | 'bootstrap' | 'push_healthy' | 'introspection_fallback' | string; + managementClientId: string; + lastVerificationAt?: string; + lastErrorCategory?: string; + version: number; +} + +export interface SecurityEventConnectionPrerequisites { + oidcConfigured: boolean; + introspectionClientConfigured: boolean; + publicBaseUrlConfigured: boolean; + managementClientId: string; +} + +export interface SecurityEventConnectionResponse { + connected: boolean; + lifecycleStatus?: 'disconnected' | string; + connection?: SecurityEventConnection; + prerequisites?: SecurityEventConnectionPrerequisites; +} + export interface GatewayTask { id: string; kind: string;