feat(backend): real-time stream anomaly detection & high-velocity drain sentinel (#1469)

backend/.env.example

@@ -147,6 +147,93 @@ OTEL_EXPORTER_OTLP_ENDPOINT=
 # Ledgers a POST /v1/streams/simulate footprint stays valid for (default: 10)
 SIMULATION_VALIDITY_LEDGERS=10
 
+# ─── Compliance / Sanctions Screening (Issue #1470) ───────────────────────────
+# Master switch for the sanctions / OFAC screening middleware that guards stream
+# creation, top-up (deposit) and withdrawal. Defaults to false so local testing
+# and self-hosted deployments stay fully permissive.
+COMPLIANCE_ENFORCEMENT_ENABLED=false
+
+# Addresses with a risk score strictly greater than this threshold (0-100) are
+# blocked alongside explicitly sanctioned addresses (default: 70)
+COMPLIANCE_RISK_THRESHOLD=70
+
+# How to react when the screening provider itself fails:
+#   fail-closed (default) -> block the request with 503
+#   fail-open             -> allow the request and record an audit event
+COMPLIANCE_FAILURE_MODE=fail-closed
+
+# Time-to-live in seconds for cached screening results (minimize provider cost)
+COMPLIANCE_CACHE_TTL_SECONDS=86400
+
+# Screening data source: 'local' (in-process lists) or 'external' (provider API)
+COMPLIANCE_PROVIDER=local
+
+# Comma-separated always-allowed addresses (checked before the denylist)
+COMPLIANCE_ALLOWLIST=
+
+# Comma-separated always-blocked addresses (useful for local testing)
+COMPLIANCE_DENYLIST=
+
+# Optional path to a JSON sanctions feed (OFAC SDN export). Accepts an array of
+# addresses, an array of { address, tags } records, or an { addresses: [] } map.
+COMPLIANCE_SANCTIONS_FILE=
+
+# External provider settings (required when COMPLIANCE_PROVIDER=external)
+COMPLIANCE_EXTERNAL_API_URL=
+COMPLIANCE_EXTERNAL_API_KEY=
+COMPLIANCE_EXTERNAL_API_TIMEOUT_MS=5000
+
+# ─── Sentinel anomaly detection / drain sentinel (Issue #1469) ────────────────
+# Real-time velocity analysis over indexed stream lifecycle events. Master
+# switch for the sentinel; when false the indexer stops feeding it and no
+# anomalies are recorded. Defaults to true.
+SENTINEL_ENABLED=true
+
+# Rolling window the current velocity is measured over (milliseconds, default 60000)
+SENTINEL_VELOCITY_WINDOW_MS=60000
+# Baseline window a spike is compared against (milliseconds, default 86400000)
+SENTINEL_BASELINE_WINDOW_MS=86400000
+# Spike factor: current window / baseline average. 4 == a 300% increase (default 4)
+SENTINEL_VELOCITY_SPIKE_MULTIPLIER=4
+# Minimum events in the window before a velocity spike can fire (default 5)
+SENTINEL_VELOCITY_MIN_EVENTS=5
+
+# Distinct streams a single address may drain before MULTI_STREAM_DRAIN fires (default 20)
+SENTINEL_MULTI_STREAM_MAX=20
+# Ledgers the multi-stream drain window spans (default 3)
+SENTINEL_MULTI_STREAM_WINDOW_LEDGERS=3
+
+# Single-withdrawal notional (USD) treated as high-value (default 100000).
+# Evaluated only when the caller supplies a USD valuation (`amountUsd`).
+SENTINEL_HIGH_VALUE_THRESHOLD_USD=100000
+# Percentage of a stream's deposit a single withdrawal may consume (default 80)
+SENTINEL_BALANCE_DRAIN_PCT=80
+
+# Stream creations per token inside the spike window before alerting (default 25)
+SENTINEL_CREATION_SPIKE_THRESHOLD=25
+# Creation spike window (milliseconds, default 300000)
+SENTINEL_CREATION_SPIKE_WINDOW_MS=300000
+# A stream with less runway than this counts toward the zero-runway flood (seconds, default 60)
+SENTINEL_ZERO_RUNWAY_MAX_SECONDS=60
+# Zero-runway flood window (milliseconds, default 300000)
+SENTINEL_ZERO_RUNWAY_WINDOW_MS=300000
+# Near-zero-runway streams in the window that constitute a flood (default 50)
+SENTINEL_ZERO_RUNWAY_FLOOD_THRESHOLD=50
+
+# Suppress repeat alerts for the same rule + subject within this window (ms, default 60000)
+SENTINEL_ALERT_COOLDOWN_MS=60000
+# Incident history retention (milliseconds, default 86400000)
+SENTINEL_RETENTION_MS=86400000
+# Trailing window the aggregate 0-100 threat score is computed over (ms, default 900000)
+SENTINEL_THREAT_WINDOW_MS=900000
+
+# Unified alert fan-out. Any subset may be configured; unset channels are skipped.
+SENTINEL_SLACK_WEBHOOK_URL=
+SENTINEL_DISCORD_WEBHOOK_URL=
+SENTINEL_PAGERDUTY_ROUTING_KEY=
+# Secret used to HMAC-sign emergency pause proposals. Falls back to JWT_SECRET.
+SENTINEL_PROPOSAL_SECRET=
+
 # ─── Admin dead-letter quarantine ─────────────────────────────────────────────
 # No configuration required. Quarantined indexer events are managed through:
 #   GET    /v1/admin/indexer/dead-letter

backend/prisma/migrations/20260928000000_add_compliance_tables/migration.sql

@@ -0,0 +1,57 @@
+-- Compliance / sanctions screening schema (Issue #1470)
+--
+-- ComplianceAuditLog: immutable audit trail for screening decisions.
+--   Every blocked or failed screening records the address, request IP, risk
+--   score and provider tags so enterprise deployments can prove wallets were
+--   screened before funds moved.
+-- KycAttestation: SEP-0009 KYC/AML identity attestations submitted by
+--   organizations, stored with their cryptographic proof.
+
+-- CreateTable
+CREATE TABLE "ComplianceAuditLog" (
+    "id" TEXT NOT NULL,
+    "eventType" TEXT NOT NULL,
+    "action" TEXT,
+    "address" TEXT NOT NULL,
+    "ipAddress" TEXT,
+    "riskScore" INTEGER NOT NULL,
+    "tags" TEXT[],
+    "isSanctioned" BOOLEAN NOT NULL DEFAULT false,
+    "metadata" TEXT,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "ComplianceAuditLog_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "KycAttestation" (
+    "id" TEXT NOT NULL,
+    "organization" TEXT NOT NULL,
+    "subjectAddress" TEXT NOT NULL,
+    "sep9Fields" TEXT NOT NULL,
+    "proof" TEXT NOT NULL,
+    "proofType" TEXT,
+    "status" TEXT NOT NULL DEFAULT 'PENDING',
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "KycAttestation_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE INDEX "ComplianceAuditLog_address_idx" ON "ComplianceAuditLog"("address");
+
+-- CreateIndex
+CREATE INDEX "ComplianceAuditLog_eventType_idx" ON "ComplianceAuditLog"("eventType");
+
+-- CreateIndex
+CREATE INDEX "ComplianceAuditLog_createdAt_idx" ON "ComplianceAuditLog"("createdAt");
+
+-- CreateIndex
+CREATE INDEX "KycAttestation_organization_idx" ON "KycAttestation"("organization");
+
+-- CreateIndex
+CREATE INDEX "KycAttestation_subjectAddress_idx" ON "KycAttestation"("subjectAddress");
+
+-- CreateIndex
+CREATE INDEX "KycAttestation_status_idx" ON "KycAttestation"("status");

backend/prisma/schema.prisma

@@ -153,6 +153,47 @@ model AlertHistory {
   @@index([sentAt])
 }
 
+// ComplianceAuditLog model - immutable security audit trail for sanctions /
+// risk screening outcomes (Issue #1470). Every blocked interaction records the
+// originating address, request IP, risk score and provider tags so enterprise
+// deployments have audit-ready proof that wallets were screened before funds
+// moved.
+model ComplianceAuditLog {
+  id           String   @id @default(uuid())
+  eventType    String   // e.g. "SCREENING_BLOCKED", "SCREENING_ERROR", "KYC_ATTESTATION_SUBMITTED"
+  action       String?  // Route action, e.g. "stream.create"
+  address      String   // Stellar address that was screened
+  ipAddress    String?  // Originating request IP
+  riskScore    Int      // 0 (clean) to 100 (blocked)
+  tags         String[] // Provider tags, e.g. ["OFAC", "Darknet"]
+  isSanctioned Boolean  @default(false)
+  metadata     String?  // JSON string for extra screening context
+  createdAt    DateTime @default(now())
+
+  @@index([address])
+  @@index([eventType])
+  @@index([createdAt])
+}
+
+// KycAttestation model - SEP-0009 identity attestations submitted by
+// organizations. Stores the standardised KYC/AML field payload alongside the
+// cryptographic proof so a deployment can prove a recipient was verified.
+model KycAttestation {
+  id             String   @id @default(uuid())
+  organization   String   // Organization / wallet that submitted the attestation
+  subjectAddress String   // Wallet the identity belongs to
+  sep9Fields     String   // JSON string of the SEP-0009 KYC/AML field payload
+  proof          String   // Cryptographic proof / signature over the payload
+  proofType      String?  // Signature scheme, e.g. "ed25519"
+  status         String   @default("PENDING") // PENDING | VERIFIED | REJECTED
+  createdAt      DateTime @default(now())
+  updatedAt      DateTime @updatedAt
+
+  @@index([organization])
+  @@index([subjectAddress])
+  @@index([status])
+}
+
 // IndexerDeadLetterEvent model - quarantines Soroban events whose processing
 // failed (unexpected payload shape, transient DB lock, RPC timeout mid-handler).
 // A quarantined event is never retried inline by the poll loop, so one malformed