feat: 接入 SSF 实时会话撤销

2026-07-14 17:06:41 +08:00
parent 30ad0e9f2c
commit f30aaeb2d4
28 changed files with 2597 additions and 91 deletions
@@ -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]
+}
@@ -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)
+	}
+}
@@ -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)
+	}
+}