diff --git a/.env.example b/.env.example index 1905d2a..25fd408 100644 --- a/.env.example +++ b/.env.example @@ -35,6 +35,22 @@ 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= +OIDC_SECURITY_EVENTS_HEARTBEAT_INTERVAL_SECONDS=60 +OIDC_SECURITY_EVENTS_STALE_AFTER_SECONDS=180 +OIDC_SECURITY_EVENTS_CLOCK_SKEW_SECONDS=60 # Controlled JIT is opt-in. When enabled, bind the validated Auth Center tid to # one existing active Gateway tenant; tokens never create Gateway tenants. OIDC_JIT_PROVISIONING_ENABLED=false diff --git a/README.md b/README.md index 5f98881..d4561c0 100644 --- a/README.md +++ b/README.md @@ -56,6 +56,8 @@ 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)。 + `pnpm dev` 会先创建数据库并执行 migration，然后并行启动： - `api:dev`：通过 `scripts/go-watch.mjs` 运行 Go API，监听 `.go`、`go.mod`、`go.sum` 变化并自动重启后端进程；watcher 会按进程组终止旧的 `go run` 和其子进程，避免热更新时残留进程占用 API 端口。 diff --git a/apps/api/docs/swagger.json b/apps/api/docs/swagger.json index bab1b40..5515870 100644 --- a/apps/api/docs/swagger.json +++ b/apps/api/docs/swagger.json @@ -5506,6 +5506,80 @@ } } }, + "/api/v1/security-events/ssf": { + "post": { + "description": "Optional endpoint. It validates a stream-specific Bearer before parsing and verifying an ES256 SET. A committed event and a duplicate jti both return an empty 202 response.", + "consumes": [ + "application/secevent+jwt" + ], + "produces": [ + "application/json" + ], + "tags": [ + "Security Events" + ], + "summary": "Receive an SSF Security Event Token", + "parameters": [ + { + "type": "string", + "description": "Bearer stream-specific-secret", + "name": "Authorization", + "in": "header", + "required": true + }, + { + "description": "Compact signed Security Event Token", + "name": "set", + "in": "body", + "required": true, + "schema": { + "type": "string" + } + } + ], + "responses": { + "202": { + "description": "Accepted" + }, + "400": { + "description": "Bad Request", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + }, + "401": { + "description": "Unauthorized", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + }, + "403": { + "description": "Forbidden", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + }, + "503": { + "description": "Service Unavailable", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } + } + } + } + }, "/api/v1/song/generations": { "post": { "security": [ diff --git a/apps/api/docs/swagger.yaml b/apps/api/docs/swagger.yaml index ac0cb65..1db16ed 100644 --- a/apps/api/docs/swagger.yaml +++ b/apps/api/docs/swagger.yaml @@ -6073,6 +6073,57 @@ paths: summary: 创建或执行 AI 任务 tags: - tasks + /api/v1/security-events/ssf: + post: + consumes: + - application/secevent+jwt + description: Optional endpoint. It validates a stream-specific Bearer before + parsing and verifying an ES256 SET. A committed event and a duplicate jti + both return an empty 202 response. + parameters: + - description: Bearer stream-specific-secret + in: header + name: Authorization + required: true + type: string + - description: Compact signed Security Event Token + in: body + name: set + required: true + schema: + type: string + produces: + - application/json + responses: + "202": + description: Accepted + "400": + description: Bad Request + schema: + additionalProperties: + type: string + type: object + "401": + description: Unauthorized + schema: + additionalProperties: + type: string + type: object + "403": + description: Forbidden + schema: + additionalProperties: + type: string + type: object + "503": + description: Service Unavailable + schema: + additionalProperties: + type: string + type: object + summary: Receive an SSF Security Event Token + tags: + - Security Events /api/v1/song/generations: post: consumes: diff --git a/apps/api/go.mod b/apps/api/go.mod index 69c6e99..df5bee9 100644 --- a/apps/api/go.mod +++ b/apps/api/go.mod @@ -9,6 +9,7 @@ require ( github.com/dop251/goja v0.0.0-20260311135729-065cd970411c github.com/dop251/goja_nodejs v0.0.0-20260212111938-1f56ff5bcf14 github.com/golang-jwt/jwt/v5 v5.2.2 + github.com/google/uuid v1.6.0 github.com/jackc/pgx/v5 v5.9.2 github.com/riverqueue/river v0.24.0 github.com/riverqueue/river/riverdriver/riverpgxv5 v0.24.0 diff --git a/apps/api/go.sum b/apps/api/go.sum index 7db61cb..792b0e9 100644 --- a/apps/api/go.sum +++ b/apps/api/go.sum @@ -19,6 +19,8 @@ github.com/golang-jwt/jwt/v5 v5.2.2 h1:Rl4B7itRWVtYIHFrSNd7vhTiz9UpLdi6gZhZ3wEeD github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk= github.com/google/pprof v0.0.0-20240727154555-813a5fbdbec8 h1:FKHo8hFI3A+7w0aUQuYXQ+6EN5stWmeY/AZqtM8xk9k= github.com/google/pprof v0.0.0-20240727154555-813a5fbdbec8/go.mod h1:K1liHPHnj73Fdn/EKuT8nrFqBihUSKXoLYU0BuatOYo= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 h1:Dj0L5fhJ9F82ZJyVOmBx6msDp/kfd1t9GRfny/mfJA0= github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= diff --git a/apps/api/internal/auth/auth.go b/apps/api/internal/auth/auth.go index b93a0b4..087b4b5 100644 --- a/apps/api/internal/auth/auth.go +++ b/apps/api/internal/auth/auth.go @@ -50,6 +50,8 @@ type User struct { APIKeyPrefix string `json:"apiKeyPrefix,omitempty"` APIKeyScopes []string `json:"apiKeyScopes,omitempty"` TokenExpiresAt time.Time `json:"-"` + TokenIssuedAt time.Time `json:"-"` + Issuer string `json:"-"` } type contextKey string diff --git a/apps/api/internal/auth/oidc.go b/apps/api/internal/auth/oidc.go index b767221..df62e60 100644 --- a/apps/api/internal/auth/oidc.go +++ b/apps/api/internal/auth/oidc.go @@ -5,6 +5,7 @@ import ( "crypto/ecdsa" "crypto/elliptic" "crypto/rsa" + "crypto/tls" "encoding/base64" "encoding/json" "errors" @@ -24,16 +25,31 @@ const maxOIDCResponseBytes = 1 << 20 const defaultOIDCHTTPTimeout = 10 * time.Second type OIDCConfig struct { - Issuer string - Audience string - TenantID string - RolePrefix string - RequiredScopes []string - JWKSCacheTTL time.Duration - IntrospectionEnabled bool - IntrospectionClientID string - IntrospectionClientSecret string - HTTPClient *http.Client + Issuer string + Audience string + TenantID string + RolePrefix string + RequiredScopes []string + JWKSCacheTTL time.Duration + IntrospectionEnabled bool + IntrospectionClientID string + IntrospectionClientSecret string + HTTPClient *http.Client + SecurityEventEvaluator func(context.Context, OIDCSecurityEventIdentity) (OIDCSecurityEventEvaluation, error) + IntrospectionObserver func(string) + JWKSRefreshFailureObserver func() +} + +type OIDCSecurityEventIdentity struct { + Issuer string + TenantID string + Subject string + IssuedAt time.Time +} + +type OIDCSecurityEventEvaluation struct { + Revoked bool + RequireIntrospection bool } type OIDCVerifier struct { @@ -75,7 +91,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 && (strings.TrimSpace(config.IntrospectionClientID) == "" || config.IntrospectionClientSecret == "") { + if (config.IntrospectionEnabled || config.SecurityEventEvaluator != nil) && (strings.TrimSpace(config.IntrospectionClientID) == "" || config.IntrospectionClientSecret == "") { return nil, errors.New("introspection client credentials are required when introspection is enabled") } if config.JWKSCacheTTL <= 0 { @@ -83,9 +99,12 @@ func NewOIDCVerifier(config OIDCConfig) (*OIDCVerifier, error) { } client := config.HTTPClient if client == nil { + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12} + transport.DisableCompression = true client = &http.Client{Timeout: defaultOIDCHTTPTimeout, CheckRedirect: func(_ *http.Request, _ []*http.Request) error { return http.ErrUseLastResponse - }} + }, Transport: transport} } return &OIDCVerifier{config: config, client: client, keys: map[string]any{}}, nil } @@ -125,6 +144,10 @@ func (v *OIDCVerifier) Verify(ctx context.Context, raw string) (*User, error) { if _, ok := numericDateClaim(claims["nbf"]); !ok { 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) @@ -136,7 +159,26 @@ func (v *OIDCVerifier) Verify(ctx context.Context, raw string) (*User, error) { if len(roles) == 0 { return nil, oidcUnauthorized("mapped role is missing", nil) } - if v.config.IntrospectionEnabled { + if v.config.SecurityEventEvaluator != nil { + evaluation, evaluateErr := v.config.SecurityEventEvaluator(ctx, OIDCSecurityEventIdentity{ + Issuer: v.config.Issuer, TenantID: v.config.TenantID, Subject: stringClaim(claims, "sub"), IssuedAt: issuedAt, + }) + if evaluateErr != nil { + return nil, NewRequestAuthError(http.StatusServiceUnavailable, "OIDC_SECURITY_EVENT_STATE_UNAVAILABLE", "认证撤销状态暂时不可用") + } + if evaluation.Revoked { + return nil, oidcUnauthorized("token was issued before the revocation watermark", nil) + } + if evaluation.RequireIntrospection { + active, introspectionErr := v.introspect(ctx, raw) + if introspectionErr != nil { + return nil, NewRequestAuthError(http.StatusServiceUnavailable, "OIDC_INTROSPECTION_UNAVAILABLE", "认证中心内省暂时不可用") + } + if !active { + return nil, oidcUnauthorized("token is inactive", nil) + } + } + } else if v.config.IntrospectionEnabled { active, err := v.introspect(ctx, raw) if err != nil || !active { return nil, oidcUnauthorized("token is inactive", err) @@ -148,7 +190,7 @@ func (v *OIDCVerifier) Verify(ctx context.Context, raw string) (*User, error) { } return &User{ ID: stringClaim(claims, "sub"), Username: username, Roles: roles, - TenantID: v.config.TenantID, Source: "oidc", TokenExpiresAt: expiresAt, + TenantID: v.config.TenantID, Source: "oidc", TokenExpiresAt: expiresAt, TokenIssuedAt: issuedAt, Issuer: v.config.Issuer, }, nil } @@ -161,6 +203,9 @@ func oidcUnauthorized(reason string, cause error) error { func (v *OIDCVerifier) key(ctx context.Context, kid string) (any, error) { if err := v.refresh(ctx, false); err != nil { + if v.config.JWKSRefreshFailureObserver != nil { + v.config.JWKSRefreshFailureObserver() + } return nil, err } v.mu.Lock() @@ -170,6 +215,9 @@ func (v *OIDCVerifier) key(ctx context.Context, kid string) (any, error) { return key, nil } if err := v.refresh(ctx, true); err != nil { + if v.config.JWKSRefreshFailureObserver != nil { + v.config.JWKSRefreshFailureObserver() + } return nil, err } v.mu.Lock() @@ -195,7 +243,7 @@ func (v *OIDCVerifier) refresh(ctx context.Context, force bool) error { if discovery.Issuer != v.config.Issuer || validatePublicURL(discovery.JWKSURI) != nil { return errors.New("OIDC discovery metadata is invalid") } - if v.config.IntrospectionEnabled && validatePublicURL(discovery.IntrospectionEndpoint) != nil { + if (v.config.IntrospectionEnabled || v.config.SecurityEventEvaluator != nil) && validatePublicURL(discovery.IntrospectionEndpoint) != nil { return errors.New("OIDC introspection metadata is invalid") } var set jsonWebKeySet @@ -222,6 +270,12 @@ func (v *OIDCVerifier) refresh(ctx context.Context, force bool) error { } func (v *OIDCVerifier) introspect(ctx context.Context, raw string) (bool, error) { + outcome := "failed" + defer func() { + if v.config.IntrospectionObserver != nil { + v.config.IntrospectionObserver(outcome) + } + }() v.mu.Lock() endpoint := v.introspectionEndpoint v.mu.Unlock() @@ -252,6 +306,11 @@ func (v *OIDCVerifier) introspect(ctx context.Context, raw string) (bool, error) if err := decoder.Decode(&result); err != nil { return false, errors.New("OIDC introspection response is invalid") } + if result.Active { + outcome = "active" + } else { + outcome = "inactive" + } return result.Active, nil } diff --git a/apps/api/internal/auth/oidc_test.go b/apps/api/internal/auth/oidc_test.go index ad625bf..16762a1 100644 --- a/apps/api/internal/auth/oidc_test.go +++ b/apps/api/internal/auth/oidc_test.go @@ -8,6 +8,7 @@ import ( "crypto/rsa" "encoding/base64" "encoding/json" + "errors" "math/big" "net/http" "net/http/httptest" @@ -153,6 +154,59 @@ func TestOIDCVerifierFailsClosedWhenIntrospectionMarksSessionInactive(t *testing } } +func TestOIDCSecurityEventEvaluationUsesWatermarkAndFallbackIntrospection(t *testing.T) { + key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + active, introspectionAvailable, introspectionCalls := true, true, 0 + var issuer string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + switch request.URL.Path { + case "/.well-known/openid-configuration": + _ = json.NewEncoder(w).Encode(map[string]any{"issuer": issuer, "jwks_uri": issuer + "/jwks", "introspection_endpoint": issuer + "/introspect"}) + case "/jwks": + _ = json.NewEncoder(w).Encode(map[string]any{"keys": []any{ecJWK("ec-key", &key.PublicKey)}}) + case "/introspect": + introspectionCalls++ + if !introspectionAvailable { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{"active": active}) + } + })) + defer server.Close() + issuer = server.URL + evaluation := OIDCSecurityEventEvaluation{RequireIntrospection: true} + verifier, err := NewOIDCVerifier(OIDCConfig{ + Issuer: issuer, Audience: "gateway-api", TenantID: "tenant-1", RolePrefix: "gateway.", + RequiredScopes: []string{"gateway.access"}, HTTPClient: server.Client(), + IntrospectionClientID: "gateway-introspection", IntrospectionClientSecret: "introspection-secret", + SecurityEventEvaluator: func(_ context.Context, identity OIDCSecurityEventIdentity) (OIDCSecurityEventEvaluation, error) { + if identity.Subject != "platform-subject" || identity.IssuedAt.IsZero() { + t.Fatal("security event evaluator received incomplete identity") + } + return evaluation, nil + }, + }) + if err != nil { + t.Fatal(err) + } + raw := signedOIDCToken(t, issuer, "ec-key", jwt.SigningMethodES256, key, nil) + if _, err := verifier.Verify(context.Background(), raw); err != nil || introspectionCalls != 1 { + t.Fatalf("fallback 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) + } + evaluation = OIDCSecurityEventEvaluation{RequireIntrospection: true} + introspectionAvailable = false + _, err = verifier.Verify(context.Background(), raw) + var requestError *RequestAuthError + if !errors.As(err, &requestError) || requestError.Status != http.StatusServiceUnavailable { + t.Fatalf("introspection outage error=%T %v", err, err) + } +} + func signedOIDCToken(t *testing.T, issuer, kid string, method jwt.SigningMethod, key any, mutate func(jwt.MapClaims)) string { t.Helper() now := time.Now() diff --git a/apps/api/internal/config/config.go b/apps/api/internal/config/config.go index 6cd3a36..3819271 100644 --- a/apps/api/internal/config/config.go +++ b/apps/api/internal/config/config.go @@ -8,6 +8,8 @@ import ( "os" "strconv" "strings" + + "github.com/google/uuid" ) const ( @@ -16,50 +18,63 @@ const ( ) type Config struct { - AppEnv string - HTTPAddr string - DatabaseURL string - IdentityMode string - JWTSecret string - ServerMainBaseURL string - ServerMainInternalToken string - ServerMainInternalKey string - ServerMainInternalSecret string - OIDCEnabled bool - OIDCIssuer string - OIDCAudience string - OIDCTenantID string - OIDCRolePrefix string - OIDCRequiredScopes []string - OIDCJWKSCacheTTLSeconds int - OIDCAcceptLegacyHS256 bool - OIDCIntrospectionEnabled bool - OIDCIntrospectionClientID string - OIDCIntrospectionClientSecret string - OIDCJITProvisioningEnabled bool - OIDCGatewayTenantKey string - OIDCBrowserSessionEnabled bool - OIDCSessionCookieSecure bool - OIDCClientID string - OIDCRedirectURI string - OIDCPostLogoutRedirectURI string - OIDCSessionEncryptionKey string - OIDCSessionIdleTTLSeconds int - OIDCSessionAbsoluteTTLSeconds int - OIDCSessionRefreshBeforeSeconds int - PublicBaseURL string - WebBaseURL string - LocalGeneratedStorageDir string - LocalUploadedStorageDir string - LocalTempAssetTTLHours int - TaskProgressCallbackEnabled bool - TaskProgressCallbackURL string - TaskProgressCallbackTimeoutMS string - TaskProgressCallbackMaxAttempts string - CORSAllowedOrigin string - GlobalHTTPProxy string - GlobalHTTPProxySource string - LogLevel slog.Level + AppEnv string + HTTPAddr string + DatabaseURL string + IdentityMode string + JWTSecret string + ServerMainBaseURL string + ServerMainInternalToken string + ServerMainInternalKey string + ServerMainInternalSecret string + OIDCEnabled bool + OIDCIssuer string + OIDCAudience string + OIDCTenantID string + OIDCRolePrefix string + OIDCRequiredScopes []string + OIDCJWKSCacheTTLSeconds int + OIDCAcceptLegacyHS256 bool + 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 + OIDCSecurityEventsHeartbeatIntervalSeconds int + OIDCSecurityEventsStaleAfterSeconds int + OIDCSecurityEventsClockSkewSeconds int + OIDCJITProvisioningEnabled bool + OIDCGatewayTenantKey string + OIDCBrowserSessionEnabled bool + OIDCSessionCookieSecure bool + OIDCClientID string + OIDCRedirectURI string + OIDCPostLogoutRedirectURI string + OIDCSessionEncryptionKey string + OIDCSessionIdleTTLSeconds int + OIDCSessionAbsoluteTTLSeconds int + OIDCSessionRefreshBeforeSeconds int + PublicBaseURL string + WebBaseURL string + LocalGeneratedStorageDir string + LocalUploadedStorageDir string + LocalTempAssetTTLHours int + TaskProgressCallbackEnabled bool + TaskProgressCallbackURL string + TaskProgressCallbackTimeoutMS string + TaskProgressCallbackMaxAttempts string + CORSAllowedOrigin string + GlobalHTTPProxy string + GlobalHTTPProxySource string + LogLevel slog.Level } func Load() Config { @@ -75,23 +90,36 @@ func Load() Config { env("SERVER_MAIN_BASE_URL", "http://localhost:3000"), "/", ), - ServerMainInternalToken: env("SERVER_MAIN_INTERNAL_TOKEN", ""), - 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", ""), "/"), - OIDCAudience: env("OIDC_AUDIENCE", ""), - OIDCTenantID: env("OIDC_TENANT_ID", ""), - OIDCRolePrefix: env("OIDC_ROLE_PREFIX", "gateway."), - OIDCRequiredScopes: splitCSV(env("OIDC_REQUIRED_SCOPES", "gateway.access")), - 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", ""), - OIDCJITProvisioningEnabled: env("OIDC_JIT_PROVISIONING_ENABLED", "false") == "true", - OIDCGatewayTenantKey: env("OIDC_GATEWAY_TENANT_KEY", ""), - OIDCBrowserSessionEnabled: env("OIDC_BROWSER_SESSION_ENABLED", "true") == "true", + ServerMainInternalToken: env("SERVER_MAIN_INTERNAL_TOKEN", ""), + 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", ""), "/"), + OIDCAudience: env("OIDC_AUDIENCE", ""), + OIDCTenantID: env("OIDC_TENANT_ID", ""), + OIDCRolePrefix: env("OIDC_ROLE_PREFIX", "gateway."), + OIDCRequiredScopes: splitCSV(env("OIDC_REQUIRED_SCOPES", "gateway.access")), + 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", ""), + 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), + OIDCJITProvisioningEnabled: env("OIDC_JIT_PROVISIONING_ENABLED", "false") == "true", + OIDCGatewayTenantKey: env("OIDC_GATEWAY_TENANT_KEY", ""), + OIDCBrowserSessionEnabled: env("OIDC_BROWSER_SESSION_ENABLED", "true") == "true", OIDCSessionCookieSecure: env("OIDC_SESSION_COOKIE_SECURE", strconv.FormatBool(!isLocalEnvironment(appEnv)), ) == "true", @@ -157,9 +185,59 @@ 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} { @@ -182,6 +260,25 @@ 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 016192c..6312c7e 100644 --- a/apps/api/internal/config/config_test.go +++ b/apps/api/internal/config/config_test.go @@ -136,3 +136,34 @@ func TestValidateRejectsOfflineAccessForBrowserSession(t *testing.T) { t.Fatalf("Validate() error = %v, want offline_access rejection", err) } } + +func TestSecurityEventsAreOptionalButFailFastWhenPartiallyConfigured(t *testing.T) { + 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") + } +} diff --git a/apps/api/internal/httpapi/security_events.go b/apps/api/internal/httpapi/security_events.go new file mode 100644 index 0000000..bbe48a8 --- /dev/null +++ b/apps/api/internal/httpapi/security_events.go @@ -0,0 +1,22 @@ +package httpapi + +import "net/http" + +// receiveSecurityEvent accepts an RFC 8417 Security Event Token over RFC 8935 Push. +// +// @Summary Receive an SSF Security Event Token +// @Description Optional endpoint. It validates a stream-specific Bearer before parsing and verifying an ES256 SET. A committed event and a duplicate jti both return an empty 202 response. +// @Tags Security Events +// @Accept application/secevent+jwt +// @Produce json +// @Param Authorization header string true "Bearer stream-specific-secret" +// @Param set body string true "Compact signed Security Event Token" +// @Success 202 +// @Failure 400 {object} map[string]string +// @Failure 401 {object} map[string]string +// @Failure 403 {object} map[string]string +// @Failure 503 {object} map[string]string +// @Router /api/v1/security-events/ssf [post] +func (s *Server) receiveSecurityEvent(w http.ResponseWriter, r *http.Request) { + s.securityEventReceiver.ServeHTTP(w, r) +} diff --git a/apps/api/internal/httpapi/server.go b/apps/api/internal/httpapi/server.go index 85c6cd4..2ab5fc1 100644 --- a/apps/api/internal/httpapi/server.go +++ b/apps/api/internal/httpapi/server.go @@ -13,21 +13,23 @@ import ( "github.com/easyai/easyai-ai-gateway/apps/api/internal/config" "github.com/easyai/easyai-ai-gateway/apps/api/internal/oidcsession" "github.com/easyai/easyai-ai-gateway/apps/api/internal/runner" + ssfreceiver "github.com/easyai/easyai-ai-gateway/apps/api/internal/securityevents" "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" ) type Server struct { - ctx context.Context - cfg config.Config - store *store.Store - oidcUserResolver oidcUserResolver - auth *auth.Authenticator - oidcClient oidcPublicClient - oidcSessions oidcSessionManager - oidcSessionCipher *oidcsession.Cipher - runner *runner.Service - logger *slog.Logger - geminiUploadSessions sync.Map + ctx context.Context + cfg config.Config + store *store.Store + oidcUserResolver oidcUserResolver + auth *auth.Authenticator + oidcClient oidcPublicClient + oidcSessions oidcSessionManager + oidcSessionCipher *oidcsession.Cipher + runner *runner.Service + logger *slog.Logger + geminiUploadSessions sync.Map + securityEventReceiver http.Handler } type oidcPublicClient interface { @@ -63,14 +65,58 @@ func NewServerWithContext(ctx context.Context, cfg config.Config, db *store.Stor server.auth.LegacyJWTEnabled = !cfg.OIDCEnabled || cfg.OIDCAcceptLegacyHS256 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 err != nil { + panic("invalid OIDC security event verifier configuration: " + 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, + HeartbeatInterval: time.Duration(cfg.OIDCSecurityEventsHeartbeatIntervalSeconds) * time.Second, + StaleAfter: time.Duration(cfg.OIDCSecurityEventsStaleAfterSeconds) * time.Second, + }) + if err != nil { + panic("invalid OIDC security event heartbeat configuration: " + 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 + } if cfg.OIDCEnabled { + var evaluator func(context.Context, auth.OIDCSecurityEventIdentity) (auth.OIDCSecurityEventEvaluation, error) + if securityEventService != nil { + evaluator = securityEventService.Evaluate + } verifier, err := auth.NewOIDCVerifier(auth.OIDCConfig{ Issuer: cfg.OIDCIssuer, Audience: cfg.OIDCAudience, TenantID: cfg.OIDCTenantID, RolePrefix: cfg.OIDCRolePrefix, RequiredScopes: cfg.OIDCRequiredScopes, - JWKSCacheTTL: time.Duration(cfg.OIDCJWKSCacheTTLSeconds) * time.Second, - IntrospectionEnabled: cfg.OIDCIntrospectionEnabled, - IntrospectionClientID: cfg.OIDCIntrospectionClientID, - IntrospectionClientSecret: cfg.OIDCIntrospectionClientSecret, + JWKSCacheTTL: time.Duration(cfg.OIDCJWKSCacheTTLSeconds) * time.Second, + IntrospectionEnabled: cfg.OIDCIntrospectionEnabled, + IntrospectionClientID: cfg.OIDCIntrospectionClientID, + IntrospectionClientSecret: cfg.OIDCIntrospectionClientSecret, + SecurityEventEvaluator: evaluator, + IntrospectionObserver: securityEventMetrics.ObserveIntrospection, + JWKSRefreshFailureObserver: func() { securityEventMetrics.ObserveJWKSRefreshFailure("oidc") }, }) if err != nil { panic("invalid OIDC configuration: " + err.Error()) @@ -118,6 +164,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.HandleFunc("GET /static/simulation/{asset}", serveSimulationAsset) mux.HandleFunc("GET /static/generated/{asset}", server.serveGeneratedStaticAsset) mux.HandleFunc("GET /static/uploaded/{asset}", server.serveUploadedStaticAsset) @@ -128,6 +175,9 @@ 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.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))) @@ -284,6 +334,8 @@ func oidcSessionRequestError(err error) error { return auth.NewRequestAuthError(http.StatusServiceUnavailable, "OIDC_SESSION_REFRESH_UNAVAILABLE", "认证中心暂时不可用，请稍后重试") case errors.Is(err, oidcsession.ErrSessionStoreUnavailable): return auth.NewRequestAuthError(http.StatusServiceUnavailable, "OIDC_SESSION_STORE_UNAVAILABLE", "登录会话存储暂时不可用") + case errors.Is(err, oidcsession.ErrSecurityStateUnavailable): + return auth.NewRequestAuthError(http.StatusServiceUnavailable, "OIDC_SECURITY_EVENT_STATE_UNAVAILABLE", "认证撤销状态暂时不可用") case errors.Is(err, oidcsession.ErrGatewayUserDisabled): return auth.NewRequestAuthError(http.StatusForbidden, "GATEWAY_USER_DISABLED", "该 Gateway 账号已停用，请联系管理员") default: diff --git a/apps/api/internal/oidcsession/service.go b/apps/api/internal/oidcsession/service.go index ebb0d84..bc33f14 100644 --- a/apps/api/internal/oidcsession/service.go +++ b/apps/api/internal/oidcsession/service.go @@ -20,6 +20,7 @@ var ( ErrSessionRefreshUnavailable = errors.New("OIDC session refresh is unavailable") ErrSessionStoreUnavailable = errors.New("OIDC session store is unavailable") ErrGatewayUserDisabled = errors.New("gateway user is disabled") + ErrSecurityStateUnavailable = errors.New("OIDC security event state is unavailable") ) type Repository interface { @@ -179,11 +180,17 @@ func (s *Service) refresh(ctx context.Context, hash []byte, record store.OIDCSes } return user, nil } + if errors.Is(verifyErr, ErrSecurityStateUnavailable) { + return nil, verifyErr + } } return nil, ErrSessionRefreshUnavailable } user, err := s.verifySessionUser(ctx, record, refreshed.AccessToken) if err != nil { + if errors.Is(err, ErrSecurityStateUnavailable) { + return nil, err + } _ = s.repository.DeleteOIDCSessionByID(ctx, record.ID) return nil, ErrSessionExpired } @@ -241,12 +248,19 @@ func (s *Service) waitForRefresh(ctx context.Context, hash []byte, original stor } return user, nil } + if errors.Is(err, ErrSecurityStateUnavailable) { + return nil, err + } } return nil, ErrSessionRefreshUnavailable } func (s *Service) verifySessionUser(ctx context.Context, record store.OIDCSession, accessToken string) (*auth.User, error) { user, err := s.verifier.Verify(ctx, accessToken) + var requestError *auth.RequestAuthError + if errors.As(err, &requestError) && requestError.Status == 503 { + return nil, ErrSecurityStateUnavailable + } if err != nil || user == nil || user.Source != "oidc" || user.ID != record.ExternalUserID { return nil, ErrSessionInvalid } diff --git a/apps/api/internal/securityevents/handler.go b/apps/api/internal/securityevents/handler.go new file mode 100644 index 0000000..8c9f7b0 --- /dev/null +++ b/apps/api/internal/securityevents/handler.go @@ -0,0 +1,185 @@ +package securityevents + +import ( + "context" + "crypto/sha256" + "crypto/subtle" + "encoding/json" + "errors" + "io" + "mime" + "net/http" + "strings" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +type Repository interface { + ApplySessionRevoked(context.Context, store.ApplySessionRevokedInput) (store.ApplySecurityEventResult, error) + ApplySecurityEventStreamUpdated(context.Context, string, string, string, string, string, string, time.Time) (bool, error) + ConfirmSecurityEventVerification(context.Context, string, string, string, string, []byte, time.Time) (bool, error) + RecordSecurityEventReceipt(context.Context, string, string, string, string, string) (bool, error) +} + +type Handler struct { + verifier *Verifier + repository Repository + issuer string + audience string + streamID string + bearerSecret string + nextSecret string + clock func() time.Time + metrics *Metrics +} + +func (h *Handler) SetMetrics(metrics *Metrics) { h.metrics = metrics } + +func NewHandler(verifier *Verifier, repository Repository, issuer, audience, streamID, bearerSecret, nextSecret string) (*Handler, error) { + if verifier == nil || repository == nil || issuer == "" || audience == "" || streamID == "" || len(bearerSecret) < 16 { + return nil, errors.New("SSF receiver dependencies and bearer secret are required") + } + return &Handler{verifier: verifier, repository: repository, issuer: issuer, audience: audience, streamID: streamID, bearerSecret: bearerSecret, nextSecret: nextSecret, clock: time.Now}, nil +} + +func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + started := h.clock() + outcome := "rejected" + var sessionsDeleted int64 + defer func() { + if h.metrics != nil { + h.metrics.ObserveReceipt(outcome, h.clock().Sub(started), sessionsDeleted) + } + }() + w.Header().Set("Cache-Control", "no-store") + if r.Method != http.MethodPost { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + if !h.validAuthorization(r.Header.Get("Authorization")) { + w.Header().Set("WWW-Authenticate", `Bearer realm="ssf-receiver"`) + writeSETError(w, http.StatusUnauthorized, "authentication_failed") + return + } + mediaType, _, err := mime.ParseMediaType(r.Header.Get("Content-Type")) + if err != nil || mediaType != "application/secevent+jwt" || r.Header.Get("Content-Encoding") != "" { + writeSETError(w, http.StatusBadRequest, "invalid_request") + return + } + r.Body = http.MaxBytesReader(w, r.Body, 64*1024) + payload, err := io.ReadAll(r.Body) + if err != nil || len(payload) == 0 { + writeSETError(w, http.StatusBadRequest, "invalid_request") + return + } + event, err := h.verifier.Verify(r.Context(), string(payload)) + if err != nil { + var protocol *protocolError + if errors.As(err, &protocol) { + writeSETError(w, http.StatusBadRequest, protocol.Code) + } else { + writeSETError(w, http.StatusBadRequest, "invalid_request") + } + return + } + switch event.EventType { + case SessionRevokedEventType: + var result store.ApplySecurityEventResult + result, err = h.repository.ApplySessionRevoked(r.Context(), store.ApplySessionRevokedInput{ + Issuer: event.Issuer, Audience: event.Audience, JTI: event.JTI, TransactionID: event.TransactionID, + SubjectIssuer: event.SubjectIssuer, TenantID: event.TenantID, Subject: event.Subject, EventTimestamp: event.EventTimestamp, + InitiatingEntity: event.InitiatingEntity, + }) + if err == nil { + sessionsDeleted = result.SessionsDeleted + if result.Duplicate { + outcome = "duplicate" + } else { + outcome = "accepted" + } + } + case VerificationEventType: + var confirmed bool + if event.State == "" { + confirmed, err = h.repository.RecordSecurityEventReceipt(r.Context(), h.issuer, h.audience, event.JTI, VerificationEventType, h.streamID) + } else { + stateHash := sha256.Sum256([]byte(event.State)) + confirmed, err = h.repository.ConfirmSecurityEventVerification(r.Context(), h.issuer, h.audience, h.streamID, event.JTI, stateHash[:], h.clock().UTC()) + } + if err != nil && strings.Contains(err.Error(), "invalid verification state") { + writeSETError(w, http.StatusBadRequest, "invalid_request") + return + } + if err == nil { + if confirmed { + outcome = "accepted" + if h.metrics != nil { + h.metrics.ObserveVerificationAccepted() + } + } else { + outcome = "duplicate" + } + } + case StreamUpdatedEventType: + var inserted bool + inserted, err = h.repository.ApplySecurityEventStreamUpdated( + r.Context(), h.issuer, h.audience, event.JTI, h.streamID, event.StreamStatus, event.Reason, h.clock().UTC(), + ) + if err == nil { + if inserted { + outcome = "accepted" + } else { + outcome = "duplicate" + } + } + } + if err != nil { + w.Header().Set("Retry-After", "1") + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusAccepted) +} + +func (h *Handler) validAuthorization(value string) bool { + if constantBearerCompare(value, h.bearerSecret) { + return true + } + return h.nextSecret != "" && constantBearerCompare(value, h.nextSecret) +} + +func constantBearerCompare(header, secret string) bool { + providedHash := sha256.Sum256([]byte(header)) + expectedHash := sha256.Sum256([]byte("Bearer " + secret)) + return subtle.ConstantTimeCompare(providedHash[:], expectedHash[:]) == 1 +} + +func writeSETError(w http.ResponseWriter, status int, code string) { + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Content-Language", "en") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(map[string]string{ + "err": code, + "description": setErrorDescription(code), + }) +} + +func setErrorDescription(code string) string { + switch code { + case "invalid_request": + return "The SET request is malformed or contains invalid claims." + case "invalid_key": + return "The SET signing key or signature is invalid." + case "invalid_issuer": + return "The SET issuer is not accepted by this receiver." + case "invalid_audience": + return "The SET audience does not identify this receiver." + case "authentication_failed": + return "Receiver authentication failed." + case "access_denied": + return "The SET is not permitted for this receiver." + default: + return "The SET request could not be processed." + } +} diff --git a/apps/api/internal/securityevents/handler_test.go b/apps/api/internal/securityevents/handler_test.go new file mode 100644 index 0000000..b8dd894 --- /dev/null +++ b/apps/api/internal/securityevents/handler_test.go @@ -0,0 +1,269 @@ +package securityevents + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "encoding/base64" + "encoding/json" + "math/big" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" + "github.com/golang-jwt/jwt/v5" + "github.com/google/uuid" +) + +type fakeRepository struct { + input *store.ApplySessionRevokedInput + dedup map[string]struct{} + confirmed bool +} + +func (f *fakeRepository) ApplySessionRevoked(_ context.Context, input store.ApplySessionRevokedInput) (store.ApplySecurityEventResult, error) { + if f.dedup == nil { + f.dedup = make(map[string]struct{}) + } + if _, exists := f.dedup[input.JTI]; exists { + return store.ApplySecurityEventResult{Duplicate: true}, nil + } + f.dedup[input.JTI] = struct{}{} + f.input = &input + return store.ApplySecurityEventResult{WatermarkMoved: true, SessionsDeleted: 2}, nil +} + +func (f *fakeRepository) ConfirmSecurityEventVerification(context.Context, string, string, string, string, []byte, time.Time) (bool, error) { + f.confirmed = true + return true, nil +} + +func (*fakeRepository) RecordSecurityEventReceipt(context.Context, string, string, string, string, string) (bool, error) { + return true, nil +} + +func (*fakeRepository) ApplySecurityEventStreamUpdated(context.Context, string, string, string, string, string, string, time.Time) (bool, error) { + return true, nil +} + +func TestReceiverAcceptsValidSessionRevokedAndDuplicate(t *testing.T) { + privateKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + var transmitter *httptest.Server + 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{"issuer": transmitter.URL + "/ssf", "jwks_uri": transmitter.URL + "/ssf/jwks.json"}) + case "/ssf/jwks.json": + _ = json.NewEncoder(w).Encode(map[string]any{"keys": []any{map[string]any{ + "kty": "EC", "kid": "current", "use": "sig", "alg": "ES256", "crv": "P-256", + "x": coordinate(privateKey.X), "y": coordinate(privateKey.Y), + }}}) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer transmitter.Close() + tenantID, subjectID, applicationID, streamID := uuid.NewString(), uuid.NewString(), uuid.NewString(), uuid.NewString() + audience := "urn:easyai:ssf:receiver:" + applicationID + verifier, err := NewVerifier(VerifierConfig{ + TransmitterIssuer: transmitter.URL + "/ssf", Audience: audience, + SubjectIssuer: "https://auth.example/issuer/tenant-a", TenantID: tenantID, StreamID: streamID, + ClockSkew: time.Minute, + }) + if err != nil { + t.Fatal(err) + } + repository := &fakeRepository{} + handler, err := NewHandler(verifier, repository, transmitter.URL+"/ssf", audience, streamID, "receiver-bearer-secret", "next-receiver-secret") + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC().Truncate(time.Second) + jti := uuid.NewString() + set := signSET(t, privateKey, map[string]any{ + "iss": transmitter.URL + "/ssf", "aud": audience, "iat": now.Unix(), "jti": jti, "txn": uuid.NewString(), + "sub_id": map[string]any{ + "format": "complex", + "user": map[string]any{"format": "iss_sub", "iss": "https://auth.example/issuer/tenant-a", "sub": subjectID}, + "tenant": map[string]any{"format": "opaque", "id": tenantID}, + }, + "events": map[string]any{SessionRevokedEventType: map[string]any{"event_timestamp": now.Unix(), "initiating_entity": "admin"}}, + }) + for _, secret := range []string{"receiver-bearer-secret", "next-receiver-secret"} { + response := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/api/v1/security-events/ssf", strings.NewReader(set)) + request.Header.Set("Authorization", "Bearer "+secret) + request.Header.Set("Content-Type", "application/secevent+jwt") + handler.ServeHTTP(response, request) + if response.Code != http.StatusAccepted || response.Body.Len() != 0 { + t.Fatalf("status=%d body=%s", response.Code, response.Body.String()) + } + } + if repository.input == nil || repository.input.Subject != subjectID || repository.input.TenantID != tenantID || + repository.input.SubjectIssuer != "https://auth.example/issuer/tenant-a" { + t.Fatalf("applied event = %#v", repository.input) + } +} + +func TestReceiverRejectsAuthenticationMediaTypeAndForbiddenClaims(t *testing.T) { + privateKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + verifier := &Verifier{config: VerifierConfig{ + TransmitterIssuer: "https://auth.example/ssf", Audience: "urn:easyai:ssf:receiver:app", + SubjectIssuer: "https://auth.example/issuer/tenant", TenantID: uuid.NewString(), StreamID: uuid.NewString(), ClockSkew: time.Minute, + }, keys: map[string]*ecdsa.PublicKey{"current": &privateKey.PublicKey}, expiresAt: time.Now().Add(time.Hour), client: http.DefaultClient} + handler, _ := NewHandler(verifier, &fakeRepository{}, "https://auth.example/ssf", verifier.config.Audience, verifier.config.StreamID, "receiver-bearer-secret", "") + + response := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/events", strings.NewReader("token")) + request.Header.Set("Content-Type", "application/secevent+jwt") + handler.ServeHTTP(response, request) + if response.Code != http.StatusUnauthorized { + t.Fatalf("missing bearer status=%d", response.Code) + } + assertSETError(t, response, "authentication_failed") + + set := signSET(t, privateKey, map[string]any{ + "iss": verifier.config.TransmitterIssuer, "aud": verifier.config.Audience, "iat": time.Now().Unix(), + "jti": uuid.NewString(), "sub": "forbidden", "exp": time.Now().Add(time.Hour).Unix(), "events": map[string]any{}, + }) + response = httptest.NewRecorder() + request = httptest.NewRequest(http.MethodPost, "/events", strings.NewReader(set)) + request.Header.Set("Authorization", "Bearer receiver-bearer-secret") + request.Header.Set("Content-Type", "application/json") + handler.ServeHTTP(response, request) + if response.Code != http.StatusBadRequest { + t.Fatalf("wrong media status=%d", response.Code) + } + assertSETError(t, response, "invalid_request") + + response = httptest.NewRecorder() + request = httptest.NewRequest(http.MethodPost, "/events", strings.NewReader(set)) + request.Header.Set("Authorization", "Bearer receiver-bearer-secret") + request.Header.Set("Content-Type", "application/secevent+jwt") + handler.ServeHTTP(response, request) + if response.Code != http.StatusBadRequest || !strings.Contains(response.Body.String(), "invalid_request") { + t.Fatalf("forbidden claims status=%d body=%s", response.Code, response.Body.String()) + } + + future := time.Now().Add(10 * time.Minute) + futureSET := signSET(t, privateKey, map[string]any{ + "iss": verifier.config.TransmitterIssuer, "aud": verifier.config.Audience, "iat": time.Now().Unix(), + "jti": uuid.NewString(), "txn": uuid.NewString(), + "sub_id": map[string]any{ + "format": "complex", + "user": map[string]any{"format": "iss_sub", "iss": verifier.config.SubjectIssuer, "sub": uuid.NewString()}, + "tenant": map[string]any{"format": "opaque", "id": verifier.config.TenantID}, + }, + "events": map[string]any{SessionRevokedEventType: map[string]any{ + "event_timestamp": future.Unix(), "initiating_entity": "admin", + }}, + }) + response = httptest.NewRecorder() + request = httptest.NewRequest(http.MethodPost, "/events", strings.NewReader(futureSET)) + request.Header.Set("Authorization", "Bearer receiver-bearer-secret") + request.Header.Set("Content-Type", "application/secevent+jwt") + handler.ServeHTTP(response, request) + if response.Code != http.StatusBadRequest { + t.Fatalf("future event timestamp status=%d", response.Code) + } +} + +func TestReceiverReturnsRFC8935IssuerAudienceAndKeyErrors(t *testing.T) { + trustedKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + untrustedKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + verifier := &Verifier{config: VerifierConfig{ + TransmitterIssuer: "https://auth.example/ssf", Audience: "urn:easyai:ssf:receiver:app", + SubjectIssuer: "https://auth.example/issuer/tenant", TenantID: uuid.NewString(), StreamID: uuid.NewString(), ClockSkew: time.Minute, + }, keys: map[string]*ecdsa.PublicKey{"current": &trustedKey.PublicKey}, expiresAt: time.Now().Add(time.Hour), client: http.DefaultClient} + handler, _ := NewHandler(verifier, &fakeRepository{}, verifier.config.TransmitterIssuer, verifier.config.Audience, verifier.config.StreamID, "receiver-bearer-secret", "") + + baseClaims := jwt.MapClaims{ + "iss": verifier.config.TransmitterIssuer, "aud": verifier.config.Audience, "iat": time.Now().Unix(), + "jti": uuid.NewString(), "events": map[string]any{}, + } + tests := []struct { + name string + key *ecdsa.PrivateKey + mutate func(jwt.MapClaims) + code string + }{ + {name: "issuer", key: trustedKey, mutate: func(claims jwt.MapClaims) { claims["iss"] = "https://wrong.example/ssf" }, code: "invalid_issuer"}, + {name: "audience", key: trustedKey, mutate: func(claims jwt.MapClaims) { claims["aud"] = "urn:wrong" }, code: "invalid_audience"}, + {name: "signature", key: untrustedKey, mutate: func(jwt.MapClaims) {}, code: "invalid_key"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + claims := jwt.MapClaims{} + for key, value := range baseClaims { + claims[key] = value + } + test.mutate(claims) + response := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/events", strings.NewReader(signSET(t, test.key, claims))) + request.Header.Set("Authorization", "Bearer receiver-bearer-secret") + request.Header.Set("Content-Type", "application/secevent+jwt") + handler.ServeHTTP(response, request) + if response.Code != http.StatusBadRequest { + t.Fatalf("status=%d body=%s", response.Code, response.Body.String()) + } + assertSETError(t, response, test.code) + }) + } +} + +func TestReceiverAcceptsStreamUpdatedEvent(t *testing.T) { + privateKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + verifier := &Verifier{config: VerifierConfig{ + TransmitterIssuer: "https://auth.example/ssf", Audience: "urn:easyai:ssf:receiver:app", + SubjectIssuer: "https://auth.example/issuer/tenant", TenantID: uuid.NewString(), StreamID: uuid.NewString(), ClockSkew: time.Minute, + }, keys: map[string]*ecdsa.PublicKey{"current": &privateKey.PublicKey}, expiresAt: time.Now().Add(time.Hour), client: http.DefaultClient} + handler, _ := NewHandler(verifier, &fakeRepository{}, verifier.config.TransmitterIssuer, verifier.config.Audience, verifier.config.StreamID, "receiver-bearer-secret", "") + set := signSET(t, privateKey, jwt.MapClaims{ + "iss": verifier.config.TransmitterIssuer, "aud": verifier.config.Audience, "iat": time.Now().Unix(), "jti": uuid.NewString(), + "sub_id": map[string]any{"format": "opaque", "id": verifier.config.StreamID}, + "events": map[string]any{StreamUpdatedEventType: map[string]any{"status": "paused", "reason": "maintenance"}}, + }) + response := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/events", strings.NewReader(set)) + request.Header.Set("Authorization", "Bearer receiver-bearer-secret") + request.Header.Set("Content-Type", "application/secevent+jwt") + handler.ServeHTTP(response, request) + if response.Code != http.StatusAccepted || response.Body.Len() != 0 { + t.Fatalf("status=%d body=%s", response.Code, response.Body.String()) + } +} + +func assertSETError(t *testing.T, response *httptest.ResponseRecorder, code string) { + t.Helper() + if response.Header().Get("Content-Language") != "en" { + t.Fatalf("Content-Language=%q", response.Header().Get("Content-Language")) + } + var payload map[string]string + if err := json.Unmarshal(response.Body.Bytes(), &payload); err != nil { + t.Fatalf("invalid error body: %v", err) + } + if payload["err"] != code || payload["description"] == "" { + t.Fatalf("error payload=%#v", payload) + } +} + +func signSET(t *testing.T, key *ecdsa.PrivateKey, claims jwt.MapClaims) string { + t.Helper() + token := jwt.NewWithClaims(jwt.SigningMethodES256, claims) + token.Header["typ"], token.Header["kid"] = "secevent+jwt", "current" + raw, err := token.SignedString(key) + if err != nil { + t.Fatal(err) + } + return raw +} + +func coordinate(value *big.Int) string { + payload := make([]byte, 32) + value.FillBytes(payload) + return base64.RawURLEncoding.EncodeToString(payload) +} diff --git a/apps/api/internal/securityevents/metrics.go b/apps/api/internal/securityevents/metrics.go new file mode 100644 index 0000000..39a5fc3 --- /dev/null +++ b/apps/api/internal/securityevents/metrics.go @@ -0,0 +1,168 @@ +package securityevents + +import ( + "context" + "fmt" + "net/http" + "sync/atomic" + "time" +) + +type MetricsSnapshotProvider interface { + SecurityEventMetrics(context.Context, string, string) (string, *time.Time, error) +} + +type Metrics struct { + accepted atomic.Uint64 + rejected atomic.Uint64 + duplicate atomic.Uint64 + sessionsDeleted atomic.Uint64 + watermarkRejected atomic.Uint64 + verificationAccepted atomic.Uint64 + heartbeatAccepted atomic.Uint64 + heartbeatFailed atomic.Uint64 + introspectionActive atomic.Uint64 + introspectionInactive atomic.Uint64 + introspectionFailed atomic.Uint64 + jwksSSFFailed atomic.Uint64 + jwksOIDCFailed atomic.Uint64 + processingCount atomic.Uint64 + processingNanos atomic.Uint64 + processingBuckets [6]atomic.Uint64 +} + +var processingDurationBounds = [...]time.Duration{ + 10 * time.Millisecond, 50 * time.Millisecond, 100 * time.Millisecond, + 500 * time.Millisecond, time.Second, 3 * time.Second, +} + +func (m *Metrics) ObserveReceipt(outcome string, duration time.Duration, sessionsDeleted int64) { + switch outcome { + case "accepted": + m.accepted.Add(1) + case "duplicate": + m.duplicate.Add(1) + default: + m.rejected.Add(1) + } + if sessionsDeleted > 0 { + m.sessionsDeleted.Add(uint64(sessionsDeleted)) + } + if duration < 0 { + duration = 0 + } + m.processingCount.Add(1) + m.processingNanos.Add(uint64(duration)) + for index, bound := range processingDurationBounds { + if duration <= bound { + m.processingBuckets[index].Add(1) + } + } +} + +func (m *Metrics) ObserveWatermarkRejection() { m.watermarkRejected.Add(1) } + +func (m *Metrics) ObserveVerificationAccepted() { m.verificationAccepted.Add(1) } + +func (m *Metrics) ObserveHeartbeat(outcome string) { + if outcome == "accepted" { + m.heartbeatAccepted.Add(1) + return + } + m.heartbeatFailed.Add(1) +} + +func (m *Metrics) ObserveIntrospection(outcome string) { + switch outcome { + case "active": + m.introspectionActive.Add(1) + case "inactive": + m.introspectionInactive.Add(1) + default: + m.introspectionFailed.Add(1) + } +} + +func (m *Metrics) ObserveJWKSRefreshFailure(source string) { + if source == "ssf" { + m.jwksSSFFailed.Add(1) + return + } + m.jwksOIDCFailed.Add(1) +} + +func (m *Metrics) Handler(provider MetricsSnapshotProvider, issuer, audience string, enabled bool) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mode := "disabled" + var lastVerificationAt *time.Time + if enabled { + ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second) + defer cancel() + var err error + mode, lastVerificationAt, err = provider.SecurityEventMetrics(ctx, issuer, audience) + if err != nil { + http.Error(w, "metrics unavailable", http.StatusServiceUnavailable) + return + } + } + w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + outcomeCounters(w, "easyai_gateway_ssf_receipts_total", "Received SETs by bounded outcome.", []outcomeValue{ + {"accepted", m.accepted.Load()}, {"rejected", m.rejected.Load()}, {"duplicate", m.duplicate.Load()}, + }) + plainCounter(w, "easyai_gateway_ssf_sessions_deleted_total", "OIDC browser sessions deleted by accepted session-revoked events.", m.sessionsDeleted.Load()) + plainCounter(w, "easyai_gateway_ssf_watermark_rejections_total", "OIDC access tokens rejected by a local revocation watermark.", m.watermarkRejected.Load()) + plainCounter(w, "easyai_gateway_ssf_verifications_total", "Valid verification SETs accepted by the receiver.", m.verificationAccepted.Load()) + outcomeCounters(w, "easyai_gateway_ssf_heartbeat_requests_total", "Verification requests by bounded outcome.", []outcomeValue{ + {"accepted", m.heartbeatAccepted.Load()}, {"failed", m.heartbeatFailed.Load()}, + }) + outcomeCounters(w, "easyai_gateway_oidc_introspection_total", "OIDC introspection requests by bounded outcome.", []outcomeValue{ + {"active", m.introspectionActive.Load()}, {"inactive", m.introspectionInactive.Load()}, {"failed", m.introspectionFailed.Load()}, + }) + outcomeCounters(w, "easyai_gateway_jwks_refresh_failures_total", "JWKS refresh failures by bounded source.", []outcomeValue{ + {"ssf", m.jwksSSFFailed.Load()}, {"oidc", m.jwksOIDCFailed.Load()}, + }) + fmt.Fprintln(w, "# HELP easyai_gateway_ssf_processing_duration_seconds SSF receiver transaction processing time.") + fmt.Fprintln(w, "# TYPE easyai_gateway_ssf_processing_duration_seconds histogram") + for index, bound := range processingDurationBounds { + fmt.Fprintf(w, "easyai_gateway_ssf_processing_duration_seconds_bucket{le=\"%.2f\"} %d\n", bound.Seconds(), m.processingBuckets[index].Load()) + } + fmt.Fprintf(w, "easyai_gateway_ssf_processing_duration_seconds_bucket{le=\"+Inf\"} %d\n", m.processingCount.Load()) + fmt.Fprintf(w, "easyai_gateway_ssf_processing_duration_seconds_sum %.6f\n", float64(m.processingNanos.Load())/float64(time.Second)) + fmt.Fprintf(w, "easyai_gateway_ssf_processing_duration_seconds_count %d\n", m.processingCount.Load()) + verificationAge := float64(0) + if lastVerificationAt != nil { + verificationAge = time.Since(*lastVerificationAt).Seconds() + if verificationAge < 0 { + verificationAge = 0 + } + } + fmt.Fprintln(w, "# HELP easyai_gateway_ssf_verification_age_seconds Seconds since the last matched verification SET.") + fmt.Fprintln(w, "# TYPE easyai_gateway_ssf_verification_age_seconds gauge") + fmt.Fprintf(w, "easyai_gateway_ssf_verification_age_seconds %.6f\n", verificationAge) + fmt.Fprintln(w, "# HELP easyai_gateway_ssf_mode Current SSF receiver mode as a one-hot gauge.") + fmt.Fprintln(w, "# TYPE easyai_gateway_ssf_mode gauge") + for _, modeName := range []string{"disabled", "bootstrap", "push_healthy", "introspection_fallback"} { + value := 0 + if mode == modeName { + value = 1 + } + fmt.Fprintf(w, "easyai_gateway_ssf_mode{mode=\"%s\"} %d\n", modeName, value) + } + }) +} + +type outcomeValue struct { + name string + value uint64 +} + +func outcomeCounters(w http.ResponseWriter, name, help string, values []outcomeValue) { + fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s counter\n", name, help, name) + for _, value := range values { + fmt.Fprintf(w, "%s{outcome=\"%s\"} %d\n", name, value.name, value.value) + } +} + +func plainCounter(w http.ResponseWriter, name, help string, value uint64) { + fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s counter\n%s %d\n", name, help, name, name, value) +} diff --git a/apps/api/internal/securityevents/metrics_test.go b/apps/api/internal/securityevents/metrics_test.go new file mode 100644 index 0000000..1ba6d30 --- /dev/null +++ b/apps/api/internal/securityevents/metrics_test.go @@ -0,0 +1,50 @@ +package securityevents + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +type metricsSnapshot struct { + mode string + last time.Time +} + +func (m metricsSnapshot) SecurityEventMetrics(context.Context, string, string) (string, *time.Time, error) { + return m.mode, &m.last, nil +} + +func TestMetricsExposeBoundedOutcomesAndState(t *testing.T) { + metrics := &Metrics{} + metrics.ObserveReceipt("accepted", 20*time.Millisecond, 2) + metrics.ObserveReceipt("duplicate", 10*time.Millisecond, 0) + metrics.ObserveWatermarkRejection() + metrics.ObserveHeartbeat("accepted") + metrics.ObserveIntrospection("failed") + metrics.ObserveJWKSRefreshFailure("ssf") + + recorder := httptest.NewRecorder() + metrics.Handler(metricsSnapshot{mode: "push_healthy", last: time.Now().Add(-time.Minute)}, "issuer", "audience", true). + ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + if recorder.Code != http.StatusOK { + t.Fatalf("metrics status=%d", recorder.Code) + } + for _, expected := range []string{ + `easyai_gateway_ssf_receipts_total{outcome="accepted"} 1`, + `easyai_gateway_ssf_receipts_total{outcome="duplicate"} 1`, + `easyai_gateway_ssf_sessions_deleted_total 2`, + `easyai_gateway_ssf_watermark_rejections_total 1`, + `easyai_gateway_ssf_processing_duration_seconds_bucket{le="0.05"} 2`, + `easyai_gateway_ssf_mode{mode="push_healthy"} 1`, + `easyai_gateway_oidc_introspection_total{outcome="failed"} 1`, + `easyai_gateway_jwks_refresh_failures_total{outcome="ssf"} 1`, + } { + if !strings.Contains(recorder.Body.String(), expected) { + t.Fatalf("missing metric %q in:\n%s", expected, recorder.Body.String()) + } + } +} diff --git a/apps/api/internal/securityevents/service.go b/apps/api/internal/securityevents/service.go new file mode 100644 index 0000000..6814161 --- /dev/null +++ b/apps/api/internal/securityevents/service.go @@ -0,0 +1,211 @@ +package securityevents + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/sha256" + "crypto/tls" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "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" +) + +type StateRepository interface { + EnsureSecurityEventStreamState(context.Context, string, string, string) error + BeginSecurityEventVerification(context.Context, string, string, []byte, time.Time) error + RecordSecurityEventHeartbeatFailure(context.Context, string, string, string) error + AdvanceSecurityEventStreamState(context.Context, string, string, time.Time, time.Duration) error + EvaluateOIDCSecurityEvent(context.Context, string, string, string, string, string, time.Time, time.Time, time.Duration) (store.SecurityEventEvaluation, error) +} + +type ServiceConfig struct { + Issuer string + Audience string + StreamID string + ManagementTokenURL string + ManagementClientID string + ManagementSecret string + HeartbeatInterval time.Duration + StaleAfter time.Duration + HTTPClient *http.Client +} + +type Service struct { + repository StateRepository + config ServiceConfig + client *http.Client + clock func() time.Time + tokenMu sync.Mutex + token string + tokenUntil time.Time + metrics *Metrics +} + +func (s *Service) SetMetrics(metrics *Metrics) { s.metrics = metrics } + +func NewService(repository StateRepository, config ServiceConfig) (*Service, error) { + if repository == nil || config.Issuer == "" || config.Audience == "" || config.StreamID == "" || + config.ManagementTokenURL == "" || config.ManagementClientID == "" || config.ManagementSecret == "" { + return nil, errors.New("SSF heartbeat configuration is incomplete") + } + if config.HeartbeatInterval <= 0 || config.StaleAfter < 2*config.HeartbeatInterval { + return nil, errors.New("SSF heartbeat timing is invalid") + } + client := config.HTTPClient + if client == nil { + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12} + transport.DisableCompression = true + client = &http.Client{Timeout: 10 * time.Second, + Transport: transport, + CheckRedirect: func(_ *http.Request, _ []*http.Request) error { return http.ErrUseLastResponse }} + } + return &Service{repository: repository, config: config, client: client, clock: time.Now}, nil +} + +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 { + return err + } + go s.run(ctx) + return nil +} + +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, + identity.IssuedAt, s.clock().UTC(), s.config.StaleAfter, + ) + if err != nil { + return auth.OIDCSecurityEventEvaluation{}, err + } + if result.Revoked && s.metrics != nil { + s.metrics.ObserveWatermarkRejection() + } + return auth.OIDCSecurityEventEvaluation{Revoked: result.Revoked, RequireIntrospection: result.RequireIntrospection}, nil +} + +func (s *Service) run(ctx context.Context) { + s.sendVerification(ctx) + ticker := time.NewTicker(s.config.HeartbeatInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + s.sendVerification(ctx) + } + } +} + +func (s *Service) sendVerification(ctx context.Context) { + outcome := "failed" + defer func() { + if s.metrics != nil { + s.metrics.ObserveHeartbeat(outcome) + } + }() + if err := s.repository.AdvanceSecurityEventStreamState(ctx, s.config.Issuer, s.config.Audience, s.clock().UTC(), s.config.StaleAfter); err != nil { + _ = s.repository.RecordSecurityEventHeartbeatFailure(ctx, s.config.Issuer, s.config.Audience, "state_advance_failed") + } + state, err := randomVerificationState() + if err != nil { + _ = s.repository.RecordSecurityEventHeartbeatFailure(ctx, s.config.Issuer, s.config.Audience, "state_generation_failed") + return + } + hash := sha256.Sum256([]byte(state)) + now := s.clock().UTC() + if err := s.repository.BeginSecurityEventVerification(ctx, s.config.Issuer, s.config.Audience, hash[:], now); err != nil { + _ = s.repository.RecordSecurityEventHeartbeatFailure(ctx, s.config.Issuer, s.config.Audience, "state_persistence_failed") + return + } + token, err := s.managementToken(ctx) + if err != nil { + _ = s.repository.RecordSecurityEventHeartbeatFailure(ctx, s.config.Issuer, s.config.Audience, "management_token_failed") + return + } + payload, _ := json.Marshal(map[string]string{"stream_id": s.config.StreamID, "state": state}) + request, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(s.config.Issuer, "/")+"/v1/verify", bytes.NewReader(payload)) + if err != nil { + return + } + request.Header.Set("Authorization", "Bearer "+token) + request.Header.Set("Content-Type", "application/json") + response, err := s.client.Do(request) + if err != nil { + _ = s.repository.RecordSecurityEventHeartbeatFailure(ctx, s.config.Issuer, s.config.Audience, "verification_request_failed") + return + } + defer response.Body.Close() + _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 64*1024)) + if response.StatusCode != http.StatusNoContent { + _ = s.repository.RecordSecurityEventHeartbeatFailure(ctx, s.config.Issuer, s.config.Audience, fmt.Sprintf("verification_http_%d", response.StatusCode)) + return + } + outcome = "accepted" +} + +func (s *Service) managementToken(ctx context.Context) (string, error) { + s.tokenMu.Lock() + defer s.tokenMu.Unlock() + if s.token != "" && s.clock().Before(s.tokenUntil) { + return s.token, nil + } + form := url.Values{"grant_type": {"client_credentials"}, "scope": {"ssf.stream.read ssf.stream.verify"}} + request, err := http.NewRequestWithContext(ctx, http.MethodPost, s.config.ManagementTokenURL, strings.NewReader(form.Encode())) + if err != nil { + return "", err + } + request.SetBasicAuth(s.config.ManagementClientID, s.config.ManagementSecret) + request.Header.Set("Content-Type", "application/x-www-form-urlencoded") + response, err := s.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("management token endpoint returned HTTP %d", response.StatusCode) + } + payload, err := io.ReadAll(io.LimitReader(response.Body, 64*1024+1)) + if err != nil || len(payload) > 64*1024 { + return "", errors.New("management token response is too large") + } + var result struct { + AccessToken string `json:"access_token"` + TokenType string `json:"token_type"` + ExpiresIn int64 `json:"expires_in"` + } + if json.Unmarshal(payload, &result) != nil || result.AccessToken == "" || !strings.EqualFold(result.TokenType, "Bearer") { + return "", errors.New("management token response is invalid") + } + if result.ExpiresIn <= 0 { + result.ExpiresIn = 60 + } + cacheFor := time.Duration(result.ExpiresIn) * time.Second + if cacheFor > 30*time.Second { + cacheFor -= 30 * time.Second + } + s.token, s.tokenUntil = result.AccessToken, s.clock().Add(cacheFor) + return s.token, nil +} + +func randomVerificationState() (string, error) { + payload := make([]byte, 32) + if _, err := rand.Read(payload); err != nil { + return "", err + } + return base64.RawURLEncoding.EncodeToString(payload), nil +} diff --git a/apps/api/internal/securityevents/service_test.go b/apps/api/internal/securityevents/service_test.go new file mode 100644 index 0000000..d7512c5 --- /dev/null +++ b/apps/api/internal/securityevents/service_test.go @@ -0,0 +1,86 @@ +package securityevents + +import ( + "context" + "crypto/sha256" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +type fakeStateRepository struct { + mutex sync.Mutex + stateHash []byte + verification chan struct{} + evaluation store.SecurityEventEvaluation +} + +func (*fakeStateRepository) EnsureSecurityEventStreamState(context.Context, string, string, string) error { + return nil +} +func (f *fakeStateRepository) BeginSecurityEventVerification(_ context.Context, _, _ string, hash []byte, _ time.Time) error { + f.mutex.Lock() + f.stateHash = append([]byte(nil), hash...) + f.mutex.Unlock() + return nil +} +func (*fakeStateRepository) RecordSecurityEventHeartbeatFailure(context.Context, string, string, string) error { + return nil +} +func (*fakeStateRepository) AdvanceSecurityEventStreamState(context.Context, string, string, time.Time, time.Duration) error { + return nil +} +func (f *fakeStateRepository) EvaluateOIDCSecurityEvent(context.Context, string, string, string, string, string, time.Time, time.Time, time.Duration) (store.SecurityEventEvaluation, error) { + return f.evaluation, nil +} + +func TestServiceRequestsVerificationWithMachineTokenAndPersistsStateFirst(t *testing.T) { + repository := &fakeStateRepository{verification: make(chan struct{}, 1)} + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/token": + clientID, secret, ok := r.BasicAuth() + if !ok || clientID != "gateway-ssf" || secret != "management-secret" { + t.Fatal("machine client authentication is missing") + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"access_token":"management-token","token_type":"Bearer","expires_in":300}`)) + case "/ssf/v1/verify": + if r.Header.Get("Authorization") != "Bearer management-token" { + t.Fatal("verification bearer is missing") + } + repository.mutex.Lock() + persisted := len(repository.stateHash) == sha256.Size + repository.mutex.Unlock() + if !persisted { + t.Fatal("verification state was not persisted before request") + } + w.WriteHeader(http.StatusNoContent) + repository.verification <- struct{}{} + } + })) + defer server.Close() + service, err := NewService(repository, ServiceConfig{ + Issuer: server.URL + "/ssf", Audience: "urn:easyai:ssf:receiver:app", StreamID: "00000000-0000-4000-8000-000000000001", + ManagementTokenURL: server.URL + "/token", ManagementClientID: "gateway-ssf", ManagementSecret: "management-secret", + HeartbeatInterval: time.Hour, StaleAfter: 2 * time.Hour, HTTPClient: server.Client(), + }) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if err := service.Start(ctx); err != nil { + t.Fatal(err) + } + select { + case <-repository.verification: + case <-time.After(2 * time.Second): + t.Fatal("verification was not requested") + } +} diff --git a/apps/api/internal/securityevents/verifier.go b/apps/api/internal/securityevents/verifier.go new file mode 100644 index 0000000..415c5ac --- /dev/null +++ b/apps/api/internal/securityevents/verifier.go @@ -0,0 +1,412 @@ +package securityevents + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/tls" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "math/big" + "net/http" + "net/url" + "strings" + "sync" + "time" + + "github.com/golang-jwt/jwt/v5" + "github.com/google/uuid" +) + +const ( + SessionRevokedEventType = "https://schemas.openid.net/secevent/caep/event-type/session-revoked" + VerificationEventType = "https://schemas.openid.net/secevent/ssf/event-type/verification" + StreamUpdatedEventType = "https://schemas.openid.net/secevent/ssf/event-type/stream-updated" +) + +type Event struct { + Issuer string + Audience string + JTI string + TransactionID string + IssuedAt time.Time + EventType string + EventTimestamp time.Time + SubjectIssuer string + Subject string + TenantID string + InitiatingEntity string + StreamID string + State string + StreamStatus string + Reason string +} + +type VerifierConfig struct { + TransmitterIssuer string + Audience string + SubjectIssuer string + TenantID string + StreamID string + ClockSkew time.Duration + HTTPClient *http.Client +} + +type Verifier struct { + config VerifierConfig + client *http.Client + mu sync.Mutex + keys map[string]*ecdsa.PublicKey + expiresAt time.Time + metrics *Metrics + clock func() time.Time +} + +func (v *Verifier) SetMetrics(metrics *Metrics) { v.metrics = metrics } + +type transmitterMetadata struct { + Issuer string `json:"issuer"` + JWKSURI string `json:"jwks_uri"` +} + +type jwkSet struct { + Keys []struct { + KID string `json:"kid"` + KTY string `json:"kty"` + Use string `json:"use"` + Alg string `json:"alg"` + Crv string `json:"crv"` + X string `json:"x"` + Y string `json:"y"` + } `json:"keys"` +} + +type protocolError struct { + Code string +} + +func (e *protocolError) Error() string { return e.Code } + +func NewVerifier(config VerifierConfig) (*Verifier, error) { + config.TransmitterIssuer = strings.TrimRight(strings.TrimSpace(config.TransmitterIssuer), "/") + config.SubjectIssuer = strings.TrimRight(strings.TrimSpace(config.SubjectIssuer), "/") + if !absoluteHTTPURL(config.TransmitterIssuer) || !absoluteHTTPURL(config.SubjectIssuer) || config.Audience == "" || config.TenantID == "" { + return nil, errors.New("SSF transmitter, audience, subject issuer, and tenant are required") + } + if _, err := uuid.Parse(config.StreamID); err != nil { + return nil, errors.New("SSF stream id must be a UUID") + } + if config.ClockSkew < 0 || config.ClockSkew > 5*time.Minute { + return nil, errors.New("SSF clock skew is invalid") + } + client := config.HTTPClient + if client == nil { + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12} + transport.DisableCompression = true + client = &http.Client{Timeout: 10 * time.Second, + Transport: transport, + CheckRedirect: func(_ *http.Request, _ []*http.Request) error { return http.ErrUseLastResponse }} + } + return &Verifier{config: config, client: client, keys: make(map[string]*ecdsa.PublicKey), clock: time.Now}, nil +} + +func (v *Verifier) Verify(ctx context.Context, raw string) (Event, error) { + parser := jwt.NewParser(jwt.WithValidMethods([]string{"ES256"})) + unverified, _, err := parser.ParseUnverified(raw, jwt.MapClaims{}) + if err != nil || unverified == nil || unverified.Header["typ"] != "secevent+jwt" || unverified.Header["alg"] != "ES256" { + return Event{}, &protocolError{Code: "invalid_request"} + } + unverifiedClaims, ok := unverified.Claims.(jwt.MapClaims) + if !ok { + return Event{}, &protocolError{Code: "invalid_request"} + } + if stringClaim(unverifiedClaims, "iss") != v.config.TransmitterIssuer { + return Event{}, &protocolError{Code: "invalid_issuer"} + } + if !claimHasAudience(unverifiedClaims["aud"], v.config.Audience) { + return Event{}, &protocolError{Code: "invalid_audience"} + } + kid, _ := unverified.Header["kid"].(string) + if kid == "" { + return Event{}, &protocolError{Code: "invalid_key"} + } + key, err := v.key(ctx, kid) + if err != nil { + return Event{}, &protocolError{Code: "invalid_key"} + } + token, err := jwt.Parse(raw, func(token *jwt.Token) (any, error) { return key, nil }, + jwt.WithValidMethods([]string{"ES256"}), jwt.WithIssuer(v.config.TransmitterIssuer), + jwt.WithAudience(v.config.Audience), jwt.WithIssuedAt(), jwt.WithLeeway(v.config.ClockSkew)) + if err != nil || !token.Valid { + if errors.Is(err, jwt.ErrTokenSignatureInvalid) { + return Event{}, &protocolError{Code: "invalid_key"} + } + return Event{}, &protocolError{Code: "invalid_request"} + } + claims, ok := token.Claims.(jwt.MapClaims) + if !ok { + return Event{}, &protocolError{Code: "invalid_request"} + } + if _, present := claims["sub"]; present { + return Event{}, &protocolError{Code: "invalid_request"} + } + if _, present := claims["exp"]; present { + return Event{}, &protocolError{Code: "invalid_request"} + } + jti := stringClaim(claims, "jti") + if _, err := uuid.Parse(jti); err != nil { + return Event{}, &protocolError{Code: "invalid_request"} + } + issuedAt, err := claims.GetIssuedAt() + if err != nil || issuedAt == nil { + return Event{}, &protocolError{Code: "invalid_request"} + } + events, ok := claims["events"].(map[string]any) + if !ok || len(events) != 1 { + return Event{}, &protocolError{Code: "invalid_request"} + } + event := Event{Issuer: v.config.TransmitterIssuer, Audience: v.config.Audience, JTI: jti, TransactionID: stringClaim(claims, "txn"), IssuedAt: issuedAt.Time} + if payload, present := events[SessionRevokedEventType]; present { + value, ok := payload.(map[string]any) + if !ok || event.TransactionID == "" { + return Event{}, &protocolError{Code: "invalid_request"} + } + if _, err := uuid.Parse(event.TransactionID); err != nil { + return Event{}, &protocolError{Code: "invalid_request"} + } + event.EventType, event.EventTimestamp, event.InitiatingEntity = SessionRevokedEventType, unixClaim(value["event_timestamp"]), stringValue(value["initiating_entity"]) + if event.EventTimestamp.IsZero() || event.EventTimestamp.After(v.now().Add(v.config.ClockSkew)) || + !validInitiator(event.InitiatingEntity) || !v.readSessionSubject(claims["sub_id"], &event) { + return Event{}, &protocolError{Code: "invalid_request"} + } + return event, nil + } + if payload, present := events[VerificationEventType]; present { + value, ok := payload.(map[string]any) + if !ok || !v.readStreamSubject(claims["sub_id"], &event) { + return Event{}, &protocolError{Code: "invalid_request"} + } + event.EventType = VerificationEventType + if rawState, present := value["state"]; present { + var valid bool + event.State, valid = rawState.(string) + if !valid || len(event.State) > 1024 { + return Event{}, &protocolError{Code: "invalid_request"} + } + } + if len(event.State) > 1024 { + return Event{}, &protocolError{Code: "invalid_request"} + } + return event, nil + } + if payload, present := events[StreamUpdatedEventType]; present { + value, ok := payload.(map[string]any) + if !ok || !v.readStreamSubject(claims["sub_id"], &event) { + return Event{}, &protocolError{Code: "invalid_request"} + } + event.EventType = StreamUpdatedEventType + event.StreamStatus, event.Reason = stringValue(value["status"]), stringValue(value["reason"]) + if event.StreamStatus != "enabled" && event.StreamStatus != "paused" && event.StreamStatus != "disabled" { + return Event{}, &protocolError{Code: "invalid_request"} + } + if len(event.Reason) > 1024 { + return Event{}, &protocolError{Code: "invalid_request"} + } + return event, nil + } + return Event{}, &protocolError{Code: "invalid_request"} +} + +func claimHasAudience(raw any, expected string) bool { + switch value := raw.(type) { + case string: + return value == expected + case []string: + for _, audience := range value { + if audience == expected { + return true + } + } + case []any: + for _, item := range value { + if audience, ok := item.(string); ok && audience == expected { + return true + } + } + } + return false +} + +func (v *Verifier) now() time.Time { + if v.clock != nil { + return v.clock().UTC() + } + return time.Now().UTC() +} + +func (v *Verifier) readSessionSubject(raw any, event *Event) bool { + subject, ok := raw.(map[string]any) + if !ok || stringValue(subject["format"]) != "complex" { + return false + } + user, userOK := subject["user"].(map[string]any) + tenant, tenantOK := subject["tenant"].(map[string]any) + if !userOK || !tenantOK || stringValue(user["format"]) != "iss_sub" || stringValue(tenant["format"]) != "opaque" { + return false + } + event.SubjectIssuer, event.Subject, event.TenantID = strings.TrimRight(stringValue(user["iss"]), "/"), stringValue(user["sub"]), stringValue(tenant["id"]) + if event.SubjectIssuer != v.config.SubjectIssuer || event.TenantID != v.config.TenantID { + return false + } + _, err := uuid.Parse(event.Subject) + return err == nil +} + +func (v *Verifier) readStreamSubject(raw any, event *Event) bool { + subject, ok := raw.(map[string]any) + if !ok || stringValue(subject["format"]) != "opaque" || stringValue(subject["id"]) != v.config.StreamID { + return false + } + event.StreamID = v.config.StreamID + return true +} + +func (v *Verifier) key(ctx context.Context, kid string) (*ecdsa.PublicKey, error) { + v.mu.Lock() + if key := v.keys[kid]; key != nil && time.Now().Before(v.expiresAt) { + v.mu.Unlock() + return key, nil + } + v.mu.Unlock() + if err := v.refresh(ctx); err != nil { + if v.metrics != nil { + v.metrics.ObserveJWKSRefreshFailure("ssf") + } + return nil, err + } + v.mu.Lock() + defer v.mu.Unlock() + key := v.keys[kid] + if key == nil { + return nil, errors.New("SSF signing key is unavailable") + } + return key, nil +} + +func (v *Verifier) refresh(ctx context.Context) error { + v.mu.Lock() + defer v.mu.Unlock() + configurationURL, err := ssfConfigurationURL(v.config.TransmitterIssuer) + if err != nil { + return err + } + var metadata transmitterMetadata + if err := v.fetchJSON(ctx, configurationURL, &metadata); err != nil { + return err + } + if strings.TrimRight(metadata.Issuer, "/") != v.config.TransmitterIssuer || !sameOrigin(metadata.JWKSURI, v.config.TransmitterIssuer) { + return errors.New("SSF metadata is not bound to configured issuer") + } + var set jwkSet + if err := v.fetchJSON(ctx, metadata.JWKSURI, &set); err != nil { + return err + } + keys := make(map[string]*ecdsa.PublicKey) + for _, item := range set.Keys { + if item.KID == "" || item.KTY != "EC" || item.Alg != "ES256" || item.Crv != "P-256" || item.Use != "sig" { + continue + } + x, xErr := decodeCoordinate(item.X) + y, yErr := decodeCoordinate(item.Y) + if xErr == nil && yErr == nil && elliptic.P256().IsOnCurve(x, y) { + keys[item.KID] = &ecdsa.PublicKey{Curve: elliptic.P256(), X: x, Y: y} + } + } + if len(keys) == 0 { + return errors.New("SSF JWKS contains no ES256 keys") + } + v.keys, v.expiresAt = keys, time.Now().Add(5*time.Minute) + return nil +} + +func (v *Verifier) fetchJSON(ctx context.Context, endpoint string, target any) error { + request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return err + } + request.Header.Set("Accept", "application/json") + response, err := v.client.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + return fmt.Errorf("SSF endpoint returned HTTP %d", response.StatusCode) + } + payload, err := io.ReadAll(io.LimitReader(response.Body, (1<<20)+1)) + if err != nil || len(payload) > 1<<20 { + return errors.New("SSF endpoint response is too large") + } + if err := json.Unmarshal(payload, target); err != nil { + return errors.New("SSF endpoint returned invalid JSON") + } + return nil +} + +func ssfConfigurationURL(issuer string) (string, error) { + parsed, err := url.Parse(issuer) + if err != nil || parsed.Scheme == "" || parsed.Host == "" { + return "", errors.New("SSF issuer is invalid") + } + path := strings.TrimSuffix(parsed.EscapedPath(), "/") + parsed.Path, parsed.RawPath = "/.well-known/ssf-configuration"+path, "" + parsed.RawQuery, parsed.Fragment = "", "" + return parsed.String(), nil +} + +func sameOrigin(candidate, issuer string) bool { + candidateURL, candidateErr := url.Parse(candidate) + issuerURL, issuerErr := url.Parse(issuer) + return candidateErr == nil && issuerErr == nil && candidateURL.Scheme == issuerURL.Scheme && strings.EqualFold(candidateURL.Host, issuerURL.Host) && candidateURL.User == nil && candidateURL.Fragment == "" +} + +func absoluteHTTPURL(raw string) bool { + parsed, err := url.Parse(raw) + return err == nil && parsed.IsAbs() && parsed.Hostname() != "" && (parsed.Scheme == "https" || parsed.Scheme == "http") && parsed.User == nil && parsed.Fragment == "" +} + +func decodeCoordinate(raw string) (*big.Int, error) { + value, err := base64.RawURLEncoding.DecodeString(raw) + if err != nil || len(value) != 32 { + return nil, errors.New("invalid EC coordinate") + } + return new(big.Int).SetBytes(value), nil +} + +func stringClaim(claims jwt.MapClaims, name string) string { return stringValue(claims[name]) } + +func stringValue(value any) string { + result, _ := value.(string) + return result +} + +func unixClaim(value any) time.Time { + switch typed := value.(type) { + case float64: + return time.Unix(int64(typed), 0).UTC() + case json.Number: + integer, err := typed.Int64() + if err == nil { + return time.Unix(integer, 0).UTC() + } + } + return time.Time{} +} + +func validInitiator(value string) bool { + return value == "admin" || value == "user" || value == "policy" || value == "system" +} diff --git a/apps/api/internal/store/security_events.go b/apps/api/internal/store/security_events.go new file mode 100644 index 0000000..5ae18ed --- /dev/null +++ b/apps/api/internal/store/security_events.go @@ -0,0 +1,341 @@ +package store + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "time" + + "github.com/jackc/pgx/v5" +) + +type ApplySessionRevokedInput struct { + Issuer string + Audience string + JTI string + TransactionID string + SubjectIssuer string + TenantID string + Subject string + EventTimestamp time.Time + InitiatingEntity string +} + +type ApplySecurityEventResult struct { + Duplicate bool + WatermarkMoved bool + SessionsDeleted int64 + AuditID string +} + +type SecurityEventEvaluation struct { + Revoked bool + RequireIntrospection bool + Mode string +} + +func (s *Store) ApplySessionRevoked(ctx context.Context, input ApplySessionRevokedInput) (ApplySecurityEventResult, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return ApplySecurityEventResult{}, err + } + defer func() { _ = tx.Rollback(ctx) }() + subjectHash := shortSecurityEventHash(input.Subject) + tag, err := tx.Exec(ctx, ` +INSERT INTO gateway_security_event_receipts ( + issuer, jti, audience, event_type, transaction_id, tenant_id, subject_hash, event_timestamp +) +VALUES ($1, $2::uuid, $3, $4, NULLIF($5, ''), $6, $7, $8) +ON CONFLICT (issuer, jti) DO NOTHING`, + input.Issuer, input.JTI, input.Audience, + "https://schemas.openid.net/secevent/caep/event-type/session-revoked", + input.TransactionID, input.TenantID, subjectHash, input.EventTimestamp, + ) + if err != nil { + return ApplySecurityEventResult{}, err + } + if tag.RowsAffected() == 0 { + if err := tx.Commit(ctx); err != nil { + return ApplySecurityEventResult{}, err + } + return ApplySecurityEventResult{Duplicate: true}, nil + } + + var revokedAt time.Time + err = tx.QueryRow(ctx, ` +INSERT INTO gateway_oidc_revocation_watermarks (issuer, tenant_id, subject, revoked_at, source_jti) +VALUES ($1, $2, $3, $4, $5::uuid) +ON CONFLICT (issuer, tenant_id, subject) DO UPDATE +SET revoked_at = EXCLUDED.revoked_at, source_jti = EXCLUDED.source_jti, updated_at = now() +WHERE EXCLUDED.revoked_at > gateway_oidc_revocation_watermarks.revoked_at +RETURNING revoked_at`, input.SubjectIssuer, input.TenantID, input.Subject, input.EventTimestamp, input.JTI).Scan(&revokedAt) + watermarkMoved := err == nil + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + return ApplySecurityEventResult{}, err + } + var sessionsDeleted int64 + if watermarkMoved { + tag, err = tx.Exec(ctx, ` +DELETE FROM gateway_oidc_sessions session +USING gateway_users gateway_user +WHERE session.gateway_user_id = gateway_user.id + AND gateway_user.source = 'oidc' + AND gateway_user.external_user_id = $1 + AND gateway_user.tenant_id = $2`, input.Subject, input.TenantID) + if err != nil { + return ApplySecurityEventResult{}, err + } + sessionsDeleted = tag.RowsAffected() + } + jtiHash := shortSecurityEventHash(input.JTI) + audit, err := s.RecordAuditLogTx(ctx, tx, AuditLogInput{ + Category: "identity", Action: "identity.oidc_session.revoked", ActorSource: "ssf", + TargetType: "oidc_subject", TargetID: subjectHash, + AfterState: map[string]any{"watermarkMoved": watermarkMoved, "sessionsDeleted": sessionsDeleted}, + Metadata: map[string]any{ + "issuer": input.SubjectIssuer, "transmitterIssuer": input.Issuer, "tenantId": input.TenantID, "jtiHash": jtiHash, + "initiatingEntity": input.InitiatingEntity, + }, + }) + if err != nil { + return ApplySecurityEventResult{}, err + } + if err := tx.Commit(ctx); err != nil { + return ApplySecurityEventResult{}, err + } + return ApplySecurityEventResult{WatermarkMoved: watermarkMoved, SessionsDeleted: sessionsDeleted, AuditID: audit.ID}, nil +} + +func (s *Store) EnsureSecurityEventStreamState(ctx context.Context, issuer, audience, streamID string) error { + tag, err := s.pool.Exec(ctx, ` +INSERT INTO gateway_security_event_stream_state(issuer,audience,stream_id,mode) +VALUES($1,$2,$3::uuid,'bootstrap') +ON CONFLICT(issuer,audience) DO UPDATE +SET stream_id=EXCLUDED.stream_id,updated_at=now() +WHERE gateway_security_event_stream_state.stream_id=EXCLUDED.stream_id`, issuer, audience, streamID) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return errors.New("security event stream id cannot be changed") + } + return nil +} + +func (s *Store) BeginSecurityEventVerification(ctx context.Context, issuer, audience string, stateHash []byte, now time.Time) error { + if len(stateHash) != sha256.Size { + return errors.New("verification state hash must be SHA-256") + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + tag, err := tx.Exec(ctx, ` +UPDATE gateway_security_event_stream_state +SET pending_state_hash=$3,pending_state_created_at=$4,updated_at=now() +WHERE issuer=$1 AND audience=$2`, issuer, audience, stateHash, now) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return errors.New("security event stream state is missing") + } + if _, err := tx.Exec(ctx, ` +INSERT INTO gateway_security_event_verification_challenges(issuer,audience,state_hash,created_at,expires_at) +VALUES($1,$2,$3,$4,$4 + interval '180 seconds') +ON CONFLICT(issuer,audience,state_hash) DO NOTHING`, issuer, audience, stateHash, now); err != nil { + return err + } + if _, err := tx.Exec(ctx, `DELETE FROM gateway_security_event_verification_challenges WHERE expires_at < $1`, now); err != nil { + return err + } + return tx.Commit(ctx) +} + +func (s *Store) ConfirmSecurityEventVerification(ctx context.Context, issuer, audience, streamID, jti string, stateHash []byte, now time.Time) (bool, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return false, err + } + defer func() { _ = tx.Rollback(ctx) }() + tag, err := tx.Exec(ctx, ` +INSERT INTO gateway_security_event_receipts(issuer,jti,audience,event_type,subject_hash) +VALUES($1,$2::uuid,$3,$4,$5) +ON CONFLICT(issuer,jti) DO NOTHING`, issuer, jti, audience, + "https://schemas.openid.net/secevent/ssf/event-type/verification", shortSecurityEventHash(streamID)) + if err != nil { + return false, err + } + if tag.RowsAffected() == 0 { + return false, tx.Commit(ctx) + } + tag, err = tx.Exec(ctx, ` +UPDATE gateway_security_event_stream_state +SET last_verification_at=$5, + bootstrap_until=CASE WHEN last_verification_at IS NULL THEN $5 + interval '360 seconds' ELSE bootstrap_until END, + consecutive_verifications=CASE + WHEN stream_status <> 'enabled' THEN 0 + WHEN mode='introspection_fallback' THEN consecutive_verifications+1 + ELSE 1 + END, + mode=CASE + WHEN stream_status <> 'enabled' THEN mode + WHEN mode='introspection_fallback' AND consecutive_verifications+1 >= 2 + AND (bootstrap_until IS NULL OR $5 >= bootstrap_until) THEN 'push_healthy' + WHEN mode='introspection_fallback' THEN 'introspection_fallback' + WHEN mode='push_healthy' THEN 'push_healthy' + ELSE 'bootstrap' + END, + fallback_since=CASE + WHEN stream_status='enabled' AND mode='introspection_fallback' AND consecutive_verifications+1 >= 2 + AND (bootstrap_until IS NULL OR $5 >= bootstrap_until) THEN NULL + ELSE fallback_since + END, + pending_state_hash=NULL,pending_state_created_at=NULL, + last_error_category=CASE WHEN stream_status='enabled' THEN NULL ELSE 'stream_' || stream_status END, + updated_at=now() +WHERE issuer=$1 AND audience=$2 AND stream_id=$3::uuid + AND EXISTS(SELECT 1 FROM gateway_security_event_verification_challenges challenge + WHERE challenge.issuer=$1 AND challenge.audience=$2 AND challenge.state_hash=$4 AND challenge.expires_at >= $5)`, + issuer, audience, streamID, stateHash, now) + if err != nil { + return false, err + } + if tag.RowsAffected() == 0 { + return false, errors.New("invalid verification state") + } + if _, err := tx.Exec(ctx, `DELETE FROM gateway_security_event_verification_challenges + WHERE issuer=$1 AND audience=$2 AND state_hash=$3`, issuer, audience, stateHash); err != nil { + return false, err + } + return true, tx.Commit(ctx) +} + +func (s *Store) RecordSecurityEventReceipt(ctx context.Context, issuer, audience, jti, eventType, subject string) (bool, error) { + tag, err := s.pool.Exec(ctx, ` +INSERT INTO gateway_security_event_receipts(issuer,jti,audience,event_type,subject_hash) +VALUES($1,$2::uuid,$3,$4,$5) +ON CONFLICT(issuer,jti) DO NOTHING`, issuer, jti, audience, eventType, shortSecurityEventHash(subject)) + return tag.RowsAffected() == 1, err +} + +// ApplySecurityEventStreamUpdated records a transmitter-originated stream +// status change and immediately moves OIDC traffic to introspection. Push is +// trusted again only after the normal consecutive-verification recovery path. +func (s *Store) ApplySecurityEventStreamUpdated(ctx context.Context, issuer, audience, jti, streamID, status, reason string, now time.Time) (bool, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return false, err + } + defer func() { _ = tx.Rollback(ctx) }() + tag, err := tx.Exec(ctx, ` +INSERT INTO gateway_security_event_receipts(issuer,jti,audience,event_type,subject_hash) +VALUES($1,$2::uuid,$3,$4,$5) +ON CONFLICT(issuer,jti) DO NOTHING`, issuer, jti, audience, + "https://schemas.openid.net/secevent/ssf/event-type/stream-updated", shortSecurityEventHash(streamID)) + if err != nil { + return false, err + } + if tag.RowsAffected() == 0 { + return false, tx.Commit(ctx) + } + tag, err = tx.Exec(ctx, ` +UPDATE gateway_security_event_stream_state +SET stream_status=$5,mode='introspection_fallback',fallback_since=COALESCE(fallback_since,$4), + consecutive_verifications=0,last_error_category=$6,updated_at=now() +WHERE issuer=$1 AND audience=$2 AND stream_id=$3::uuid`, issuer, audience, streamID, now, status, "stream_"+status) + if err != nil { + return false, err + } + if tag.RowsAffected() == 0 { + return false, errors.New("security event stream state is missing") + } + if _, err := s.RecordAuditLogTx(ctx, tx, AuditLogInput{ + Category: "identity", Action: "identity.ssf_stream.updated", ActorSource: "ssf", + TargetType: "ssf_stream", TargetID: shortSecurityEventHash(streamID), + AfterState: map[string]any{"status": status}, + Metadata: map[string]any{"issuer": issuer, "audience": audience, "reasonCategory": shortSecurityEventHash(reason)}, + }); err != nil { + return false, err + } + return true, tx.Commit(ctx) +} + +func (s *Store) RecordSecurityEventHeartbeatFailure(ctx context.Context, issuer, audience, category string) error { + _, err := s.pool.Exec(ctx, `UPDATE gateway_security_event_stream_state + SET last_error_category=$3,updated_at=now() WHERE issuer=$1 AND audience=$2`, issuer, audience, category) + return err +} + +func (s *Store) AdvanceSecurityEventStreamState(ctx context.Context, issuer, audience string, now time.Time, staleAfter time.Duration) error { + if staleAfter <= 0 { + return errors.New("security event stale threshold must be positive") + } + _, err := s.pool.Exec(ctx, ` +UPDATE gateway_security_event_stream_state +SET mode='introspection_fallback',fallback_since=COALESCE(fallback_since,$3), + consecutive_verifications=0,last_error_category='verification_stale',updated_at=now() +WHERE issuer=$1 AND audience=$2 AND mode <> 'introspection_fallback' + AND COALESCE(last_verification_at,created_at) < $3 - make_interval(secs => $4::double precision)`, + issuer, audience, now, staleAfter.Seconds()) + return err +} + +func (s *Store) EvaluateOIDCSecurityEvent(ctx context.Context, transmitterIssuer, audience, subjectIssuer, tenantID, subject string, issuedAt, now time.Time, staleAfter time.Duration) (SecurityEventEvaluation, error) { + var mode, streamStatus string + var lastVerification, bootstrapUntil *time.Time + var createdAt time.Time + var revokedAt *time.Time + err := s.pool.QueryRow(ctx, ` +SELECT state.mode,state.stream_status,state.last_verification_at,state.bootstrap_until,state.created_at,watermark.revoked_at +FROM gateway_security_event_stream_state state +LEFT JOIN gateway_oidc_revocation_watermarks watermark + ON watermark.issuer=$3 AND watermark.tenant_id=$4 AND watermark.subject=$5 +WHERE state.issuer=$1 AND state.audience=$2`, transmitterIssuer, audience, subjectIssuer, tenantID, subject).Scan( + &mode, &streamStatus, &lastVerification, &bootstrapUntil, &createdAt, &revokedAt, + ) + if err != nil { + return SecurityEventEvaluation{}, err + } + evaluation := SecurityEventEvaluation{Mode: mode, Revoked: revokedAt != nil && !issuedAt.After(*revokedAt)} + stale := (lastVerification != nil && now.Sub(*lastVerification) > staleAfter) || + (lastVerification == nil && (mode != "bootstrap" || now.Sub(createdAt) > staleAfter)) + if stale { + evaluation.Mode, evaluation.RequireIntrospection = "introspection_fallback", true + _, err := s.pool.Exec(ctx, ` +UPDATE gateway_security_event_stream_state +SET mode='introspection_fallback',fallback_since=COALESCE(fallback_since,$3),consecutive_verifications=0, + last_error_category='verification_stale',updated_at=now() +WHERE issuer=$1 AND audience=$2 AND mode <> 'introspection_fallback'`, transmitterIssuer, audience, now) + return evaluation, err + } + if mode == "bootstrap" { + if streamStatus != "enabled" || bootstrapUntil == nil || now.Before(*bootstrapUntil) { + evaluation.RequireIntrospection = true + return evaluation, nil + } + evaluation.Mode = "push_healthy" + _, err := s.pool.Exec(ctx, `UPDATE gateway_security_event_stream_state SET mode='push_healthy',updated_at=now() +WHERE issuer=$1 AND audience=$2 AND mode='bootstrap'`, transmitterIssuer, audience) + return evaluation, err + } + evaluation.RequireIntrospection = mode == "introspection_fallback" + return evaluation, nil +} + +func (s *Store) SecurityEventMetrics(ctx context.Context, issuer, audience string) (string, *time.Time, error) { + var mode string + var lastVerificationAt *time.Time + err := s.pool.QueryRow(ctx, ` +SELECT mode,last_verification_at +FROM gateway_security_event_stream_state +WHERE issuer=$1 AND audience=$2`, issuer, audience).Scan(&mode, &lastVerificationAt) + return mode, lastVerificationAt, err +} + +func shortSecurityEventHash(value string) string { + digest := sha256.Sum256([]byte(value)) + return hex.EncodeToString(digest[:])[:16] +} diff --git a/apps/api/internal/store/security_events_integration_test.go b/apps/api/internal/store/security_events_integration_test.go new file mode 100644 index 0000000..f2d9952 --- /dev/null +++ b/apps/api/internal/store/security_events_integration_test.go @@ -0,0 +1,132 @@ +package store + +import ( + "context" + "crypto/sha256" + "os" + "strings" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestSecurityEventWatermarkVerificationAndFallbackLifecycle(t *testing.T) { + databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL")) + if databaseURL == "" { + t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run security event PostgreSQL integration tests") + } + ctx := context.Background() + probe, err := pgxpool.New(ctx, databaseURL) + if err != nil { + t.Fatal(err) + } + var databaseName string + if err := probe.QueryRow(ctx, `SELECT current_database()`).Scan(&databaseName); err != nil { + probe.Close() + t.Fatal(err) + } + probe.Close() + if !strings.Contains(strings.ToLower(databaseName), "test") { + t.Fatalf("refusing to migrate non-test database %q", databaseName) + } + applyOIDCJITTestMigrations(t, ctx, databaseURL) + db, err := Connect(ctx, databaseURL) + if err != nil { + t.Fatal(err) + } + defer db.Close() + issuer := "https://auth.test.example/ssf" + subjectIssuer := "https://auth.test/realms/tenant" + audience := "urn:easyai:ssf:receiver:" + uuid.NewString() + streamID, tenantID, subject := uuid.NewString(), uuid.NewString(), uuid.NewString() + t.Cleanup(func() { + _, _ = db.pool.Exec(context.Background(), `DELETE FROM gateway_security_event_receipts WHERE issuer=$1`, issuer) + _, _ = db.pool.Exec(context.Background(), `DELETE FROM gateway_oidc_revocation_watermarks WHERE issuer=$1`, subjectIssuer) + _, _ = db.pool.Exec(context.Background(), `DELETE FROM gateway_security_event_stream_state WHERE issuer=$1`, issuer) + _, _ = db.pool.Exec(context.Background(), `DELETE FROM gateway_audit_logs WHERE actor_source='ssf' AND target_id=ANY($1)`, []string{shortSecurityEventHash(subject), shortSecurityEventHash(streamID)}) + }) + if err := db.EnsureSecurityEventStreamState(ctx, issuer, audience, streamID); err != nil { + t.Fatal(err) + } + now := time.Now().UTC().Truncate(time.Second) + evaluation, err := db.EvaluateOIDCSecurityEvent(ctx, issuer, audience, subjectIssuer, tenantID, subject, now, now, 180*time.Second) + if err != nil || !evaluation.RequireIntrospection || evaluation.Mode != "bootstrap" { + t.Fatalf("bootstrap evaluation=%#v error=%v", evaluation, err) + } + + confirmAt := func(label string, at time.Time) { + t.Helper() + stateHash := sha256.Sum256([]byte(label)) + if err := db.BeginSecurityEventVerification(ctx, issuer, audience, stateHash[:], at); err != nil { + t.Fatal(err) + } + matched, err := db.ConfirmSecurityEventVerification(ctx, issuer, audience, streamID, uuid.NewString(), stateHash[:], at) + if err != nil || !matched { + t.Fatalf("confirm matched=%v error=%v", matched, err) + } + } + inserted, err := db.ApplySecurityEventStreamUpdated(ctx, issuer, audience, uuid.NewString(), streamID, "enabled", "onboarding", now) + if err != nil || !inserted { + t.Fatalf("enable stream inserted=%v error=%v", inserted, err) + } + confirmAt("first-verification-state", now) + confirmAt("second-verification-state", now.Add(time.Minute)) + evaluation, err = db.EvaluateOIDCSecurityEvent(ctx, issuer, audience, subjectIssuer, tenantID, subject, now, now.Add(time.Minute), 180*time.Second) + if err != nil || !evaluation.RequireIntrospection || evaluation.Mode != "introspection_fallback" { + t.Fatalf("bootstrap overlap evaluation=%#v error=%v", evaluation, err) + } + confirmAt("post-bootstrap-verification-state", now.Add(361*time.Second)) + evaluation, err = db.EvaluateOIDCSecurityEvent(ctx, issuer, audience, subjectIssuer, tenantID, subject, now, now.Add(361*time.Second), 180*time.Second) + if err != nil || evaluation.RequireIntrospection || evaluation.Mode != "push_healthy" { + t.Fatalf("healthy evaluation=%#v error=%v", evaluation, err) + } + streamUpdateJTI := uuid.NewString() + inserted, err = db.ApplySecurityEventStreamUpdated(ctx, issuer, audience, streamUpdateJTI, streamID, "paused", "maintenance", now.Add(362*time.Second)) + if err != nil || !inserted { + t.Fatalf("stream update inserted=%v error=%v", inserted, err) + } + inserted, err = db.ApplySecurityEventStreamUpdated(ctx, issuer, audience, streamUpdateJTI, streamID, "paused", "maintenance", now) + if err != nil || inserted { + t.Fatalf("duplicate stream update inserted=%v error=%v", inserted, err) + } + confirmAt("paused-verification-one", now.Add(363*time.Second)) + confirmAt("paused-verification-two", now.Add(364*time.Second)) + evaluation, err = db.EvaluateOIDCSecurityEvent(ctx, issuer, audience, subjectIssuer, tenantID, subject, now, now.Add(364*time.Second), 180*time.Second) + if err != nil || !evaluation.RequireIntrospection || evaluation.Mode != "introspection_fallback" { + t.Fatalf("stream update fallback=%#v error=%v", evaluation, err) + } + _, _ = db.pool.Exec(ctx, `UPDATE gateway_security_event_stream_state SET stream_status='enabled',mode='push_healthy',fallback_since=NULL, + consecutive_verifications=0,last_verification_at=$3 WHERE issuer=$1 AND audience=$2`, issuer, audience, now) + + revokedAt := now.Add(-10 * time.Second) + input := ApplySessionRevokedInput{Issuer: issuer, Audience: audience, JTI: uuid.NewString(), TransactionID: uuid.NewString(), SubjectIssuer: subjectIssuer, TenantID: tenantID, Subject: subject, EventTimestamp: revokedAt, InitiatingEntity: "admin"} + result, err := db.ApplySessionRevoked(ctx, input) + if err != nil || !result.WatermarkMoved || result.AuditID == "" { + t.Fatalf("apply result=%#v error=%v", result, err) + } + duplicate, err := db.ApplySessionRevoked(ctx, input) + if err != nil || !duplicate.Duplicate { + t.Fatalf("duplicate result=%#v error=%v", duplicate, err) + } + evaluation, _ = db.EvaluateOIDCSecurityEvent(ctx, issuer, audience, subjectIssuer, tenantID, subject, revokedAt, now, 180*time.Second) + if !evaluation.Revoked { + t.Fatal("token at watermark was accepted") + } + evaluation, _ = db.EvaluateOIDCSecurityEvent(ctx, issuer, audience, "https://other.test/issuer", tenantID, subject, revokedAt, now, 180*time.Second) + if evaluation.Revoked { + t.Fatal("watermark crossed OIDC issuer boundary") + } + evaluation, _ = db.EvaluateOIDCSecurityEvent(ctx, issuer, audience, subjectIssuer, tenantID, subject, now, now, 180*time.Second) + if evaluation.Revoked { + t.Fatal("token after watermark was rejected") + } + + _, _ = db.pool.Exec(ctx, `UPDATE gateway_security_event_stream_state SET last_verification_at=$3 + WHERE issuer=$1 AND audience=$2`, issuer, audience, now.Add(-181*time.Second)) + evaluation, err = db.EvaluateOIDCSecurityEvent(ctx, issuer, audience, subjectIssuer, tenantID, subject, now, now, 180*time.Second) + if err != nil || !evaluation.RequireIntrospection || evaluation.Mode != "introspection_fallback" { + t.Fatalf("fallback evaluation=%#v error=%v", evaluation, err) + } +} diff --git a/apps/api/internal/store/security_events_test.go b/apps/api/internal/store/security_events_test.go new file mode 100644 index 0000000..c550c9f --- /dev/null +++ b/apps/api/internal/store/security_events_test.go @@ -0,0 +1,10 @@ +package store + +import "testing" + +func TestSecurityEventHashesAreStableAndSanitized(t *testing.T) { + first := shortSecurityEventHash("sensitive-subject") + if first != shortSecurityEventHash("sensitive-subject") || len(first) != 16 || first == "sensitive-subject" { + t.Fatalf("unsafe or unstable hash: %q", first) + } +} diff --git a/apps/api/migrations/0062_oidc_security_events.sql b/apps/api/migrations/0062_oidc_security_events.sql new file mode 100644 index 0000000..698d862 --- /dev/null +++ b/apps/api/migrations/0062_oidc_security_events.sql @@ -0,0 +1,70 @@ +CREATE TABLE IF NOT EXISTS gateway_security_event_receipts ( + issuer text NOT NULL, + jti uuid NOT NULL, + audience text NOT NULL, + event_type text NOT NULL, + transaction_id text, + tenant_id text, + subject_hash text, + event_timestamp timestamptz, + received_at timestamptz NOT NULL DEFAULT now(), + processed_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (issuer, jti) +); + +CREATE INDEX IF NOT EXISTS idx_gateway_security_event_receipts_received + ON gateway_security_event_receipts(received_at DESC); + +CREATE TABLE IF NOT EXISTS gateway_oidc_revocation_watermarks ( + issuer text NOT NULL, + tenant_id text NOT NULL, + subject text NOT NULL, + revoked_at timestamptz NOT NULL, + source_jti uuid NOT NULL, + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (issuer, tenant_id, subject) +); + +CREATE INDEX IF NOT EXISTS idx_gateway_oidc_revocation_watermarks_lookup + ON gateway_oidc_revocation_watermarks(issuer, tenant_id, subject, revoked_at); + +CREATE TABLE IF NOT EXISTS gateway_security_event_stream_state ( + issuer text NOT NULL, + audience text NOT NULL, + stream_id uuid NOT NULL, + stream_status text NOT NULL DEFAULT 'unknown', + mode text NOT NULL DEFAULT 'bootstrap', + last_verification_at timestamptz, + bootstrap_until timestamptz, + consecutive_verifications integer NOT NULL DEFAULT 0, + pending_state_hash bytea, + pending_state_created_at timestamptz, + fallback_since timestamptz, + last_error_category text, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (issuer, audience), + UNIQUE (stream_id), + CONSTRAINT gateway_security_event_stream_state_mode + CHECK (mode IN ('bootstrap','push_healthy','introspection_fallback')), + CONSTRAINT gateway_security_event_stream_state_status + CHECK (stream_status IN ('unknown','enabled','paused','disabled')), + CONSTRAINT gateway_security_event_stream_state_pending_hash + CHECK (pending_state_hash IS NULL OR octet_length(pending_state_hash) = 32) +); + +CREATE TABLE IF NOT EXISTS gateway_security_event_verification_challenges ( + issuer text NOT NULL, + audience text NOT NULL, + state_hash bytea NOT NULL, + created_at timestamptz NOT NULL, + expires_at timestamptz NOT NULL, + PRIMARY KEY (issuer, audience, state_hash), + FOREIGN KEY (issuer, audience) + REFERENCES gateway_security_event_stream_state(issuer, audience) ON DELETE CASCADE, + CONSTRAINT gateway_security_event_verification_challenge_hash + CHECK (octet_length(state_hash) = 32) +); + +CREATE INDEX IF NOT EXISTS idx_gateway_security_event_challenges_expiry + ON gateway_security_event_verification_challenges(expires_at); diff --git a/docker-compose.yml b/docker-compose.yml index 67df0f4..eda6f9b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -14,6 +14,22 @@ x-api-environment: &api-environment OIDC_REQUIRED_SCOPES: ${OIDC_REQUIRED_SCOPES:-gateway.access} OIDC_JWKS_CACHE_TTL_SECONDS: ${OIDC_JWKS_CACHE_TTL_SECONDS:-300} OIDC_ACCEPT_LEGACY_HS256: ${OIDC_ACCEPT_LEGACY_HS256:-true} + 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_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} OIDC_JIT_PROVISIONING_ENABLED: ${OIDC_JIT_PROVISIONING_ENABLED:-false} OIDC_GATEWAY_TENANT_KEY: ${OIDC_GATEWAY_TENANT_KEY:-} OIDC_BROWSER_SESSION_ENABLED: ${OIDC_BROWSER_SESSION_ENABLED:-true} diff --git a/docker/nginx.conf b/docker/nginx.conf index 6b892c3..510bd73 100644 --- a/docker/nginx.conf +++ b/docker/nginx.conf @@ -10,6 +10,29 @@ server { return 308 /gateway-api/; } + # RFC 8935 receiver: exact path, no redirect, short bounded upstream timeout. + location = /api/v1/security-events/ssf { + client_max_body_size 64k; + client_body_buffer_size 64k; + client_body_timeout 5s; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_set_header Connection ""; + proxy_connect_timeout 3s; + proxy_read_timeout 10s; + proxy_send_timeout 10s; + proxy_redirect off; + proxy_pass http://api:8088; + } + + # Metrics stay on the API's monitoring network and are not exposed by Web. + location = /gateway-api/metrics { + return 404; + } + location /gateway-api/ { proxy_http_version 1.1; proxy_set_header Host $host; diff --git a/docs/security/ssf-session-revocation.md b/docs/security/ssf-session-revocation.md new file mode 100644 index 0000000..f2cc589 --- /dev/null +++ b/docs/security/ssf-session-revocation.md @@ -0,0 +1,56 @@ +# SSF/CAEP 实时会话撤销运行手册 + +Gateway 可选接收 Auth Center 通过 RFC 8935 Push 投递的 RFC 8417 Security Event Token，并处理 OpenID CAEP `session-revoked`。默认关闭；关闭时不会注册接收路由或启动状态机，也不会改变 OIDC、BFF、API Key 或 Legacy JWT 行为。 + +## 启用前提 + +- 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. 运行数据库迁移，部署仍保持功能关闭。 +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，再执行测试用户撤销。 + +健康时 Gateway 每 60 秒请求一次 Verification，不按业务 QPS 请求 Auth Center。180 秒没有匹配的有效 Verification 会进入 `introspection_fallback`；只有最近收到的 Stream 状态为 `enabled` 时，连续两次 Verification 成功才会恢复，`paused`/`disabled` 状态下心跳不会错误恢复推送信任。降级期间内省不可用会对 OIDC 返回可审计的 503，API Key 继续工作。 + +## 零停机 Bearer 轮换 + +1. 在 Gateway 写入 `OIDC_SECURITY_EVENTS_BEARER_SECRET_NEXT`，同时接受 current/next。 +2. 更新 Auth Center 的 Stream `credential_ref` 指向 next。 +3. 发起 Verification，并确认投递和状态机健康。 +4. 等待至少 180 秒后，将 next 提升为 current 并清空旧值。 + +Gateway 使用常量时间比较两个值。响应、日志、审计和指标都不得包含 Authorization Header 或原始 SET。 + +## 监控与告警 + +`GET /metrics` 输出以下低基数指标；生产入口应仅允许监控网络访问该路径： + +- `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 刷新失败。 + +至少配置以下告警：Verification age 超过 120 秒、fallback 超过 5 分钟、内省 failed 增长、SET rejected 增长、撤销端到端 P99 超过 3 秒。日志只能携带脱敏 Trace ID、Audit ID 和错误类别。 + +## 回滚 + +先在 Auth Center 暂停 Stream，再保持 Gateway 启用但持续使用 RFC 7662，确认没有遗漏撤销后才关闭接收功能。保留迁移表和 Secret，不做破坏性数据库回滚。Legacy HS256 Token 和 API Key 不在本协议撤销范围内。 + +## 验收证据 + +真实链路验收只能使用明确标记为测试用途的 Application，且必须在自动化测试通过后执行。报告需包含脱敏 Trace、Claims 结构、截图、Audit ID、投递延迟、fallback 与恢复证据。涉及人工、真实设备或第三方操作时记录 `SKIPPED_HUMAN_OR_THIRD_PARTY_REQUIRED`，Mock、契约和安全测试不得跳过。