From 6b44d3ee3c5cf621307401b525a515ed49c290ad Mon Sep 17 00:00:00 2001 From: devonahi Date: Tue, 20 Oct 2026 17:47:06 +0100 Subject: [PATCH] feat(analytics): flag-gated slices for spikes 616/617/618/619 (#479 #480 #481 #482) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit phase-178 (#479): leaderboard integrity — multiplicative scoring decayed × farmPenalty × sybilDiscount (7d half-life decay, event-burst / counterparty-concentration / point-velocity rules floored at 0.15, sybil band discounts with missing band = caution), public GET + admin-keyed POST /api/leaderboard scoring before persisting. phase-179 (#480): retention and GDPR erasure — TTL sweeps for expired SEP-10 sessions, consumed challenges and 90-day notifications (fail-safe on missing timestamps); POST /api/user/erasure gated by legal hold (423), fingerprint-only receipts in gdpr-erasure-log.json, consent ledger deliberately retained per Art. 17(3)(b). phase-180 (#481): analytics warehouse export — flat row model with unconditional redaction, schema-versioned manifest emitted as NDJSON header or Parquet key_value_metadata (hyparquet-writer), admin-keyed GET /api/analytics/export streaming downloads with checksum headers. phase-181 (#482): anomaly detection — robust z-score (median+MAD) on volume and per-run counter deltas, deterministic balance-drop tiers, detect-before-push baselines, hourly cron routing findings to webhooks with distinct kind+metric titles. All behind default-off flags; 45 tests, findings docs, flag rows and technical sections updated. --- .env.local.example | 14 + PROJECT_ARCHITECTURE.md | 4 + app/api/analytics/export/route.ts | 73 ++++ app/api/cron/anomaly-detect/route.ts | 78 ++++ app/api/cron/retention-sweep/route.ts | 66 ++++ app/api/leaderboard/route.ts | 146 +++++++ app/api/user/erasure/route.ts | 81 ++++ docs/TECHNICAL.md | 44 +++ .../616-leaderboard-integrity-findings.md | 30 ++ .../617-retention-gdpr-erasure-findings.md | 32 ++ ...618-analytics-warehouse-export-findings.md | 29 ++ docs/spikes/619-anomaly-detection-findings.md | 29 ++ lib/__tests__/analytics-warehouse.test.ts | 210 ++++++++++ lib/__tests__/anomaly-detection.test.ts | 287 ++++++++++++++ lib/__tests__/leaderboard-integrity.test.ts | 181 +++++++++ lib/__tests__/retention-erasure.test.ts | 264 +++++++++++++ lib/analytics-warehouse.ts | 267 +++++++++++++ lib/anomaly-detection.ts | 374 ++++++++++++++++++ lib/feature-flags.ts | 17 + lib/leaderboard-integrity.ts | 238 +++++++++++ lib/retention-erasure.ts | 292 ++++++++++++++ lib/security-counters.ts | 19 +- lib/server-data-paths.ts | 3 + package-lock.json | 17 + package.json | 2 + vercel.json | 8 + 26 files changed, 2804 insertions(+), 1 deletion(-) create mode 100644 app/api/analytics/export/route.ts create mode 100644 app/api/cron/anomaly-detect/route.ts create mode 100644 app/api/cron/retention-sweep/route.ts create mode 100644 app/api/leaderboard/route.ts create mode 100644 app/api/user/erasure/route.ts create mode 100644 docs/spikes/616-leaderboard-integrity-findings.md create mode 100644 docs/spikes/617-retention-gdpr-erasure-findings.md create mode 100644 docs/spikes/618-analytics-warehouse-export-findings.md create mode 100644 docs/spikes/619-anomaly-detection-findings.md create mode 100644 lib/__tests__/analytics-warehouse.test.ts create mode 100644 lib/__tests__/anomaly-detection.test.ts create mode 100644 lib/__tests__/leaderboard-integrity.test.ts create mode 100644 lib/__tests__/retention-erasure.test.ts create mode 100644 lib/analytics-warehouse.ts create mode 100644 lib/anomaly-detection.ts create mode 100644 lib/leaderboard-integrity.ts create mode 100644 lib/retention-erasure.ts diff --git a/.env.local.example b/.env.local.example index 27d962ec..46c1294b 100644 --- a/.env.local.example +++ b/.env.local.example @@ -221,3 +221,17 @@ NEXT_PUBLIC_CLASSIC_LIQ_ISSUER= # PHASE_PROD_CRON_SECRET_SHA256= # PHASE_PROD_CONTRACT_IDS=C...,C... # PHASE_PREVIEW_CRON_ENABLED=1 + +# ─── Spikes 616–619 (phase-178..181) — default off ──────────────────────── +# phase-178 leaderboard integrity: anti-farm scoring, decay, sybil discount +# NEXT_PUBLIC_FEATURE_PHASE_178=1 +# FEATURE_PHASE_178=1 +# phase-179 retention + GDPR erasure: TTL sweeps, wallet deletion, receipt log +# NEXT_PUBLIC_FEATURE_PHASE_179=1 +# FEATURE_PHASE_179=1 +# phase-180 analytics warehouse export (GET /api/analytics/export?format=ndjson|parquet) +# NEXT_PUBLIC_FEATURE_PHASE_180=1 +# FEATURE_PHASE_180=1 +# phase-181 anomaly detection cron (volume, counter deltas, balance drops) +# NEXT_PUBLIC_FEATURE_PHASE_181=1 +# FEATURE_PHASE_181=1 diff --git a/PROJECT_ARCHITECTURE.md b/PROJECT_ARCHITECTURE.md index fe4ca2fc..4ae4090d 100644 --- a/PROJECT_ARCHITECTURE.md +++ b/PROJECT_ARCHITECTURE.md @@ -216,6 +216,10 @@ is a fixed-size digest of the canonical payload, keeping it under wallet | `phase-171` | `NEXT_PUBLIC_FEATURE_PHASE_171` / `FEATURE_PHASE_171` | Backup and DR snapshots (`lib/backup-snapshot.ts`): lock-consistent sidecar set + `VACUUM INTO` SQLite image, sha256/row-count manifest, `integrity_check`/`foreign_key_check` verify, tmp+rename restore with a restore marker that makes `/api/cron/faucet-health` return 503, and an AES-256-GCM sealed secrets envelope. `scripts/backup-drill.ts` runs snapshot→verify→restore→parity. Spike 605 (issue #469) — partial slice: no off-box shipping or scheduled snapshots | off | Unset var, restart — entry points throw `FLAG_DISABLED`, restore marker ignored; snapshots on disk are inert | | `phase-172` | `NEXT_PUBLIC_FEATURE_PHASE_172` / `FEATURE_PHASE_172` | SQLite→Postgres rehearsal (`lib/db-parity.ts`): dialect helpers (`?`→`$n`, type map, explicit upsert), typed per-table parity (value, `type_coercion`, checksum), per-test schema names, and a `DualWriter` (primary first, bounded/timeout-guarded secondary, drift log, `primary-only` rollback). Needs `PHASE_DATABASE_URL` to enter `dual` mode. Spike 604 (issue #468) — partial slice: no Postgres driver, async CAS re-plumbing or pool storm test | off | Unset var, restart — `resolveDualWriteMode` returns `primary-only`; SQLite stays authoritative, nothing in production paths calls the module | | `phase-173` | `NEXT_PUBLIC_FEATURE_PHASE_173` / `FEATURE_PHASE_173` | Vercel preview isolation (`lib/preview-isolation.ts`): per-preview data root (`/previews/`), `validatePhaseEnv` fails a preview that shares the production issuer, cron secret, contract IDs, or runs mainnet (declared via `PHASE_PROD_ISSUER`, `PHASE_PROD_CRON_SECRET_SHA256`, `PHASE_PROD_CONTRACT_IDS`), preview cron skipped unless `PHASE_PREVIEW_CRON_ENABLED=1`, and `x-vercel-cron` no longer trusted in previews. Spike 603 (issue #467) — partial slice: no per-PR contract deploy, preview CI job, or branch-DB seeding | off | Unset var, restart — data root, cron auth and env validation revert; per-preview directories stay inert | +| `phase-178` | `NEXT_PUBLIC_FEATURE_PHASE_178` / `FEATURE_PHASE_178` | Leaderboard integrity (`lib/leaderboard-integrity.ts`): multiplicative scoring `decayed × farmPenalty × sybilDiscount` — exponential half-life decay (7d), anti-farm rules (`event_burst`, `counterparty_concentration`, `point_velocity`) floored at 0.15, and sybil band discounts (trusted 1 / caution 0.7 / suspect 0.35, missing band = caution). `GET /api/leaderboard` ranks; admin `POST /api/leaderboard` scores before persisting to `leaderboard-entries.json`. Spike 616 (issue #479) — partial slice: no on-chain sybil fetch in-route; callers supply the band | off | Unset var, restart — routes 404, `leaderboard-entries.json` stays inert | +| `phase-179` | `NEXT_PUBLIC_FEATURE_PHASE_179` / `FEATURE_PHASE_179` | Retention and GDPR erasure (`lib/retention-erasure.ts`): TTL sweeps (expired SEP-10 sessions, consumed challenges, 90-day notifications — fail-safe on missing timestamps) via `GET /api/cron/retention-sweep` (daily 03:15, preview + restore fenced); `POST /api/user/erasure` deletes wallet-owned records across five sidecar domains behind `assertNoLegalHold` (423 on hold) and appends fingerprint-only receipts to `gdpr-erasure-log.json`. Spike 617 (issue #480) — partial slice: SQLite tables not walked; consent ledger deliberately retained per Art. 17(3)(b) | off | Unset var, restart — routes 404, receipts stay inert; nothing is deleted while off | +| `phase-180` | `NEXT_PUBLIC_FEATURE_PHASE_180` / `FEATURE_PHASE_180` | Analytics warehouse export (`lib/analytics-warehouse.ts`): flat row model (`counter`/`health`/`volume` tables) with an unconditional redaction pass (wallets fingerprinted, secrets stripped), schema-versioned manifest (`phase-warehouse-export` 1.0.0, rows-only sha256 checksum) emitted as NDJSON header line or Parquet `key_value_metadata` (`hyparquet-writer`). Admin-keyed `GET /api/analytics/export?format=ndjson\|parquet` streams downloads with `X-Warehouse-Schema-Version`/`X-Warehouse-Checksum`. Spike 618 (issue #481) — partial slice: three source tables only | off | Unset var, restart — route 404; no stored warehouse state, collectors read live sources | +| `phase-181` | `NEXT_PUBLIC_FEATURE_PHASE_181` / `FEATURE_PHASE_181` | Anomaly detection (`lib/anomaly-detection.ts`): robust z-score (median+MAD, σ floor) on wash-excluded volume and per-run counter deltas (own counters excluded), deterministic balance drop tiers (≥15% warning, ≥35% critical), detection-before-push baselines in `anomaly-baselines.json` (capped ring). `GET /api/cron/anomaly-detect` (hourly, minute 5, preview + restore fenced) routes findings to webhooks with distinct kind+metric titles. Spike 619 (issue #482) — partial slice: no per-channel thresholds or suppressions | off | Unset var, restart — cron 404s, baselines stay inert, no alerts routed | diff --git a/app/api/analytics/export/route.ts b/app/api/analytics/export/route.ts new file mode 100644 index 00000000..e3e9b55d --- /dev/null +++ b/app/api/analytics/export/route.ts @@ -0,0 +1,73 @@ +import { NextRequest, NextResponse } from "next/server" +import { + isWarehouseExportEnabled, + collectWarehouseRows, + redactWarehouseRows, + formatWarehouseNdjson, + rowsToParquetBuffer, + WarehouseExportError, + WAREHOUSE_SCHEMA_VERSION, +} from "@/lib/analytics-warehouse" +import { checkAdminKey } from "@/lib/admin-key" +import { incSecurityCounter } from "@/lib/security-counters" + +export const dynamic = "force-dynamic" + +export async function GET(request: NextRequest) { + if (!isWarehouseExportEnabled()) { + return NextResponse.json( + { error: "Analytics warehouse export is disabled (set NEXT_PUBLIC_FEATURE_PHASE_180=1)" }, + { status: 404 }, + ) + } + + if (checkAdminKey(request) !== "ok") { + return NextResponse.json( + { error: "Unauthorized: valid x-admin-key required" }, + { status: 401 }, + ) + } + + const formatParam = request.nextUrl.searchParams.get("format")?.toLowerCase() ?? "ndjson" + if (formatParam !== "ndjson" && formatParam !== "parquet") { + return NextResponse.json({ error: "format must be ndjson or parquet" }, { status: 400 }) + } + + try { + const rows = redactWarehouseRows(await collectWarehouseRows()) + incSecurityCounter("warehouse_records_redacted", rows.length) + incSecurityCounter("warehouse_export_runs") + const generatedAt = Date.now() + + if (formatParam === "parquet") { + const { manifest, buffer } = rowsToParquetBuffer(rows, { generatedAt }) + return new NextResponse(Buffer.from(buffer), { + status: 200, + headers: { + "Content-Type": "application/vnd.apache.parquet", + "Content-Disposition": `attachment; filename="phase-warehouse-${generatedAt}.parquet"`, + "X-Warehouse-Schema-Version": WAREHOUSE_SCHEMA_VERSION, + "X-Warehouse-Checksum": manifest.checksum, + }, + }) + } + + const { manifest, body } = formatWarehouseNdjson(rows, { generatedAt }) + return new NextResponse(body, { + status: 200, + headers: { + "Content-Type": "application/x-ndjson; charset=utf-8", + "Content-Disposition": `attachment; filename="phase-warehouse-${generatedAt}.ndjson"`, + "X-Warehouse-Schema-Version": WAREHOUSE_SCHEMA_VERSION, + "X-Warehouse-Checksum": manifest.checksum, + }, + }) + } catch (error) { + incSecurityCounter("warehouse_export_failed") + if (error instanceof WarehouseExportError) { + const status = error.code === "FLAG_DISABLED" ? 404 : error.code === "UNAUTHORIZED" ? 401 : 500 + return NextResponse.json({ error: error.message, code: error.code }, { status }) + } + throw error + } +} diff --git a/app/api/cron/anomaly-detect/route.ts b/app/api/cron/anomaly-detect/route.ts new file mode 100644 index 00000000..a61e3231 --- /dev/null +++ b/app/api/cron/anomaly-detect/route.ts @@ -0,0 +1,78 @@ +import { NextRequest, NextResponse } from "next/server" +import { + isAnomalyDetectionEnabled, + collectAnomalySample, + detectAnomalies, + normalizeBaselineStore, + routeAnomalyAlerts, +} from "@/lib/anomaly-detection" +import { isRestoreInProgress } from "@/lib/backup-snapshot" +import { isCronBlockedInPreview, previewTrustsVercelCronHeader } from "@/lib/preview-isolation" +import { readStore, updateStore } from "@/lib/json-store" +import { incSecurityCounter } from "@/lib/security-counters" + +export const dynamic = "force-dynamic" + +function validateCronAuth(req: NextRequest): boolean { + const authHeader = req.headers.get("authorization") + const cronSecret = process.env.CRON_SECRET?.trim() + + if (process.env.NODE_ENV === "development" && !cronSecret) { + return true + } + + const vercelCronHeader = req.headers.get("x-vercel-cron") + if (vercelCronHeader && previewTrustsVercelCronHeader()) { + return true + } + + if (!cronSecret) return false + if (!authHeader || !authHeader.startsWith("Bearer ")) return false + return authHeader.substring(7) === cronSecret +} + +export async function GET(req: NextRequest) { + if (!validateCronAuth(req)) { + return NextResponse.json({ error: "Unauthorized - valid cron secret required" }, { status: 401 }) + } + + if (!isAnomalyDetectionEnabled()) { + return NextResponse.json( + { error: "Anomaly detection is disabled (set NEXT_PUBLIC_FEATURE_PHASE_181=1)" }, + { status: 404 }, + ) + } + + if (isCronBlockedInPreview()) { + incSecurityCounter("preview_cron_blocked") + return NextResponse.json({ ok: true, skipped: "preview_cron_disabled" }) + } + + // Baselines built on half-restored data would fire false alerts and poison history. + if (await isRestoreInProgress()) { + return NextResponse.json( + { error: "Restore in progress" }, + { status: 503, headers: { "Retry-After": "60" } }, + ) + } + + const stored = await readStore("anomalyBaselines", () => ({})) + const baseline = normalizeBaselineStore(stored) + const sample = await collectAnomalySample() + const { findings, baseline: updated } = detectAnomalies(baseline, sample) + + await updateStore("anomalyBaselines", (data) => { + Object.assign(data, updated) + }) + + const { routed } = await routeAnomalyAlerts(findings) + incSecurityCounter("anomaly_checks_run") + + return NextResponse.json({ + ok: true, + checked_at: sample.ts, + findings, + alerts_routed: routed, + timestamp: new Date().toISOString(), + }) +} diff --git a/app/api/cron/retention-sweep/route.ts b/app/api/cron/retention-sweep/route.ts new file mode 100644 index 00000000..3cb1a164 --- /dev/null +++ b/app/api/cron/retention-sweep/route.ts @@ -0,0 +1,66 @@ +import { NextRequest, NextResponse } from "next/server" +import { isRetentionErasureEnabled, runRetentionSweeps, RetentionErasureError } from "@/lib/retention-erasure" +import { isRestoreInProgress } from "@/lib/backup-snapshot" +import { isCronBlockedInPreview, previewTrustsVercelCronHeader } from "@/lib/preview-isolation" +import { incSecurityCounter } from "@/lib/security-counters" + +export const dynamic = "force-dynamic" + +function validateCronAuth(req: NextRequest): boolean { + const authHeader = req.headers.get("authorization") + const cronSecret = process.env.CRON_SECRET?.trim() + + if (process.env.NODE_ENV === "development" && !cronSecret) { + return true + } + + const vercelCronHeader = req.headers.get("x-vercel-cron") + if (vercelCronHeader && previewTrustsVercelCronHeader()) { + return true + } + + if (!cronSecret) return false + if (!authHeader || !authHeader.startsWith("Bearer ")) return false + return authHeader.substring(7) === cronSecret +} + +export async function GET(req: NextRequest) { + if (!validateCronAuth(req)) { + return NextResponse.json({ error: "Unauthorized - valid cron secret required" }, { status: 401 }) + } + + if (!isRetentionErasureEnabled()) { + return NextResponse.json( + { error: "Retention sweeps are disabled (set NEXT_PUBLIC_FEATURE_PHASE_179=1)" }, + { status: 404 }, + ) + } + + if (isCronBlockedInPreview()) { + incSecurityCounter("preview_cron_blocked") + return NextResponse.json({ ok: true, skipped: "preview_cron_disabled" }) + } + + // Sweeping prunes sidecars — running mid-restore would fight the restore write. + if (await isRestoreInProgress()) { + return NextResponse.json( + { error: "Restore in progress" }, + { status: 503, headers: { "Retry-After": "60" } }, + ) + } + + try { + const result = await runRetentionSweeps() + return NextResponse.json({ + ok: true, + total_removed: result.total_removed, + sweeps: result.sweeps, + timestamp: new Date().toISOString(), + }) + } catch (error) { + if (error instanceof RetentionErasureError) { + return NextResponse.json({ error: error.message, code: error.code }, { status: 400 }) + } + throw error + } +} diff --git a/app/api/leaderboard/route.ts b/app/api/leaderboard/route.ts new file mode 100644 index 00000000..c141e80b --- /dev/null +++ b/app/api/leaderboard/route.ts @@ -0,0 +1,146 @@ +import { NextRequest, NextResponse } from "next/server" +import { + isLeaderboardIntegrityEnabled, + rankLeaderboard, + scoreLeaderboardCandidate, + LeaderboardIntegrityError, + LEADERBOARD_SCHEMA_VERSION, + type LeaderboardCandidate, + type SybilBand, +} from "@/lib/leaderboard-integrity" +import { checkAdminKey } from "@/lib/admin-key" +import { readStore, updateStore } from "@/lib/json-store" +import { readBoundedBody } from "@/lib/bounded-body" +import { incSecurityCounter } from "@/lib/security-counters" + +export const dynamic = "force-dynamic" + +type EntriesStore = { + schema?: string + version?: string + entries?: Record +} + +const G_ADDRESS_RE = /^G[A-Z2-7]{55}$/ +const SYBIL_BANDS: SybilBand[] = ["trusted", "caution", "suspect"] +const MAX_EVENTS_PER_ENTRY = 500 +const MAX_BODY_BYTES = 256 * 1024 + +function normalizeStore(raw: unknown): EntriesStore { + const parsed = (raw ?? {}) as Partial + return { + schema: LEADERBOARD_SCHEMA_VERSION, + version: LEADERBOARD_SCHEMA_VERSION, + entries: parsed.entries && typeof parsed.entries === "object" ? parsed.entries : {}, + } +} + +function parseCandidate(body: unknown): LeaderboardCandidate { + if (typeof body !== "object" || body === null) { + throw new LeaderboardIntegrityError("VALIDATION_FAILED", "JSON object body required") + } + const raw = body as Record + if (typeof raw.wallet !== "string" || !G_ADDRESS_RE.test(raw.wallet)) { + throw new LeaderboardIntegrityError("VALIDATION_FAILED", "wallet must be a Stellar G... address") + } + if (!Array.isArray(raw.events) || raw.events.length > MAX_EVENTS_PER_ENTRY) { + throw new LeaderboardIntegrityError( + "VALIDATION_FAILED", + `events must be an array of at most ${MAX_EVENTS_PER_ENTRY} entries`, + ) + } + const events = raw.events.map((entry) => { + const e = (entry ?? {}) as Record + if (!Number.isFinite(e.points as number) || !Number.isFinite(e.at as number)) { + throw new LeaderboardIntegrityError("VALIDATION_FAILED", "each event needs finite points and at (epoch ms)") + } + if (e.counterparty !== undefined && e.counterparty !== null && typeof e.counterparty !== "string") { + throw new LeaderboardIntegrityError("VALIDATION_FAILED", "counterparty must be a string when present") + } + return { + points: e.points as number, + at: e.at as number, + ...(typeof e.counterparty === "string" && e.counterparty.length > 0 ? { counterparty: e.counterparty } : {}), + } + }) + const band = + raw.sybil_band === undefined || raw.sybil_band === null + ? undefined + : SYBIL_BANDS.includes(raw.sybil_band as SybilBand) + ? (raw.sybil_band as SybilBand) + : (() => { + throw new LeaderboardIntegrityError("VALIDATION_FAILED", "sybil_band must be trusted|caution|suspect") + })() + return { wallet: raw.wallet, events, ...(band ? { sybilBand: band } : {}) } +} + +export async function GET(request: NextRequest) { + if (!isLeaderboardIntegrityEnabled()) { + return NextResponse.json( + { error: "Leaderboard integrity is disabled (set NEXT_PUBLIC_FEATURE_PHASE_178=1)" }, + { status: 404 }, + ) + } + + const limitParam = Number(request.nextUrl.searchParams.get("limit") ?? "50") + const limit = Number.isFinite(limitParam) ? Math.max(1, Math.min(Math.floor(limitParam), 200)) : 50 + + const store = await readStore("leaderboardEntries", () => ({})) + const entries = Object.values(normalizeStore(store).entries ?? {}) + const ranked = rankLeaderboard(entries).slice(0, limit) + + incSecurityCounter("leaderboard_scored") + return NextResponse.json({ + schema: LEADERBOARD_SCHEMA_VERSION, + generated_at: Date.now(), + count: ranked.length, + entries: ranked, + }) +} + +export async function POST(request: NextRequest) { + if (!isLeaderboardIntegrityEnabled()) { + return NextResponse.json( + { error: "Leaderboard integrity is disabled (set NEXT_PUBLIC_FEATURE_PHASE_178=1)" }, + { status: 404 }, + ) + } + + if (checkAdminKey(request) !== "ok") { + return NextResponse.json({ error: "Unauthorized: valid x-admin-key required" }, { status: 401 }) + } + + let body: unknown + try { + const text = await readBoundedBody(request, MAX_BODY_BYTES) + body = JSON.parse(text || "{}") + } catch { + return NextResponse.json({ error: "Invalid JSON body" }, { status: 400 }) + } + + try { + const candidate = parseCandidate(body) + // Score before persisting so malformed events never reach the store. + const score = scoreLeaderboardCandidate(candidate) + await updateStore("leaderboardEntries", (data) => { + const store = normalizeStore(data) + store.entries = store.entries ?? {} + store.entries[candidate.wallet] = candidate + data.schema = store.schema + data.version = store.version + data.entries = store.entries + }) + incSecurityCounter("leaderboard_entries_ingested") + if (score.farmPenalty < 1) incSecurityCounter("leaderboard_farm_penalized") + if (score.sybilDiscount < 1) incSecurityCounter("leaderboard_sybil_discounted") + return NextResponse.json({ ok: true, score }) + } catch (error) { + if (error instanceof LeaderboardIntegrityError) { + return NextResponse.json( + { error: error.message, code: error.code }, + { status: error.code === "FLAG_DISABLED" ? 404 : 400 }, + ) + } + throw error + } +} diff --git a/app/api/user/erasure/route.ts b/app/api/user/erasure/route.ts new file mode 100644 index 00000000..87fed92f --- /dev/null +++ b/app/api/user/erasure/route.ts @@ -0,0 +1,81 @@ +import { NextRequest, NextResponse } from "next/server" +import { + isRetentionErasureEnabled, + executeWalletErasure, + RetentionErasureError, +} from "@/lib/retention-erasure" +import { checkAdminKey } from "@/lib/admin-key" +import { readBoundedBody } from "@/lib/bounded-body" + +export const dynamic = "force-dynamic" + +const MAX_BODY_BYTES = 16 * 1024 + +/** + * GDPR Article 17 erasure. Authorization is fail-closed and mirrors the + * portability export route: an admin key grants cross-wallet erasure, + * otherwise the caller's wallet header must match the erasure target — a + * missing header is 401, a mismatched one is 403. + */ +export async function POST(request: NextRequest) { + if (!isRetentionErasureEnabled()) { + return NextResponse.json( + { error: "GDPR erasure is disabled (set NEXT_PUBLIC_FEATURE_PHASE_179=1)" }, + { status: 404 }, + ) + } + + let body: unknown + try { + const text = await readBoundedBody(request, MAX_BODY_BYTES) + body = JSON.parse(text || "{}") + } catch { + return NextResponse.json({ error: "Invalid JSON body" }, { status: 400 }) + } + + const wallet = typeof (body as Record)?.wallet === "string" + ? ((body as Record).wallet).trim() + : "" + if (!wallet) { + return NextResponse.json({ error: "wallet is required" }, { status: 400 }) + } + + const adminOk = checkAdminKey(request) === "ok" + const callerWallet = + request.headers.get("x-wallet-address")?.trim() || + request.headers.get("acting_wallet")?.trim() || + "" + + if (!adminOk) { + if (!callerWallet) { + return NextResponse.json( + { error: "Unauthorized: x-wallet-address header or x-admin-key required" }, + { status: 401 }, + ) + } + if (callerWallet !== wallet) { + return NextResponse.json( + { error: "Unauthorized: caller wallet does not match erasure target" }, + { status: 403 }, + ) + } + } + + try { + const receipt = await executeWalletErasure(wallet, { + initiatedBy: adminOk && callerWallet !== wallet ? "admin" : "wallet", + }) + return NextResponse.json({ ok: true, receipt }) + } catch (error) { + if (error instanceof RetentionErasureError) { + const status = + error.code === "FLAG_DISABLED" + ? 404 + : error.code === "LEGAL_HOLD_ACTIVE" + ? 423 + : 400 + return NextResponse.json({ error: error.message, code: error.code }, { status }) + } + throw error + } +} diff --git a/docs/TECHNICAL.md b/docs/TECHNICAL.md index 8696fae2..1b22098c 100644 --- a/docs/TECHNICAL.md +++ b/docs/TECHNICAL.md @@ -743,3 +743,47 @@ Flag: `phase-173` (`NEXT_PUBLIC_FEATURE_PHASE_173` / `FEATURE_PHASE_173`). Defau - `app/api/cron/faucet-health/route.ts`: preview runs return `{ skipped: "preview_cron_disabled" }` unless `PHASE_PREVIEW_CRON_ENABLED=1`; the unauthenticated-header shortcut is disabled in previews. - Caveat: Vercel `/tmp` is per-instance, so namespacing prevents cross-preview bleed, not loss between instances. - Counters: `preview_isolation_violation`, `preview_cron_blocked`. + +## 24. Leaderboard integrity: anti-farm scoring, decay and sybil discount (spike 616) + +Flag: `phase-178` (`NEXT_PUBLIC_FEATURE_PHASE_178` / `FEATURE_PHASE_178`). Default off. See `PROJECT_ARCHITECTURE.md` §10 and `docs/spikes/616-leaderboard-integrity-findings.md`. + +- `lib/leaderboard-integrity.ts`: `scoreLeaderboardCandidate` computes `finalScore = decayedPoints × farmPenalty × sybilDiscount` from an explicit `now`. Decay is `0.5 ** (age / 7d)` (same shape as the pricing oracle's wash discount). `detectFarmPatterns` multiplies down per rule — `event_burst` (×0.5, >12 events in 1h), `counterparty_concentration` (×0.6, one counterparty >80% of ≥6 counterpartied events), `point_velocity` (×0.7, >1000 points/day) — floored at `farmPenaltyFloor` 0.15. `sybilDiscountFor`: trusted 1, caution 0.7, suspect 0.35, missing band = caution. +- Routes: `GET /api/leaderboard?limit=` ranks stored entries (ties break by wallet asc); `POST /api/leaderboard` requires `x-admin-key` and scores before persisting, so malformed events never reach `leaderboard-entries.json`. +- Counters: `leaderboard_entries_ingested`, `leaderboard_scored`, `leaderboard_farm_penalized`, `leaderboard_sybil_discounted`. +- Not covered: fetching sybil bands from Horizon inside the route (callers supply the band from phase-145 helpers). + +## 25. Retention TTL sweeps and GDPR erasure (spike 617) + +Flag: `phase-179` (`NEXT_PUBLIC_FEATURE_PHASE_179` / `FEATURE_PHASE_179`). Default off. See `docs/spikes/617-retention-gdpr-erasure-findings.md`. + +- `lib/retention-erasure.ts` sweeps: `sweepByTimestamp` (pure; records with a missing/non-finite timestamp are KEPT), `sweepAuthSessions` (expired SEP-10 sessions — `expires_at` normalized from seconds to ms — and consumed challenges past 24h), `sweepNotificationStore` (90-day notifications, empty wallet buckets dropped). `runRetentionSweeps()` applies both through `updateStore` and counts `retention_sweep_runs` / `retention_records_expired`. +- `executeWalletErasure(wallet)`: validates the `G...` address, runs `assertNoLegalHold("wallet", …, "erasure")` (a hit → `RetentionErasureError LEGAL_HOLD_ACTIVE`, HTTP 423 in the route, `gdpr_erasure_blocked` counted, nothing deleted), then erases five registered domains (`ERASURE_TARGETS`: notifications, age declarations, follows, push subscriptions, auth sessions) and appends a receipt to `gdpr-erasure-log.json` — fingerprint only, capped at 1000 receipts, schema `phase-gdpr-article-17-erasure` 1.0.0. +- Deliberate exclusion: `terms-consents.json` is never erased (Art. 17(3)(b) proof of legal basis). SQLite tables are a documented partial slice. +- `POST /api/user/erasure`: fail-closed auth — admin key (`checkAdminKey`) or a matching `x-wallet-address`/`acting_wallet` header; missing header 401, mismatch 403, hold 423. +- `GET /api/cron/retention-sweep`: daily 03:15 UTC, `validateCronAuth` + `isCronBlockedInPreview` + `isRestoreInProgress` (503 + `Retry-After`). + +## 26. Analytics warehouse export: Parquet and NDJSON (spike 618) + +Flag: `phase-180` (`NEXT_PUBLIC_FEATURE_PHASE_180` / `FEATURE_PHASE_180`). Default off. See `docs/spikes/618-analytics-warehouse-export-findings.md`. + +- `lib/analytics-warehouse.ts`: flat `WarehouseRow` (`table`, `ts`, `metric`, `value`, `status`, `wallet_fingerprint`) collected from the security-counter snapshot, distributor health history, and wash-excluded accepted-offer volume (`wash_flagged = 0`, matching `lib/wash-screening.ts` aggregates). +- Redaction is unconditional: `redactWarehouseRow` fingerprints raw `G...` fields and rewrites wallet/secret-looking text in `metric`/`status`; it is idempotent (uses a fresh non-global regex because `WALLET_KEY_RE` is `/g` and `.test()` on it is stateful). +- Manifest (`buildWarehouseManifest`): `phase-warehouse-export` 1.0.0, `generated_at`, `row_count`, sorted `tables`, `redacted: true`, `checksum` = sha256 over the rows-only NDJSON body (reproducible across formats; mirrored in `X-Warehouse-Checksum`). +- Formats: NDJSON emits the manifest as line 1 then one row per line (`application/x-ndjson`); Parquet uses `hyparquet-writer`'s `parquetWriteBuffer` with typed columns and the manifest in `key_value_metadata` (`phase.schema_version`, `phase.checksum`, `phase.redacted`). +- `GET /api/analytics/export?format=ndjson|parquet`: `checkAdminKey` fail-closed (401 unconfigured/denied), 400 on unknown format, `Content-Disposition` attachment, headers `X-Warehouse-Schema-Version` + `X-Warehouse-Checksum`. +- Counters: `warehouse_export_runs`, `warehouse_records_redacted`, `warehouse_export_failed`. + +## 27. Anomaly detection with alert routing (spike 619) + +Flag: `phase-181` (`NEXT_PUBLIC_FEATURE_PHASE_181` / `FEATURE_PHASE_181`). Default off. See `docs/spikes/619-anomaly-detection-findings.md`. + +- `lib/anomaly-detection.ts` detectors over one `AnomalySample`: + - volume — `robustStats` (median + MAD×1.4826 → stddev fallback → σ floor 1e-9 so a flat baseline with a non-zero value still breaches), `z ≥ 3.5` warning / `≥ 6` critical, `minSamples` 8. + - counters — `counterDeltas` per run (resets and first-seen keys skipped), z-scored like volume; `collectAnomalySample` excludes `anomaly_*` counters so the detector cannot alert on itself. + - balances — `detectBalanceDrops`: ≥15% warning, ≥35% critical vs the previous run; no baseline needed. +- Baseline hygiene: `detectAnomalies` measures BEFORE pushing, so a sample never pollutes its own reference; history rings cap at `maxHistory` 100 in `anomaly-baselines.json` (schema `phase-anomaly-baselines` 1.0.0). +- `routeAnomalyAlerts` fans findings out through `sendWebhookAlert` with titles `Anomaly : ` — distinct per kind+metric so the 15-minute `type:title` dedup cannot mask a second metric; attempts count `anomaly_alerts_routed`. +- `GET /api/cron/anomaly-detect`: hourly at minute 5, `validateCronAuth` + preview fence + restore marker, then `collectAnomalySample` (volume SQL, counter snapshot, latest distributor health balances) → `detectAnomalies` → persist → route alerts → respond with findings. +- Counters: `anomaly_checks_run`, `anomaly_volume_breaches`, `anomaly_counter_breaches`, `anomaly_balance_breaches`, `anomaly_alerts_routed`. +- Not covered: per-channel alert thresholds, alert suppression windows beyond the webhook dedup, and a volume baseline warmer than one value per run. diff --git a/docs/spikes/616-leaderboard-integrity-findings.md b/docs/spikes/616-leaderboard-integrity-findings.md new file mode 100644 index 00000000..c10daed4 --- /dev/null +++ b/docs/spikes/616-leaderboard-integrity-findings.md @@ -0,0 +1,30 @@ +# Spike 616: Leaderboard Integrity — Anti-Farm Scoring with Decay and Sybil Discount + +**Issue**: #479 +**Type**: Advanced Integration Spike +**Feature Flag**: `phase-178` (`NEXT_PUBLIC_FEATURE_PHASE_178` / `FEATURE_PHASE_178`) +**Precedents Analyzed**: sybil trust bands (phase-145), wash-trade detectors (phase-77), pricing half-life decay, quest-arc farmable caps + +## Context & Motivation + +PHASE had no leaderboard, and any naive points table would have been gameable three ways at once: burst campaigns (post a year of points in an hour), counterparty concentration (ring-trading one partner), and fresh sybil wallets farming at full value. Existing anti-abuse modules scored wallets (`lib/sybil-resistance.ts`) and flagged wash patterns (`lib/wash-trading.ts`), but nothing fed those signals into a ranking, and nothing decayed stale activity. + +## Key Architectural Decisions + +1. **Pure multiplicative scoring (`lib/leaderboard-integrity.ts`)**: + - `finalScore = decayedPoints × farmPenalty × sybilDiscount`, every stage unit-testable with an explicit `now`. + - Decay is exponential half-life (`0.5 ** (age / 7d)`), matching `lib/pricing-oracle.ts`'s wash discount so table behavior is consistent across modules. + +2. **Anti-farm signals with a floor**: + - Three rules — `event_burst` (×0.5), `counterparty_concentration` (×0.6), `point_velocity` (×0.7) — multiply down but stop at `farmPenaltyFloor` (0.15): the goal is to compress campaign ROI, not to make one rule irreversible. + +3. **Sybil discount from existing bands**: + - Callers pass the band produced by `scoreWalletHistory()`; `trusted` keeps 100%, `caution` 70%, `suspect` 35%. A missing band resolves to `caution`, so an unscored wallet can never score at trusted rate. + +4. **Storage and routes**: + - Entries live in the `leaderboard-entries.json` sidecar (registered in `lib/server-data-paths.ts`). + - `GET /api/leaderboard?limit=` returns the ranked top list; `POST /api/leaderboard` is admin-keyed and scores before persisting so malformed events never reach the store. + +## Rollback + +Unset `NEXT_PUBLIC_FEATURE_PHASE_178` / `FEATURE_PHASE_178` or set to `0`. Routes return `404`; `leaderboard-entries.json` stays inert — no migration to reverse. diff --git a/docs/spikes/617-retention-gdpr-erasure-findings.md b/docs/spikes/617-retention-gdpr-erasure-findings.md new file mode 100644 index 00000000..356f5ff8 --- /dev/null +++ b/docs/spikes/617-retention-gdpr-erasure-findings.md @@ -0,0 +1,32 @@ +# Spike 617: Data Retention and GDPR Erasure Pipeline with TTL Sweeps and Wallet Deletion + +**Issue**: #480 +**Type**: Advanced Integration Spike +**Feature Flag**: `phase-179` (`NEXT_PUBLIC_FEATURE_PHASE_179` / `FEATURE_PHASE_179`) +**Precedents Analyzed**: Article 20 portability export (phase-167), legal-hold deny-delete (phase-169), generation-job TTL prunes, SEP-10 sessions + +## Context & Motivation + +GDPR Article 20 (portability) shipped in spike 634, but Article 17 (erasure) did not exist: nothing deleted a wallet's records across stores, and no scheduled sweep expired time-shaped data — SEP-10 sessions, consumed challenges and notifications accumulated until something else rewrote them. Deletion also had to coexist with the phase-169 legal hold, which exists precisely to block destructive operations during investigations. + +## Key Architectural Decisions + +1. **Two halves, one flag (`lib/retention-erasure.ts`)**: + - **TTL sweeps** — `sweepByTimestamp` (pure, fail-safe: records without a parseable timestamp are kept), `sweepAuthSessions` (expired SEP-10 sessions in seconds or ms, consumed challenges past 24h), `sweepNotificationStore` (90-day notifications, dry wallet buckets removed), orchestrated by `runRetentionSweeps()`. + - **Wallet erasure** — `executeWalletErasure()` walks five registered domains (`notifications`, `age_declarations`, `follows`, `push_subscriptions`, `auth_sessions`), counts removed records per domain, and appends a receipt to `gdpr-erasure-log.json`. + +2. **Legal hold gates every erasure**: + - `assertNoLegalHold("wallet", …)` runs before any write; a hit throws `LEGAL_HOLD_ACTIVE` (HTTP 423 via the route) and increments `gdpr_erasure_blocked` — held data survives the attempt verbatim. + +3. **Receipts are minimized**: + - The receipt stores `wallet_fingerprint` (`fingerprintWallet()`), never the raw address; the log is capped at 1000 receipts and versioned `phase-gdpr-article-17-erasure` / `1.0.0`. + +4. **Deliberate exclusions**: + - The terms-consent ledger is NOT erased: consent records prove the legal basis kept under Art. 17(3)(b); withdrawal is recorded, not shredded. + - SQLite tables (signals, replies, listings, offers, tax events) are a documented partial slice — erasure covers JSON sidecars only. + +5. **Cron**: `GET /api/cron/retention-sweep` runs daily at 03:15 UTC behind `validateCronAuth`, the preview cron fence and the phase-171 restore marker (sweeping mid-restore would fight the restore write). + +## Rollback + +Unset `NEXT_PUBLIC_FEATURE_PHASE_179` / `FEATURE_PHASE_179` or set to `0`. The erasure route and sweep cron return `404`; stored receipts stay inert. diff --git a/docs/spikes/618-analytics-warehouse-export-findings.md b/docs/spikes/618-analytics-warehouse-export-findings.md new file mode 100644 index 00000000..f38d33b0 --- /dev/null +++ b/docs/spikes/618-analytics-warehouse-export-findings.md @@ -0,0 +1,29 @@ +# Spike 618: Analytics Warehouse Export with Parquet and NDJSON, Schema-Versioned and Redacted + +**Issue**: #481 +**Type**: Advanced Integration Spike +**Feature Flag**: `phase-180` (`NEXT_PUBLIC_FEATURE_PHASE_180` / `FEATURE_PHASE_180`) +**Precedents Analyzed**: world NDJSON streaming (phase-112), moderation audit export with role redaction, Article 20 checksummed bundles, log-pipeline redaction + +## Context & Motivation + +PHASE had export paths for individual domains (portability bundles, moderation audits, tax CSV, world markdown) but no aggregate analytics extract, no columnar format for warehouse loaders, and no single schema version consumers could pin. Analysts were left scraping sidecars. The repo also had zero Parquet support and no data-frame dependency. + +## Key Architectural Decisions + +1. **Flat row model (`lib/analytics-warehouse.ts`)**: + - One shape for every source: `table` (`counter` | `health` | `volume`), `ts` (epoch ms), `metric`, `value`, `status`, `wallet_fingerprint`. + - Collectors: security-counter snapshot, distributor health history, and wash-excluded accepted-offer volume from SQLite (the same `wash_flagged = 0` filter `lib/wash-screening.ts` uses so warehouse volume matches explore-ranking volume). + +2. **One manifest, two formats**: + - NDJSON: manifest line first (`schema`, `version`, `row_count`, `tables`, `checksum`), then one row per line — the shape `streamWorldExportNdjson` established. + - Parquet: `hyparquet-writer` (the repo's first data-format dependency, chosen over hand-rolling Thrift page encoding) writes typed columns with the same manifest in `key_value_metadata` (`phase.schema_version`, `phase.checksum`, `phase.redacted`). + - The checksum (`warehouseRowsChecksum`) hashes rows only, so it is reproducible regardless of format; it is mirrored in the `X-Warehouse-Checksum` response header. + +3. **Redaction is unconditional**: + - Exports never contain raw `G...` addresses or `S...` seeds: `redactWarehouseRow` fingerprints wallet fields and rewrites wallet/secret-looking text inside metric and status (idempotent — already-fingerprinted values pass through unchanged). + - `GET /api/analytics/export?format=ndjson|parquet` is admin-keyed (`checkAdminKey`, fail-closed) and returns `Content-Disposition` downloads. + +## Rollback + +Unset `NEXT_PUBLIC_FEATURE_PHASE_180` / `FEATURE_PHASE_180` or set to `0`. The export route returns `404`; no stored warehouse state exists — collectors read live sources only. diff --git a/docs/spikes/619-anomaly-detection-findings.md b/docs/spikes/619-anomaly-detection-findings.md new file mode 100644 index 00000000..1f2c1c86 --- /dev/null +++ b/docs/spikes/619-anomaly-detection-findings.md @@ -0,0 +1,29 @@ +# Spike 619: Anomaly Detection on Volume, Counters and Balances with Alert Routing + +**Issue**: #482 +**Type**: Advanced Integration Spike +**Feature Flag**: `phase-181` (`NEXT_PUBLIC_FEATURE_PHASE_181` / `FEATURE_PHASE_181`) +**Precedents Analyzed**: faucet-health cron with webhook fan-out, distributor-health ring buffer, wash-trade scoring thresholds, security-counter registry + +## Context & Motivation + +PHASE already emitted the raw signals — market volume, process counters, distributor balances — and already had an alert transport (`lib/webhook-alerts.ts`: Discord/Telegram/Slack/generic with a 15-minute dedup), but nothing compared current values to history. A volume pump, a counter spinning out of control or a balance draining would only be noticed by a human reading dashboards. + +## Key Architectural Decisions + +1. **Three detectors matched to three signal shapes (`lib/anomaly-detection.ts`)**: + - **volume** — robust z-score (median + MAD × 1.4826, falling back to standard deviation, then a tiny sigma floor so a flat baseline with a non-zero value still breaches) against persisted per-metric history; `z ≥ 3.5` warning, `≥ 6` critical. + - **counters** — cumulative snapshots reduced to per-run deltas first (`counterDeltas`; resets and first-seen keys skipped), then z-scored. Anomaly's own counters are excluded from the sample so the detector can never alert on itself. + - **balances** — deterministic percentage drop vs the previous run: `≥ 15%` warning, `≥ 35%` critical. A drain needs no baseline. + +2. **Detection before self-poisoning**: + - Each sample is measured against the baseline and only then pushed into it, so a spike never becomes its own reference. History is a capped ring (`maxHistory`, default 100) persisted in `anomaly-baselines.json`. + +3. **Alert routing**: + - Findings map to webhook alerts with a distinct `title` per kind+metric (dedup keys on `type:title`, so one flapping metric cannot suppress another) and route through `sendWebhookAlert`; attempts count into `anomaly_alerts_routed`. + +4. **Cron**: `GET /api/cron/anomaly-detect` runs hourly (minute 5) behind `validateCronAuth`, the preview cron fence and the phase-171 restore marker — baselines built on half-restored data would fire false alerts and poison history. + +## Rollback + +Unset `NEXT_PUBLIC_FEATURE_PHASE_181` / `FEATURE_PHASE_181` or set to `0`. The cron returns `404`; `anomaly-baselines.json` stays inert and no alerts are routed. diff --git a/lib/__tests__/analytics-warehouse.test.ts b/lib/__tests__/analytics-warehouse.test.ts new file mode 100644 index 00000000..9d58d87a --- /dev/null +++ b/lib/__tests__/analytics-warehouse.test.ts @@ -0,0 +1,210 @@ +import { describe, it, beforeEach, afterEach, vi } from "vitest" +import * as assert from "node:assert/strict" +import { mkdtempSync, rmSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { parquetMetadata, parquetReadObjects } from "hyparquet" +import { + WAREHOUSE_SCHEMA, + WAREHOUSE_SCHEMA_VERSION, + buildWarehouseManifest, + collectCounterRows, + collectVolumeRows, + collectWarehouseRows, + formatWarehouseNdjson, + isWarehouseExportEnabled, + redactWarehouseRow, + redactWarehouseRows, + redactWarehouseText, + rowsToParquetBuffer, + warehouseRowsChecksum, + type WarehouseRow, +} from "@/lib/analytics-warehouse" +import { getDb, resetDbForTests } from "@/lib/sqlite-db" +import { fingerprintWallet } from "@/lib/redaction" +import { resetSecurityCounters, incSecurityCounter } from "@/lib/security-counters" + +const WALLET = `G${"A".repeat(55)}` +const SECRET = "SDQXX5YXNQJ6X4Y3MXBXL4SVGSJPBIEUT3LOEBJTAI7XTMSEFAPKKMEJ" +const NOW = Date.parse("2026-06-01T00:00:00.000Z") + +function enable(on: boolean) { + if (on) { + process.env.NEXT_PUBLIC_FEATURE_PHASE_180 = "1" + process.env.FEATURE_PHASE_180 = "1" + } else { + delete process.env.NEXT_PUBLIC_FEATURE_PHASE_180 + delete process.env.FEATURE_PHASE_180 + } +} + +function row(over: Partial = {}): WarehouseRow { + return { + table: "counter", + ts: NOW, + metric: "wash_trades_flagged", + value: 3, + status: null, + wallet_fingerprint: null, + ...over, + } +} + +let dir: string + +beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "phase-warehouse-")) + vi.stubEnv("PHASE_SERVER_DATA_DIR", dir) + resetDbForTests() + resetSecurityCounters() + enable(true) +}) + +afterEach(() => { + enable(false) + resetDbForTests() + vi.unstubAllEnvs() + rmSync(dir, { recursive: true, force: true }) +}) + +describe("phase-180 warehouse redaction", () => { + it("gates on the feature flag and ships a rollback note", () => { + assert.equal(isWarehouseExportEnabled(), true) + enable(false) + assert.equal(isWarehouseExportEnabled(), false) + }) + + it("fingerprints wallets and strips secrets from text", () => { + const text = redactWarehouseText(`payout ${WALLET} signed with ${SECRET}`) + assert.ok(!text.includes(WALLET)) + assert.ok(!text.includes(SECRET)) + assert.ok(text.includes(fingerprintWallet(WALLET))) + assert.ok(text.includes("[REDACTED_SECRET_KEY]")) + }) + + it("fingerprints raw wallet fields and is idempotent across runs", () => { + const raw = row({ table: "health", metric: `health for ${WALLET}`, wallet_fingerprint: WALLET }) + const once = redactWarehouseRow(raw) + const twice = redactWarehouseRow(once) + assert.equal(once.wallet_fingerprint, fingerprintWallet(WALLET)) + assert.ok(!once.metric.includes(WALLET)) + assert.equal(twice.wallet_fingerprint, once.wallet_fingerprint) + assert.equal(twice.metric, once.metric) + + const rows = redactWarehouseRows([raw, row({ value: null })]) + assert.equal(rows.length, 2) + assert.equal(rows[1].wallet_fingerprint, null) + }) +}) + +describe("phase-180 warehouse formats", () => { + it("builds a schema-versioned manifest with a stable rows-only checksum", () => { + const rows = [row(), row({ table: "volume", metric: "market_volume_phaselq_24h", value: 100 })] + const manifest = buildWarehouseManifest(rows, { generatedAt: NOW }) + assert.equal(manifest.schema, WAREHOUSE_SCHEMA) + assert.equal(manifest.version, WAREHOUSE_SCHEMA_VERSION) + assert.equal(manifest.row_count, 2) + assert.deepEqual(manifest.tables, ["counter", "volume"]) + assert.equal(manifest.redacted, true) + assert.equal(manifest.checksum, warehouseRowsChecksum(rows)) + assert.equal(manifest.checksum, buildWarehouseManifest(rows, { generatedAt: NOW + 1 }).checksum) + assert.notEqual(manifest.checksum, warehouseRowsChecksum([row()])) + }) + + it("emits NDJSON with a manifest header line and one row per line", () => { + const rows = redactWarehouseRows([row(), row({ metric: "distributor_health", status: "healthy", wallet_fingerprint: WALLET })]) + const { manifest, body } = formatWarehouseNdjson(rows, { generatedAt: NOW }) + const lines = body.split("\n").filter((line) => line.length > 0) + assert.equal(lines.length, 3) + const header = JSON.parse(lines[0]) + assert.equal(header.schema, WAREHOUSE_SCHEMA) + assert.equal(header.checksum, manifest.checksum) + assert.equal(header.row_count, 2) + const parsed = lines.slice(1).map((line) => JSON.parse(line)) + assert.equal(parsed.length, 2) + assert.equal(parsed[0].table, "counter") + assert.ok(!lines.join("\n").includes(WALLET)) + }) + + it("emits empty-safe NDJSON for a zero-row export", () => { + const { manifest, body } = formatWarehouseNdjson([], { generatedAt: NOW }) + assert.equal(manifest.row_count, 0) + assert.equal(body.trim().split("\n").length, 1) + }) + + it("round-trips rows through Parquet with schema metadata", async () => { + const rows = redactWarehouseRows([ + row(), + row({ table: "health", metric: "distributor_health", value: null, status: "healthy", wallet_fingerprint: WALLET, ts: NOW + 5 }), + row({ table: "volume", metric: "market_offers_accepted_24h", value: 12, ts: NOW + 9 }), + ]) + const { manifest, buffer } = rowsToParquetBuffer(rows, { generatedAt: NOW }) + + const bytes = new Uint8Array(buffer) + assert.equal(String.fromCharCode(bytes[0], bytes[1], bytes[2], bytes[3]), "PAR1") + assert.equal( + String.fromCharCode( + bytes[bytes.length - 4], + bytes[bytes.length - 3], + bytes[bytes.length - 2], + bytes[bytes.length - 1], + ), + "PAR1", + ) + + const meta = parquetMetadata(buffer) + const kv = Object.fromEntries((meta.key_value_metadata ?? []).map((entry) => [entry.key, entry.value])) + assert.equal(kv["phase.schema"], WAREHOUSE_SCHEMA) + assert.equal(kv["phase.schema_version"], WAREHOUSE_SCHEMA_VERSION) + assert.equal(kv["phase.checksum"], manifest.checksum) + assert.equal(kv["phase.redacted"], "true") + + const parsed = await parquetReadObjects({ file: buffer }) + assert.equal(parsed.length, 3) + assert.equal(parsed[0].table, "counter") + assert.equal(Number(parsed[1].ts), NOW + 5) + assert.equal(parsed[1].value, null) + assert.equal(parsed[1].status, "healthy") + assert.equal(parsed[2].metric, "market_offers_accepted_24h") + assert.equal(parsed[2].value, 12) + const serialized = JSON.stringify(parsed, (key, value) => + typeof value === "bigint" ? String(value) : value, + ) + assert.ok(!serialized.includes(WALLET)) + }) +}) + +describe("phase-180 warehouse collectors", () => { + it("collects counter rows from the security counter snapshot", () => { + incSecurityCounter("wash_trades_flagged", 4) + const rows = collectCounterRows(NOW) + const flagged = rows.find((r) => r.metric === "wash_trades_flagged") + assert.ok(flagged) + assert.equal(flagged.table, "counter") + assert.equal(flagged.value, 4) + assert.equal(flagged.ts, NOW) + assert.equal(flagged.wallet_fingerprint, null) + }) + + it("collects zero volume from an empty database and skips a broken one", async () => { + const rows = await collectVolumeRows({ now: NOW }) + assert.equal(rows.length, 2) + assert.equal(rows.find((r) => r.metric === "market_volume_phaselq_24h")?.value, 0) + assert.equal(rows.find((r) => r.metric === "market_offers_accepted_24h")?.value, 0) + + const db = getDb() + db.exec("DROP TABLE offers") + const afterDrop = await collectVolumeRows({ now: NOW }) + assert.deepEqual(afterDrop, []) + }) + + it("collects counter, health and volume tables in one pass", async () => { + incSecurityCounter("preview_verify_total") + const rows = await collectWarehouseRows({ now: NOW }) + const tables = [...new Set(rows.map((r) => r.table))].sort() + assert.deepEqual(tables, ["counter", "volume"]) + assert.ok(rows.some((r) => r.table === "counter" && r.metric === "preview_verify_total")) + const redacted = redactWarehouseRows(rows) + assert.equal(redacted.length, rows.length) + }) +}) diff --git a/lib/__tests__/anomaly-detection.test.ts b/lib/__tests__/anomaly-detection.test.ts new file mode 100644 index 00000000..1b82c130 --- /dev/null +++ b/lib/__tests__/anomaly-detection.test.ts @@ -0,0 +1,287 @@ +import { describe, it, beforeEach, afterEach, vi } from "vitest" +import * as assert from "node:assert/strict" +import { mkdtempSync, rmSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { + ANOMALY_DEFAULTS, + ANOMALY_BASELINE_SCHEMA, + collectAnomalySample, + counterDeltas, + detectAnomalies, + detectBalanceDrops, + detectValueFinding, + emptyBaselineStore, + isAnomalyDetectionEnabled, + anomaly181RollbackNote, + anomalyAlertFor, + normalizeBaselineStore, + robustStats, + routeAnomalyAlerts, + zScore, + type AnomalyFinding, +} from "@/lib/anomaly-detection" +import type { WebhookAlertPayload, WebhookAlertType } from "@/lib/webhook-alerts" +import { getDb, resetDbForTests } from "@/lib/sqlite-db" +import { resetSecurityCounters, incSecurityCounter, getSecurityCounter } from "@/lib/security-counters" + +const NOW = Date.parse("2026-06-01T00:00:00.000Z") +const HISTORY = [10, 10, 10, 10, 10, 10, 10, 10, 10, 10] + +function enable(on: boolean) { + if (on) { + process.env.NEXT_PUBLIC_FEATURE_PHASE_181 = "1" + process.env.FEATURE_PHASE_181 = "1" + } else { + delete process.env.NEXT_PUBLIC_FEATURE_PHASE_181 + delete process.env.FEATURE_PHASE_181 + } +} + +let dir: string + +beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "phase-anomaly-")) + vi.stubEnv("PHASE_SERVER_DATA_DIR", dir) + resetDbForTests() + resetSecurityCounters() + enable(true) +}) + +afterEach(() => { + enable(false) + resetDbForTests() + vi.unstubAllEnvs() + rmSync(dir, { recursive: true, force: true }) +}) + +describe("phase-181 anomaly statistics", () => { + it("gates on the feature flag and ships a rollback note", () => { + assert.equal(isAnomalyDetectionEnabled(), true) + enable(false) + assert.equal(isAnomalyDetectionEnabled(), false) + assert.match(anomaly181RollbackNote(), /phase-181/) + }) + + it("computes robust median/MAD stats with a sigma floor", () => { + const flat = robustStats(HISTORY) + assert.equal(flat.samples, 10) + assert.equal(flat.median, 10) + assert.ok(flat.sigma > 0) + + const varied = robustStats([10, 12, 11, 9, 10, 13, 8, 10, 11, 12]) + assert.ok(varied.sigma > 0 && varied.sigma < 10) + + assert.equal(robustStats([]).samples, 0) + }) + + it("scores a spike against a flat baseline and a matching value as zero", () => { + assert.ok(zScore(HISTORY, 500) >= ANOMALY_DEFAULTS.criticalZ) + assert.equal(zScore(HISTORY, 10), 0) + assert.equal(zScore([], 10), 0) + }) + + it("emits findings only past the sample floor and severity thresholds", () => { + assert.equal(detectValueFinding("volume", "m", HISTORY.slice(0, 3), 500), null) + assert.equal(detectValueFinding("volume", "m", HISTORY, 10), null) + + // Varied baseline (median 10, sigma≈1.48): 11 fits, 16 is a warning (z≈4.0), + // 500 is critical. + const fits = [10, 11, 9, 10, 12, 8, 10, 11, 9, 10] + assert.equal(detectValueFinding("volume", "m", fits, 11), null) + + const warning = detectValueFinding("volume", "m", fits, 16) + assert.ok(warning) + assert.equal(warning.severity, "warning") + assert.equal(warning.kind, "volume") + + const critical = detectValueFinding("counter", "m", fits, 500) + assert.ok(critical) + assert.equal(critical.severity, "critical") + assert.equal(critical.kind, "counter") + assert.match(critical.message, /counter anomaly on m/) + }) + + it("reduces counters to deltas and skips resets and first-seen keys", () => { + const deltas = counterDeltas( + { wash_volume_excluded: 100, market_volume_drift: 50, stale: 10 }, + { wash_volume_excluded: 130, market_volume_drift: 40, stale: 10, fresh: 999 }, + ) + assert.deepEqual(deltas, { wash_volume_excluded: 30, stale: 0 }) + }) + + it("rates balance drops by percentage with warning and critical tiers", () => { + assert.deepEqual(detectBalanceDrops({ distributor_xlm: 100 }, { distributor_xlm: 90 }), []) + assert.deepEqual(detectBalanceDrops({ distributor_xlm: 0 }, { distributor_xlm: 100 }), []) + + const warning = detectBalanceDrops({ distributor_xlm: 100 }, { distributor_xlm: 80 }) + assert.equal(warning.length, 1) + assert.equal(warning[0].severity, "warning") + assert.equal(warning[0].kind, "balance") + + const critical = detectBalanceDrops({ distributor_xlm: 100 }, { distributor_xlm: 60 }) + assert.equal(critical.length, 1) + assert.equal(critical[0].severity, "critical") + assert.match(critical[0].message, /dropped 40%/) + }) +}) + +describe("phase-181 detection runs", () => { + it("normalizes garbage baselines and ships an empty store", () => { + const store = normalizeBaselineStore({ metrics: "nope", counters_prev: null }) + assert.equal(store.schema, ANOMALY_BASELINE_SCHEMA) + assert.deepEqual(store.metrics, {}) + assert.deepEqual(normalizeBaselineStore(undefined).balances_prev, {}) + assert.deepEqual(emptyBaselineStore().metrics, {}) + }) + + it("detects a volume breach only after the baseline has enough samples", () => { + let baseline = emptyBaselineStore() + let outcome = detectAnomalies( + baseline, + { ts: NOW, volume: { market_volume_phaselq_24h: 100 }, counters: {}, balances: {} }, + ) + assert.deepEqual(outcome.findings, []) + baseline = outcome.baseline + + for (let i = 0; i < ANOMALY_DEFAULTS.minSamples; i++) { + outcome = detectAnomalies( + baseline, + { ts: NOW + i, volume: { market_volume_phaselq_24h: 100 }, counters: {}, balances: {} }, + ) + assert.deepEqual(outcome.findings, []) + baseline = outcome.baseline + } + assert.equal(baseline.metrics?.["volume:market_volume_phaselq_24h"]?.length, ANOMALY_DEFAULTS.minSamples + 1) + + outcome = detectAnomalies(baseline, { + ts: NOW + 99, + volume: { market_volume_phaselq_24h: 5000 }, + counters: {}, + balances: {}, + }) + const finding = outcome.findings.find((f) => f.kind === "volume") + assert.ok(finding, "expected a volume finding on the spike") + assert.equal(finding.value, 5000) + assert.equal(getSecurityCounter("anomaly_volume_breaches"), 1) + // Detection ran before the push, so the spike is now part of the history. + const history = outcome.baseline.metrics?.["volume:market_volume_phaselq_24h"] ?? [] + assert.equal(history[history.length - 1], 5000) + }) + + it("detects counter delta breaches on the second run and balances by drop", () => { + const first = detectAnomalies(emptyBaselineStore(), { + ts: NOW, + volume: {}, + counters: { wash_trades_flagged: 10 }, + balances: { distributor_xlm: 100 }, + }) + assert.deepEqual(first.findings, []) + assert.deepEqual(first.baseline.counters_prev, { wash_trades_flagged: 10 }) + + // Warm the counter-delta history with quiet runs, then spike the counter. + let baseline = first.baseline + for (let i = 0; i < ANOMALY_DEFAULTS.minSamples; i++) { + baseline = detectAnomalies(baseline, { + ts: NOW + i, + volume: {}, + counters: { wash_trades_flagged: 10 }, + balances: { distributor_xlm: 100 }, + }).baseline + } + const spike = detectAnomalies(baseline, { + ts: NOW + 50, + volume: {}, + counters: { wash_trades_flagged: 1000 }, + balances: { distributor_xlm: 55 }, + }) + const counterFinding = spike.findings.find((f) => f.kind === "counter") + assert.ok(counterFinding, "expected a counter-delta finding") + assert.equal(counterFinding.value, 990) + const balanceFinding = spike.findings.find((f) => f.kind === "balance") + assert.ok(balanceFinding, "expected a balance-drop finding") + assert.equal(balanceFinding.severity, "critical") + assert.equal(getSecurityCounter("anomaly_counter_breaches"), 1) + assert.equal(getSecurityCounter("anomaly_balance_breaches"), 1) + }) + + it("caps baseline history at maxHistory", () => { + let baseline = emptyBaselineStore() + for (let i = 0; i < 20; i++) { + baseline = detectAnomalies( + baseline, + { ts: NOW + i, volume: { m: i }, counters: {}, balances: {} }, + { maxHistory: 5, zThreshold: 9999 }, + ).baseline + } + assert.equal(baseline.metrics?.["volume:m"]?.length, 5) + }) +}) + +describe("phase-181 alert routing", () => { + const finding = (over: Partial = {}): AnomalyFinding => ({ + kind: "volume", + severity: "warning", + metric: "market_volume_phaselq_24h", + message: "volume anomaly on market_volume_phaselq_24h", + value: 5000, + z: 7.2, + ...over, + }) + + it("maps findings to webhook payloads with a distinct title per kind and metric", () => { + const warning = anomalyAlertFor(finding()) + assert.equal(warning.type, "warning") + assert.equal(warning.payload.title, "Anomaly volume: market_volume_phaselq_24h") + assert.equal(warning.payload.z, 7.2) + + const critical = anomalyAlertFor(finding({ kind: "balance", severity: "critical", metric: "distributor_xlm", z: null })) + assert.equal(critical.type, "critical") + assert.equal(critical.payload.title, "Anomaly balance: distributor_xlm") + assert.equal("z" in critical.payload, false) + }) + + it("routes every finding through the router and counts the attempts", async () => { + const seen: Array<{ type: string; title: string }> = [] + const router = async (type: WebhookAlertType, payload: WebhookAlertPayload) => { + seen.push({ type, title: payload.title }) + return { sent: ["discord"], failed: [] } + } + const result = await routeAnomalyAlerts( + [ + finding(), + finding({ kind: "counter", metric: "wash_trades_flagged" }), + finding({ kind: "balance", severity: "critical", metric: "distributor_xlm" }), + ], + router, + ) + assert.equal(result.routed, 3) + assert.equal(result.sent, 3) + assert.equal(seen.length, 3) + assert.equal(new Set(seen.map((s) => s.title)).size, 3, "titles must not collide across metrics") + assert.equal(seen.filter((s) => s.type === "critical").length, 1) + assert.equal(getSecurityCounter("anomaly_alerts_routed"), 3) + + assert.deepEqual(await routeAnomalyAlerts([], router), { routed: 0, sent: 0 }) + assert.equal(getSecurityCounter("anomaly_alerts_routed"), 3, "empty routing must not inflate the counter") + }) +}) + +describe("phase-181 sample collection", () => { + it("excludes anomaly's own counters so the detector cannot alert on itself", async () => { + incSecurityCounter("anomaly_checks_run") + incSecurityCounter("wash_trades_flagged") + const sample = await collectAnomalySample({ now: NOW }) + assert.equal("anomaly_checks_run" in sample.counters, false) + assert.equal(sample.counters.wash_trades_flagged, 1) + assert.equal(sample.ts, NOW) + assert.ok("market_volume_phaselq_24h" in sample.volume) + assert.ok(Number.isFinite(sample.volume.market_volume_phaselq_24h)) + }) + + it("tolerates an unreadable database by omitting volume metrics", async () => { + getDb().exec("DROP TABLE offers") + const sample = await collectAnomalySample({ now: NOW }) + assert.deepEqual(sample.volume, {}) + }) +}) diff --git a/lib/__tests__/leaderboard-integrity.test.ts b/lib/__tests__/leaderboard-integrity.test.ts new file mode 100644 index 00000000..85f5453a --- /dev/null +++ b/lib/__tests__/leaderboard-integrity.test.ts @@ -0,0 +1,181 @@ +import { describe, it, beforeEach, afterEach } from "vitest" +import * as assert from "node:assert/strict" +import { + decayFactor, + detectFarmPatterns, + isLeaderboardIntegrityEnabled, + leaderboard178RollbackNote, + rankLeaderboard, + scoreLeaderboardCandidate, + sybilDiscountFor, + topLeaderboard, + LeaderboardIntegrityError, + LEADERBOARD_DEFAULTS, + SYBIL_BAND_DISCOUNT, + type LeaderboardCandidate, +} from "@/lib/leaderboard-integrity" + +const NOW = Date.parse("2026-06-01T00:00:00.000Z") +const HOUR = 60 * 60 * 1000 +const DAY = 24 * HOUR +const WALLET = `G${"A".repeat(55)}` +const WHALE = `G${"B".repeat(55)}` +const OTHER_CP = `G${"C".repeat(55)}` + +function enable(on: boolean) { + if (on) { + process.env.NEXT_PUBLIC_FEATURE_PHASE_178 = "1" + process.env.FEATURE_PHASE_178 = "1" + } else { + delete process.env.NEXT_PUBLIC_FEATURE_PHASE_178 + delete process.env.FEATURE_PHASE_178 + } +} + +function candidate(over: Partial = {}): LeaderboardCandidate { + return { + wallet: WALLET, + events: [{ points: 10, at: NOW }], + ...over, + } +} + +describe("phase-178 leaderboard integrity", () => { + beforeEach(() => enable(true)) + afterEach(() => enable(false)) + + it("gates on the feature flag and ships a rollback note", () => { + assert.equal(isLeaderboardIntegrityEnabled(), true) + enable(false) + assert.equal(isLeaderboardIntegrityEnabled(), false) + assert.match(leaderboard178RollbackNote(), /phase-178/) + }) + + it("applies exponential half-life decay", () => { + const halfLife = LEADERBOARD_DEFAULTS.halfLifeMs + assert.equal(decayFactor(0, halfLife), 1) + assert.equal(decayFactor(halfLife, halfLife), 0.5) + assert.equal(decayFactor(2 * halfLife, halfLife), 0.25) + // Negative ages (clock skew) clamp to fresh, never boost. + assert.equal(decayFactor(-DAY, halfLife), 1) + }) + + it("discounts sybil bands and treats a missing band as caution", () => { + assert.equal(sybilDiscountFor("trusted"), 1) + assert.equal(sybilDiscountFor("caution"), 0.7) + assert.equal(sybilDiscountFor("suspect"), 0.35) + assert.equal(sybilDiscountFor(undefined), SYBIL_BAND_DISCOUNT.caution) + }) + + it("flags a burst of events inside one window", () => { + const events = Array.from({ length: 13 }, (_, i) => ({ points: 1, at: NOW - 30 * 60 * 1000 + i * 1000 })) + const farm = detectFarmPatterns({ events }) + assert.ok(farm.signals.includes("event_burst")) + assert.equal(farm.farmed, true) + assert.equal(farm.penalty, 0.5) + }) + + it("flags counterparty concentration when one wallet dominates", () => { + const events = Array.from({ length: 8 }, (_, i) => ({ + points: 5, + at: NOW - (i + 1) * 6 * HOUR, + counterparty: i < 7 ? WHALE : OTHER_CP, + })) + const farm = detectFarmPatterns({ events }) + assert.ok(farm.signals.includes("counterparty_concentration")) + assert.equal(farm.penalty, 0.6) + }) + + it("flags point velocity over the per-day cap", () => { + const events = [ + { points: 800, at: NOW - 2 * DAY }, + { points: 800, at: NOW - DAY }, + { points: 800, at: NOW }, + ] + const farm = detectFarmPatterns({ events }) + assert.ok(farm.signals.includes("point_velocity")) + assert.equal(farm.penalty, 0.7) + }) + + it("floors the combined farm penalty", () => { + const events = [ + ...Array.from({ length: 13 }, (_, i) => ({ points: 100, at: NOW - 30 * 60 * 1000 + i * 1000 })), + ...Array.from({ length: 13 }, (_, i) => ({ points: 100, at: NOW - 40 * 60 * 1000 + i * 1000 })), + ] + const farm = detectFarmPatterns({ events }, { farmPenaltyFloor: 0.9 }) + assert.ok(farm.signals.includes("event_burst")) + assert.equal(farm.penalty, 0.9) + }) + + it("scores decay, farm penalty and sybil discount multiplicatively", () => { + const score = scoreLeaderboardCandidate( + candidate({ + events: [ + { points: 50, at: NOW }, + { points: 50, at: NOW - LEADERBOARD_DEFAULTS.halfLifeMs }, + ], + sybilBand: "trusted", + }), + {}, + NOW, + ) + assert.equal(score.rawPoints, 100) + assert.equal(score.decayedPoints, 75) + assert.equal(score.farmPenalty, 1) + assert.equal(score.sybilDiscount, 1) + assert.equal(score.finalScore, 75) + + const suspect = scoreLeaderboardCandidate( + candidate({ + events: [ + { points: 50, at: NOW }, + { points: 50, at: NOW - LEADERBOARD_DEFAULTS.halfLifeMs }, + ], + sybilBand: "suspect", + }), + {}, + NOW, + ) + assert.equal(suspect.finalScore, 26.25) + assert.ok(suspect.signals.includes("sybil_suspect")) + + const unknownBand = scoreLeaderboardCandidate(candidate(), {}, NOW) + assert.equal(unknownBand.sybilDiscount, 0.7) + assert.ok(unknownBand.signals.includes("sybil_caution")) + }) + + it("multiplies farm penalty with the sybil discount", () => { + const events = Array.from({ length: 20 }, () => ({ points: 1, at: NOW })) + const score = scoreLeaderboardCandidate(candidate({ events, sybilBand: "suspect" }), {}, NOW) + assert.equal(score.farmPenalty, 0.5) + assert.equal(score.sybilDiscount, 0.35) + assert.equal(score.finalScore, 3.5) + assert.ok(score.signals.includes("event_burst")) + assert.ok(score.signals.includes("sybil_suspect")) + }) + + it("ranks descending with a stable wallet tie-break and caps the top list", () => { + const a = candidate({ wallet: "GAAAA", events: [{ points: 10, at: NOW }] }) + const b = candidate({ wallet: "GBBBB", events: [{ points: 10, at: NOW }] }) + const c = candidate({ wallet: "GCCCC", events: [{ points: 100, at: NOW }] }) + const ranked = rankLeaderboard([a, b, c], {}, NOW) + assert.deepEqual(ranked.map((s) => s.wallet), ["GCCCC", "GAAAA", "GBBBB"]) + assert.equal(topLeaderboard([a, b, c], 2, {}, NOW).length, 2) + assert.equal(topLeaderboard([a, b, c], 999, {}, NOW).length, 3) + }) + + it("rejects malformed candidates instead of guessing", () => { + assert.throws( + () => scoreLeaderboardCandidate(candidate({ wallet: "" })), + (err: unknown) => err instanceof LeaderboardIntegrityError && err.code === "VALIDATION_FAILED", + ) + assert.throws( + () => scoreLeaderboardCandidate(candidate({ events: [{ points: NaN, at: NOW }] })), + (err: unknown) => err instanceof LeaderboardIntegrityError && err.code === "VALIDATION_FAILED", + ) + assert.throws( + () => scoreLeaderboardCandidate({ ...candidate(), events: "nope" as never }), + (err: unknown) => err instanceof LeaderboardIntegrityError && err.code === "VALIDATION_FAILED", + ) + }) +}) diff --git a/lib/__tests__/retention-erasure.test.ts b/lib/__tests__/retention-erasure.test.ts new file mode 100644 index 00000000..a3c98296 --- /dev/null +++ b/lib/__tests__/retention-erasure.test.ts @@ -0,0 +1,264 @@ +import { describe, it, beforeEach, afterEach, vi } from "vitest" +import * as assert from "node:assert/strict" +import { mkdtempSync, rmSync, readFileSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { + ERASURE_TARGETS, + RETENTION_TTL_MS, + eraseWalletAuthSessions, + eraseWalletKeyedRecord, + executeWalletErasure, + isRetentionErasureEnabled, + retention179RollbackNote, + runRetentionSweeps, + sweepAuthSessions, + sweepByTimestamp, + sweepNotificationStore, + RetentionErasureError, +} from "@/lib/retention-erasure" +import { createLegalHold } from "@/lib/legal-hold" +import { readStore, writeJsonFileAtomic } from "@/lib/json-store" +import { serverDataJsonPath } from "@/lib/server-data-paths" +import { getSecurityCounter, resetSecurityCounters } from "@/lib/security-counters" + +const WALLET = `G${"A".repeat(55)}` +const OTHER = `G${"B".repeat(55)}` +const NOW = Date.parse("2026-06-01T00:00:00.000Z") +const DAY = 24 * 60 * 60 * 1000 + +function enable(on: boolean) { + if (on) { + process.env.NEXT_PUBLIC_FEATURE_PHASE_179 = "1" + process.env.FEATURE_PHASE_179 = "1" + } else { + delete process.env.NEXT_PUBLIC_FEATURE_PHASE_179 + delete process.env.FEATURE_PHASE_179 + } +} + +function enableLegalHold(on: boolean) { + if (on) { + process.env.NEXT_PUBLIC_FEATURE_PHASE_169 = "1" + process.env.FEATURE_PHASE_169 = "1" + } else { + delete process.env.NEXT_PUBLIC_FEATURE_PHASE_169 + delete process.env.FEATURE_PHASE_169 + } +} + +let dir: string + +beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "phase-retention-")) + vi.stubEnv("PHASE_SERVER_DATA_DIR", dir) + enable(true) + enableLegalHold(false) + resetSecurityCounters() +}) + +afterEach(() => { + enable(false) + enableLegalHold(false) + vi.unstubAllEnvs() + rmSync(dir, { recursive: true, force: true }) +}) + +describe("phase-179 TTL sweeps", () => { + it("gates on the feature flag and ships a rollback note", () => { + assert.equal(isRetentionErasureEnabled(), true) + enable(false) + assert.equal(isRetentionErasureEnabled(), false) + assert.match(retention179RollbackNote(), /phase-179/) + }) + + it("sweeps by timestamp and fails safe on missing timestamps", () => { + const { kept, removed } = sweepByTimestamp( + [ + { at: NOW - RETENTION_TTL_MS.notification - 1 }, + { at: NOW - 1000 }, + { at: null as unknown as number }, + { at: NaN }, + ], + { now: NOW, ttlMs: RETENTION_TTL_MS.notification, timestampOf: (r) => r.at }, + ) + assert.equal(removed.length, 1) + assert.equal(kept.length, 3) + }) + + it("drops expired SEP-10 sessions (seconds) and stale consumed challenges", () => { + const store = { + sessions: { + expired: { wallet: WALLET, expires_at: Math.floor(NOW / 1000) - 60 }, + fresh: { wallet: OTHER, expires_at: Math.floor(NOW / 1000) + 3600 }, + expired_ms: { wallet: WALLET, expires_at: NOW - 60_000 }, + }, + consumed_challenges: { + old: NOW / 1000 - 48 * 3600, + new: NOW / 1000 - 60, + }, + } + const removed = sweepAuthSessions(store, NOW) + assert.equal(removed, 3) + assert.deepEqual(Object.keys(store.sessions), ["fresh"]) + assert.deepEqual(Object.keys(store.consumed_challenges), ["new"]) + }) + + it("prunes expired notifications and empties dry wallet buckets", () => { + const store: Record> = { + [WALLET]: [{ created_at: NOW - RETENTION_TTL_MS.notification - 1 }, { created_at: NOW - DAY }], + [OTHER]: [{ created_at: NOW - RETENTION_TTL_MS.notification - 1 }], + untouched: [{ created_at: NOW - DAY }], + } + const removed = sweepNotificationStore(store, NOW) + assert.equal(removed, 2) + assert.deepEqual(store[WALLET], [{ created_at: NOW - DAY }]) + assert.equal(OTHER in store, false) + assert.equal("untouched" in store, true) + }) + + it("runs both sweeps through runRetentionSweeps and counts the removals", async () => { + await writeJsonFileAtomic(serverDataJsonPath("authSessions"), { + sessions: { a: { wallet: WALLET, expires_at: Math.floor(NOW / 1000) - 10 } }, + consumed_challenges: {}, + }) + await writeJsonFileAtomic(serverDataJsonPath("notifications"), { + [WALLET]: [{ created_at: NOW - RETENTION_TTL_MS.notification - 1 }], + }) + + const result = await runRetentionSweeps(NOW) + assert.equal(result.ok, true) + assert.equal(result.total_removed, 2) + assert.deepEqual( + result.sweeps.map((s) => s.store), + ["authSessions", "notifications"], + ) + assert.equal(getSecurityCounter("retention_sweep_runs"), 1) + assert.equal(getSecurityCounter("retention_records_expired"), 2) + + const auth = await readStore<{ sessions: Record }>("authSessions", () => ({ sessions: {} })) + assert.deepEqual(auth.sessions, {}) + const notifs = await readStore>("notifications", () => ({})) + assert.deepEqual(notifs, {}) + + enable(false) + await assert.rejects( + () => runRetentionSweeps(NOW), + (err: unknown) => err instanceof RetentionErasureError && err.code === "FLAG_DISABLED", + ) + }) +}) + +describe("phase-179 wallet erasure", () => { + it("removes only the target wallet from wallet-keyed stores", () => { + const store: Record = { + [WALLET]: [{ id: 1 }, { id: 2 }], + [OTHER]: [{ id: 3 }], + } + assert.equal(eraseWalletKeyedRecord(store, WALLET), 2) + assert.equal(WALLET in store, false) + assert.equal(OTHER in store, true) + assert.equal(eraseWalletKeyedRecord(store, "GNOPE"), 0) + }) + + it("removes only the target wallet's SEP-10 sessions", () => { + const store = { + sessions: { + t1: { wallet: WALLET }, + t2: { wallet: OTHER }, + t3: { wallet: WALLET }, + }, + } + assert.equal(eraseWalletAuthSessions(store, WALLET), 2) + assert.deepEqual(Object.keys(store.sessions), ["t2"]) + }) + + it("erases across every registered domain and writes a fingerprinted receipt", async () => { + await writeJsonFileAtomic(serverDataJsonPath("notifications"), { + [WALLET]: [{ id: "n1" }, { id: "n2" }], + [OTHER]: [{ id: "n3" }], + }) + await writeJsonFileAtomic(serverDataJsonPath("ageDeclarations"), { + [WALLET]: { band: "18+", declared_at: 1 }, + [OTHER]: { band: "21+", declared_at: 2 }, + }) + await writeJsonFileAtomic(serverDataJsonPath("profileFollows"), { + [WALLET]: { following: [] }, + [OTHER]: { following: [] }, + }) + await writeJsonFileAtomic(serverDataJsonPath("pushSubscriptions"), { + [WALLET]: [{ endpoint: "https://push.example/1" }], + [OTHER]: [{ endpoint: "https://push.example/2" }], + }) + await writeJsonFileAtomic(serverDataJsonPath("authSessions"), { + sessions: { mine: { wallet: WALLET }, theirs: { wallet: OTHER } }, + consumed_challenges: {}, + }) + + const receipt = await executeWalletErasure(WALLET, { initiatedBy: "wallet", now: NOW }) + + assert.equal(receipt.schema, "phase-gdpr-article-17-erasure") + assert.equal(receipt.version, "1.0.0") + assert.equal(receipt.total_removed, 6) + assert.equal(receipt.initiated_by, "wallet") + assert.ok(!receipt.wallet_fingerprint.includes(WALLET), "receipt must not contain the raw wallet") + assert.match(receipt.wallet_fingerprint, /fp: [0-9a-f]{8}/) + assert.deepEqual( + receipt.domains.map((d) => d.domain).sort(), + ERASURE_TARGETS.map((t) => t.domain).sort(), + ) + + const notifications = await readStore>("notifications", () => ({})) + assert.deepEqual(Object.keys(notifications), [OTHER]) + const auth = await readStore<{ sessions: Record }>("authSessions", () => ({ sessions: {} })) + assert.deepEqual(Object.keys(auth.sessions), ["theirs"]) + const age = await readStore>("ageDeclarations", () => ({})) + assert.deepEqual(Object.keys(age), [OTHER]) + + const logRaw = readFileSync(serverDataJsonPath("gdprErasureLog"), "utf8") + const log = JSON.parse(logRaw) as { receipts: Array<{ wallet_fingerprint: string }> } + assert.equal(log.receipts.length, 1) + assert.ok(!logRaw.includes(WALLET)) + + assert.equal(getSecurityCounter("gdpr_erasure_completed"), 1) + assert.equal(getSecurityCounter("gdpr_erasure_records"), 6) + + enable(false) + await assert.rejects( + () => executeWalletErasure(WALLET), + (err: unknown) => err instanceof RetentionErasureError && err.code === "FLAG_DISABLED", + ) + }) + + it("rejects a malformed wallet before touching any store", async () => { + await assert.rejects( + () => executeWalletErasure("not-a-wallet"), + (err: unknown) => err instanceof RetentionErasureError && err.code === "VALIDATION_FAILED", + ) + assert.equal(getSecurityCounter("gdpr_erasure_completed"), 0) + }) + + it("blocks erasure behind an active legal hold and records the denial", async () => { + await writeJsonFileAtomic(serverDataJsonPath("notifications"), { + [WALLET]: [{ id: "n1" }], + }) + enableLegalHold(true) + await createLegalHold({ + target_type: "wallet", + target_id: WALLET, + matter_id: "M-1", + reason: "fraud investigation", + custodian: "ops@phase.example", + }) + + await assert.rejects( + () => executeWalletErasure(WALLET), + (err: unknown) => err instanceof RetentionErasureError && err.code === "LEGAL_HOLD_ACTIVE", + ) + assert.equal(getSecurityCounter("gdpr_erasure_blocked"), 1) + assert.equal(getSecurityCounter("gdpr_erasure_completed"), 0) + + const notifications = await readStore>("notifications", () => ({})) + assert.ok(WALLET in notifications, "held data must survive the erasure attempt") + }) +}) diff --git a/lib/analytics-warehouse.ts b/lib/analytics-warehouse.ts new file mode 100644 index 00000000..0243704a --- /dev/null +++ b/lib/analytics-warehouse.ts @@ -0,0 +1,267 @@ +/** + * Analytics warehouse export: schema-versioned NDJSON and Parquet, redacted. + * Spike 618 / Issue #481. + * + * The warehouse is a flat, append-only row model (`table`, `ts`, `metric`, + * `value`, `status`, `wallet_fingerprint`) collected from three sources: + * process counters, distributor health history and wash-excluded market + * volume. Two output formats share one manifest: + * + * - NDJSON: first line is a manifest object (`schema`, `version`, + * `row_count`, `checksum`), then one row per line — the streaming shape + * `lib/narrative-world-store.ts` established. + * - Parquet: columnar buffer via `hyparquet-writer`, with the same manifest + * in `key_value_metadata` (`phase.schema_version`, `phase.checksum`, …). + * + * Redaction is unconditional: exports never contain raw `G...` addresses or + * `S...` seeds. Wallet fields become `fingerprintWallet()` strings and any + * wallet/secret-looking text inside metric or status is rewritten in place + * (`redactWarehouseRow`). Flag: phase-180. + * Rollback: unset the flag — the export route 404s and no stored state needs + * reversing; the Parquet/NDJSON helpers stay importable for tests. + */ + +import { parquetWriteBuffer } from "hyparquet-writer" +import { isFeatureEnabled } from "@/lib/feature-flags" +import { computeSha256 } from "@/lib/data-portability" +import { fingerprintWallet, redactSecretKeys, WALLET_KEY_RE } from "@/lib/redaction" +import { snapshotSecurityCounters } from "@/lib/security-counters" +import { getHealthHistory } from "@/lib/distributor-health-store" +import { getDb } from "@/lib/sqlite-db" + +export const WAREHOUSE_SCHEMA = "phase-warehouse-export" +export const WAREHOUSE_SCHEMA_VERSION = "1.0.0" + +export const WAREHOUSE_TABLES = ["counter", "health", "volume"] as const +export type WarehouseTable = (typeof WAREHOUSE_TABLES)[number] + +export type WarehouseRow = { + table: WarehouseTable + ts: number + metric: string + value: number | null + status: string | null + wallet_fingerprint: string | null +} + +export function isWarehouseExportEnabled(): boolean { + return isFeatureEnabled("phase-180") +} + +export function warehouse180RollbackNote(): string { + return "Rollback phase-180: unset NEXT_PUBLIC_FEATURE_PHASE_180 / FEATURE_PHASE_180 or set 0/false and restart. The export route 404s; no stored warehouse state exists." +} + +export class WarehouseExportError extends Error { + readonly code: "FLAG_DISABLED" | "UNAUTHORIZED" | "EXPORT_FAILED" + readonly details?: unknown + constructor(code: WarehouseExportError["code"], message: string, details?: unknown) { + super(message) + this.name = "WarehouseExportError" + this.code = code + this.details = details + } +} + +// ─── Redaction ─────────────────────────────────────────────────────────────── + +/** Fingerprints raw wallets and strips secrets from free text. */ +export function redactWarehouseText(text: string): string { + if (!text) return text + return redactSecretKeys(text).replace(WALLET_KEY_RE, (wallet) => fingerprintWallet(wallet)) +} + +/** + * Makes one row safe for export. `wallet_fingerprint` is only ever written by + * collectors as a raw address (the field name describes the post-redaction + * role), so a raw `G...` here is fingerprinted rather than trusted. A fresh + * non-global regex because `WALLET_KEY_RE` carries `/g` and `.test()` on it is + * stateful across calls. + */ +const FULL_WALLET_RE = /^G[A-Z2-7]{55}$/ + +export function redactWarehouseRow(row: WarehouseRow): WarehouseRow { + return { + ...row, + metric: redactWarehouseText(row.metric), + status: row.status === null ? null : redactWarehouseText(row.status), + wallet_fingerprint: + row.wallet_fingerprint === null + ? null + : FULL_WALLET_RE.test(row.wallet_fingerprint) + ? fingerprintWallet(row.wallet_fingerprint) + : redactWarehouseText(row.wallet_fingerprint), + } +} + +export function redactWarehouseRows(rows: WarehouseRow[]): WarehouseRow[] { + return rows.map(redactWarehouseRow) +} + +// ─── Collectors ───────────────────────────────────────────────────────────── + +export type CollectOptions = { + now?: number + healthLimit?: number + volumeWindowMs?: number +} + +export const DEFAULT_VOLUME_WINDOW_MS = 24 * 60 * 60 * 1000 + +export function collectCounterRows(now: number = Date.now()): WarehouseRow[] { + const snapshot = snapshotSecurityCounters() + return Object.entries(snapshot).map(([metric, value]) => ({ + table: "counter" as const, + ts: now, + metric, + value, + status: null, + wallet_fingerprint: null, + })) +} + +export async function collectHealthRows( + opts: CollectOptions = {}, +): Promise { + const history = await getHealthHistory(opts.healthLimit ?? 50) + return history.map((record) => ({ + table: "health" as const, + ts: record.checkedAt, + metric: "distributor_health", + value: record.distributorXlm, + status: record.status, + wallet_fingerprint: record.distributorAddress || null, + })).concat( + history.map((record) => ({ + table: "health" as const, + ts: record.checkedAt, + metric: "distributor_phase_liq_stroops", + value: record.distributorPhaseLiqStroops === null ? null : Number(record.distributorPhaseLiqStroops), + status: record.status, + wallet_fingerprint: record.issuerAddress || null, + })), + ) +} + +export async function collectVolumeRows( + opts: CollectOptions = {}, +): Promise { + const now = opts.now ?? Date.now() + const since = now - (opts.volumeWindowMs ?? DEFAULT_VOLUME_WINDOW_MS) + try { + const row = getDb() + .prepare( + `SELECT COALESCE(SUM(amount_phaselq), 0) AS total, COUNT(*) AS offers + FROM offers + WHERE status = 'accepted' AND wash_flagged = 0 AND created_at >= ?`, + ) + .get(since) as { total: number; offers: number } + return [ + { + table: "volume", + ts: now, + metric: "market_volume_phaselq_24h", + value: Number(row?.total ?? 0), + status: null, + wallet_fingerprint: null, + }, + { + table: "volume", + ts: now, + metric: "market_offers_accepted_24h", + value: Number(row?.offers ?? 0), + status: null, + wallet_fingerprint: null, + }, + ] + } catch { + // A missing/unmigrated database yields no rows rather than failing the export. + return [] + } +} + +export async function collectWarehouseRows(opts: CollectOptions = {}): Promise { + const now = opts.now ?? Date.now() + const [health, volume] = await Promise.all([collectHealthRows(opts), collectVolumeRows(opts)]) + return [...collectCounterRows(now), ...health, ...volume] +} + +// ─── Manifest + formats ───────────────────────────────────────────────────── + +export type WarehouseManifest = { + schema: typeof WAREHOUSE_SCHEMA + version: string + generated_at: number + row_count: number + tables: string[] + redacted: boolean + checksum: string +} + +/** Rows only — the checksum must be reproducible without the manifest line. */ +export function warehouseRowsChecksum(rows: WarehouseRow[]): string { + return computeSha256(rows.map((row) => JSON.stringify(row)).join("\n")) +} + +export function buildWarehouseManifest( + rows: WarehouseRow[], + opts: { generatedAt?: number; redacted?: boolean } = {}, +): WarehouseManifest { + const tables = [...new Set(rows.map((row) => row.table))].sort() + return { + schema: WAREHOUSE_SCHEMA, + version: WAREHOUSE_SCHEMA_VERSION, + generated_at: opts.generatedAt ?? Date.now(), + row_count: rows.length, + tables, + redacted: opts.redacted ?? true, + checksum: warehouseRowsChecksum(rows), + } +} + +export function formatWarehouseNdjson( + rows: WarehouseRow[], + opts: { generatedAt?: number } = {}, +): { manifest: WarehouseManifest; body: string } { + const manifest = buildWarehouseManifest(rows, { generatedAt: opts.generatedAt, redacted: true }) + const body = + JSON.stringify(manifest) + + "\n" + + rows.map((row) => JSON.stringify(row)).join("\n") + + (rows.length > 0 ? "\n" : "") + return { manifest, body } +} + +export function rowsToParquetBuffer( + rows: WarehouseRow[], + opts: { generatedAt?: number } = {}, +): { manifest: WarehouseManifest; buffer: ArrayBuffer } { + const manifest = buildWarehouseManifest(rows, { generatedAt: opts.generatedAt, redacted: true }) + try { + const buffer = parquetWriteBuffer({ + columnData: [ + { name: "table", data: rows.map((row) => row.table), type: "STRING" }, + { name: "ts", data: rows.map((row) => BigInt(row.ts)), type: "INT64" }, + { name: "metric", data: rows.map((row) => row.metric), type: "STRING" }, + { name: "value", data: rows.map((row) => row.value), type: "DOUBLE" }, + { name: "status", data: rows.map((row) => row.status), type: "STRING" }, + { name: "wallet_fingerprint", data: rows.map((row) => row.wallet_fingerprint), type: "STRING" }, + ], + kvMetadata: [ + { key: "phase.schema", value: manifest.schema }, + { key: "phase.schema_version", value: manifest.version }, + { key: "phase.generated_at", value: String(manifest.generated_at) }, + { key: "phase.row_count", value: String(manifest.row_count) }, + { key: "phase.checksum", value: manifest.checksum }, + { key: "phase.redacted", value: "true" }, + ], + }) + return { manifest, buffer } + } catch (error) { + throw new WarehouseExportError( + "EXPORT_FAILED", + "Parquet encoding failed", + error instanceof Error ? error.message : String(error), + ) + } +} diff --git a/lib/anomaly-detection.ts b/lib/anomaly-detection.ts new file mode 100644 index 00000000..d3911d94 --- /dev/null +++ b/lib/anomaly-detection.ts @@ -0,0 +1,374 @@ +/** + * Anomaly detection on volume, counters and balances with alert routing. + * Spike 619 / Issue #482. + * + * Three detectors over one sample (`AnomalySample`), each matched to how its + * signal actually behaves: + * + * - volume — robust z-score (median + MAD, falling back to standard + * deviation, with a tiny sigma floor so a constant baseline and + * a sudden non-zero value still breach) against the persisted + * per-metric history. + * - counters — cumulative snapshots are reduced to per-run deltas first + * (a reset, `curr < prev`, is skipped), then z-scored like + * volume. Anomaly's own counters are excluded from the sample + * so the detector never alerts on itself. + * - balances — deterministic drop percentage vs the previous run: ≥35% is + * critical, ≥15% is warning. No baseline needed; a drain is a + * drain. + * + * Findings route through `sendWebhookAlert` (Discord/Telegram/Slack/generic, + * 15-minute title dedup) via `routeAnomalyAlerts`, with a distinct title per + * kind+metric so one flapping metric cannot suppress another. + * + * Baselines persist in the `anomalyBaselines` sidecar (capped ring buffer). + * Flag: phase-181 — rollback: unset the flag, cron returns 404, stored + * baselines stay inert. + */ + +import { isFeatureEnabled } from "@/lib/feature-flags" +import { sendWebhookAlert, type WebhookAlertPayload, type WebhookAlertType } from "@/lib/webhook-alerts" +import { incSecurityCounter, snapshotSecurityCounters } from "@/lib/security-counters" +import { getHealthHistory } from "@/lib/distributor-health-store" +import { getDb } from "@/lib/sqlite-db" + +export const ANOMALY_BASELINE_SCHEMA = "phase-anomaly-baselines" +export const ANOMALY_BASELINE_VERSION = "1.0.0" + +export const ANOMALY_DEFAULTS = { + zThreshold: 3.5, + criticalZ: 6, + minSamples: 8, + maxHistory: 100, + balanceDropWarnPct: 15, + balanceDropCriticalPct: 35, + volumeWindowMs: 24 * 60 * 60 * 1000, +} as const + +export function isAnomalyDetectionEnabled(): boolean { + return isFeatureEnabled("phase-181") +} + +export function anomaly181RollbackNote(): string { + return "Rollback phase-181: unset NEXT_PUBLIC_FEATURE_PHASE_181 / FEATURE_PHASE_181 or set 0/false and restart. The anomaly cron 404s; anomaly-baselines.json stays inert." +} + +export class AnomalyDetectionError extends Error { + readonly code: "FLAG_DISABLED" | "COLLECTION_FAILED" + readonly details?: unknown + constructor(code: AnomalyDetectionError["code"], message: string, details?: unknown) { + super(message) + this.name = "AnomalyDetectionError" + this.code = code + this.details = details + } +} + +export type AnomalyKind = "volume" | "counter" | "balance" +export type AnomalySeverity = "warning" | "critical" + +export type AnomalyFinding = { + kind: AnomalyKind + severity: AnomalySeverity + metric: string + message: string + value: number + z: number | null +} + +export type AnomalySample = { + ts: number + volume: Record + counters: Record + balances: Record +} + +export type BaselineStore = { + schema?: string + version?: string + metrics?: Record + counters_prev?: Record + balances_prev?: Record +} + +export function emptyBaselineStore(): BaselineStore { + return { + schema: ANOMALY_BASELINE_SCHEMA, + version: ANOMALY_BASELINE_VERSION, + metrics: {}, + counters_prev: {}, + balances_prev: {}, + } +} + +export function normalizeBaselineStore(raw: unknown): BaselineStore { + const parsed = (raw ?? {}) as Partial + return { + schema: ANOMALY_BASELINE_SCHEMA, + version: ANOMALY_BASELINE_VERSION, + metrics: parsed.metrics && typeof parsed.metrics === "object" ? parsed.metrics : {}, + counters_prev: parsed.counters_prev && typeof parsed.counters_prev === "object" ? parsed.counters_prev : {}, + balances_prev: parsed.balances_prev && typeof parsed.balances_prev === "object" ? parsed.balances_prev : {}, + } +} + +// ─── Statistics (pure) ─────────────────────────────────────────────────────── + +export type RobustStats = { + samples: number + mean: number + median: number + sigma: number +} + +function median(values: number[]): number { + const sorted = [...values].sort((a, b) => a - b) + const mid = Math.floor(sorted.length / 2) + return sorted.length % 2 === 0 ? (sorted[mid - 1] + sorted[mid]) / 2 : sorted[mid] +} + +/** + * Median + MAD scaled by 1.4826 (robust to a single spike already in the + * history), falling back to standard deviation when MAD is 0, then to a tiny + * sigma floor so a flat baseline with a non-zero current value still scores. + */ +export function robustStats(history: number[]): RobustStats { + if (history.length === 0) return { samples: 0, mean: 0, median: 0, sigma: 0 } + const mean = history.reduce((sum, v) => sum + v, 0) / history.length + const med = median(history) + const deviations = history.map((v) => Math.abs(v - med)) + let sigma = median(deviations) * 1.4826 + if (sigma === 0) { + const variance = history.reduce((sum, v) => sum + (v - mean) ** 2, 0) / history.length + sigma = Math.sqrt(variance) + } + if (sigma === 0) sigma = 1e-9 + return { samples: history.length, mean, median: med, sigma } +} + +export function zScore(history: number[], value: number): number { + const stats = robustStats(history) + if (stats.samples === 0) return 0 + const z = (value - stats.median) / stats.sigma + return Math.min(z, 999) +} + +/** z-score finding against a metric's history; null while the baseline is thin or the value fits. */ +export function detectValueFinding( + kind: AnomalyKind, + metric: string, + history: number[], + value: number, + opts: Partial = {}, +): AnomalyFinding | null { + const o = { ...ANOMALY_DEFAULTS, ...opts } + if (history.length < o.minSamples) return null + const z = zScore(history, value) + if (z < o.zThreshold) return null + const severity: AnomalySeverity = z >= o.criticalZ ? "critical" : "warning" + const stats = robustStats(history) + return { + kind, + severity, + metric, + value, + z: Math.round(z * 10) / 10, + message: + `${kind} anomaly on ${metric}: ${value} vs baseline median ${Math.round(stats.median * 100) / 100} (z=${Math.round(z * 10) / 10})`, + } +} + +/** Per-run deltas for cumulative counters; resets and first-seen keys are skipped. */ +export function counterDeltas( + prev: Record, + curr: Record, +): Record { + const deltas: Record = {} + for (const [name, value] of Object.entries(curr)) { + const before = prev[name] + if (before === undefined || !Number.isFinite(before) || !Number.isFinite(value)) continue + if (value < before) continue + deltas[name] = value - before + } + return deltas +} + +/** Percentage-drop findings vs the previous run's balances. */ +export function detectBalanceDrops( + prev: Record, + curr: Record, + opts: Partial = {}, +): AnomalyFinding[] { + const o = { ...ANOMALY_DEFAULTS, ...opts } + const findings: AnomalyFinding[] = [] + for (const [metric, value] of Object.entries(curr)) { + const before = prev[metric] + if (before === undefined || !Number.isFinite(before) || before <= 0 || !Number.isFinite(value)) continue + const dropPct = ((before - value) / before) * 100 + if (dropPct >= o.balanceDropWarnPct) { + const severity: AnomalySeverity = + dropPct >= o.balanceDropCriticalPct ? "critical" : "warning" + findings.push({ + kind: "balance", + severity, + metric, + value, + z: null, + message: `balance anomaly on ${metric}: dropped ${Math.round(dropPct)}% (${Math.round(before * 100) / 100} → ${Math.round(value * 100) / 100})`, + }) + } + } + return findings +} + +// ─── Detection over a sample ───────────────────────────────────────────────── + +export type DetectionOutcome = { + findings: AnomalyFinding[] + baseline: BaselineStore +} + +function pushHistory(history: number[], value: number, maxLen: number): number[] { + const next = [...history, value] + return next.length > maxLen ? next.slice(next.length - maxLen) : next +} + +/** + * Evaluates a sample against (and then extends) the baseline. Detection runs + * BEFORE the current value is pushed, so a sample never pollutes the baseline + * it is measured against. + */ +export function detectAnomalies( + rawBaseline: BaselineStore, + sample: AnomalySample, + opts: Partial = {}, +): DetectionOutcome { + const o = { ...ANOMALY_DEFAULTS, ...opts } + const baseline = normalizeBaselineStore(rawBaseline) + const metrics = baseline.metrics ?? {} + const findings: AnomalyFinding[] = [] + + for (const [metric, value] of Object.entries(sample.volume)) { + const key = `volume:${metric}` + const finding = detectValueFinding("volume", metric, metrics[key] ?? [], value, o) + if (finding) { + findings.push(finding) + incSecurityCounter("anomaly_volume_breaches") + } + metrics[key] = pushHistory(metrics[key] ?? [], value, o.maxHistory) + } + + const deltas = counterDeltas(baseline.counters_prev ?? {}, sample.counters) + for (const [name, delta] of Object.entries(deltas)) { + const key = `counter:${name}` + const finding = detectValueFinding("counter", name, metrics[key] ?? [], delta, o) + if (finding) { + findings.push(finding) + incSecurityCounter("anomaly_counter_breaches") + } + metrics[key] = pushHistory(metrics[key] ?? [], delta, o.maxHistory) + } + + const drops = detectBalanceDrops(baseline.balances_prev ?? {}, sample.balances, o) + for (const finding of drops) { + findings.push(finding) + incSecurityCounter("anomaly_balance_breaches") + } + + baseline.counters_prev = { ...sample.counters } + baseline.balances_prev = { ...sample.balances } + baseline.metrics = metrics + + return { findings, baseline } +} + +// ─── Data collection (cron) ────────────────────────────────────────────────── + +/** Anomaly's own counters are excluded so the detector cannot alert on itself. */ +const SELF_COUNTER_PREFIX = "anomaly_" + +export async function collectAnomalySample( + opts: { now?: number; volumeWindowMs?: number } = {}, +): Promise { + const now = opts.now ?? Date.now() + const since = now - (opts.volumeWindowMs ?? ANOMALY_DEFAULTS.volumeWindowMs) + + const volume: Record = {} + try { + const row = getDb() + .prepare( + `SELECT COALESCE(SUM(amount_phaselq), 0) AS total, COUNT(*) AS offers + FROM offers + WHERE status = 'accepted' AND wash_flagged = 0 AND created_at >= ?`, + ) + .get(since) as { total: number; offers: number } + volume.market_volume_phaselq_24h = Number(row?.total ?? 0) + volume.market_offers_accepted_24h = Number(row?.offers ?? 0) + } catch { + // No database yet — volume metrics are simply absent from this sample. + } + + const counters: Record = {} + for (const [name, value] of Object.entries(snapshotSecurityCounters())) { + if (name.startsWith(SELF_COUNTER_PREFIX)) continue + counters[name] = value + } + + const balances: Record = {} + try { + const latest = (await getHealthHistory(1))[0] + if (latest) { + if (typeof latest.distributorXlm === "number") balances.distributor_xlm = latest.distributorXlm + if (latest.distributorPhaseLiqStroops !== null) { + balances.distributor_phase_liq = Number(latest.distributorPhaseLiqStroops) / 10_000_000 + } + if (typeof latest.issuerXlm === "number") balances.issuer_xlm = latest.issuerXlm + } + } catch { + // Health store unreadable — balance checks skip this run. + } + + return { ts: now, volume, counters, balances } +} + +// ─── Alert routing ─────────────────────────────────────────────────────────── + +export type AlertRouter = ( + type: WebhookAlertType, + payload: WebhookAlertPayload, +) => Promise<{ sent: string[]; failed: string[] }> + +export function anomalyAlertFor(finding: AnomalyFinding): { + type: WebhookAlertType + payload: WebhookAlertPayload +} { + return { + type: finding.severity, + payload: { + // Distinct per kind+metric: webhook dedup keys on type+title. + title: `Anomaly ${finding.kind}: ${finding.metric}`, + message: finding.message, + kind: finding.kind, + severity: finding.severity, + metric: finding.metric, + value: finding.value, + ...(finding.z !== null ? { z: finding.z } : {}), + }, + } +} + +export async function routeAnomalyAlerts( + findings: AnomalyFinding[], + router: AlertRouter = sendWebhookAlert, +): Promise<{ routed: number; sent: number }> { + let routed = 0 + let sent = 0 + for (const finding of findings) { + const { type, payload } = anomalyAlertFor(finding) + const result = await router(type, payload) + routed++ + if (result.sent.length > 0) sent++ + } + if (routed > 0) incSecurityCounter("anomaly_alerts_routed", routed) + return { routed, sent } +} diff --git a/lib/feature-flags.ts b/lib/feature-flags.ts index 0f101f30..333211e6 100644 --- a/lib/feature-flags.ts +++ b/lib/feature-flags.ts @@ -67,6 +67,10 @@ * - phase-171: backup and disaster-recovery snapshots for sidecars, SQLite and the secrets map (spike 605 / issue #469) * - phase-172: SQLite to Postgres migration rehearsal with dual-write, parity and rollback (spike 604 / issue #468) * - phase-173: Vercel preview-environment isolation for data roots, secrets, contracts and cron (spike 603 / issue #467) + * - phase-178: leaderboard integrity with anti-farm scoring, time decay and sybil discount (spike 616 / issue #479) + * - phase-179: data retention and GDPR erasure with TTL sweeps and wallet deletion (spike 617 / issue #480) + * - phase-180: analytics warehouse export with Parquet and NDJSON, schema-versioned and redacted (spike 618 / issue #481) + * - phase-181: anomaly detection on volume, counters and balances with webhook alert routing (spike 619 / issue #482) */ export type PhaseFeatureFlag = @@ -74,6 +78,10 @@ export type PhaseFeatureFlag = | "phase-175" | "phase-176" | "phase-177" + | "phase-178" + | "phase-179" + | "phase-180" + | "phase-181" | "phase-66" | "phase-77" @@ -168,6 +176,10 @@ const FLAG_ENV_MAP: Record = { "phase-175": ["NEXT_PUBLIC_FEATURE_PHASE_175", "FEATURE_PHASE_175"], "phase-176": ["NEXT_PUBLIC_FEATURE_PHASE_176", "FEATURE_PHASE_176"], "phase-177": ["NEXT_PUBLIC_FEATURE_PHASE_177", "FEATURE_PHASE_177"], + "phase-178": ["NEXT_PUBLIC_FEATURE_PHASE_178", "FEATURE_PHASE_178"], + "phase-179": ["NEXT_PUBLIC_FEATURE_PHASE_179", "FEATURE_PHASE_179"], + "phase-180": ["NEXT_PUBLIC_FEATURE_PHASE_180", "FEATURE_PHASE_180"], + "phase-181": ["NEXT_PUBLIC_FEATURE_PHASE_181", "FEATURE_PHASE_181"], "phase-66": ["NEXT_PUBLIC_FEATURE_PHASE_66", "FEATURE_PHASE_66"], "phase-77": ["NEXT_PUBLIC_FEATURE_PHASE_77", "FEATURE_PHASE_77"], "phase-78": ["NEXT_PUBLIC_FEATURE_PHASE_78", "FEATURE_PHASE_78"], @@ -272,6 +284,10 @@ export function isFeatureEnabled(flag: PhaseFeatureFlag): boolean { "phase-175": process.env.NEXT_PUBLIC_FEATURE_PHASE_175, "phase-176": process.env.NEXT_PUBLIC_FEATURE_PHASE_176, "phase-177": process.env.NEXT_PUBLIC_FEATURE_PHASE_177, + "phase-178": process.env.NEXT_PUBLIC_FEATURE_PHASE_178, + "phase-179": process.env.NEXT_PUBLIC_FEATURE_PHASE_179, + "phase-180": process.env.NEXT_PUBLIC_FEATURE_PHASE_180, + "phase-181": process.env.NEXT_PUBLIC_FEATURE_PHASE_181, }; if (isTruthy(newFlags[flag])) return true; // Public keys are checked first so the same helper works in browser and server code. @@ -305,6 +321,7 @@ export function getEnabledFeatureFlags(): PhaseFeatureFlag[] { // Keep enumeration explicit so dashboards have stable, reviewable flag order. const all: PhaseFeatureFlag[] = [ "phase-174", "phase-175", "phase-176", "phase-177", + "phase-178", "phase-179", "phase-180", "phase-181", "phase-66", "phase-77", "phase-78", diff --git a/lib/leaderboard-integrity.ts b/lib/leaderboard-integrity.ts new file mode 100644 index 00000000..85bf628e --- /dev/null +++ b/lib/leaderboard-integrity.ts @@ -0,0 +1,238 @@ +/** + * Leaderboard integrity: anti-farm scoring with decay and sybil discount. + * Spike 616 / Issue #479. + * + * A points table is trivially gameable without three safeguards: + * 1. Anti-farm signals — burst posting, counterparty concentration and + * point velocity each multiply the score down (floor at + * `farmPenaltyFloor`) so a campaign earns less per unit of effort than + * a steady participant. + * 2. Time decay — every event's points are weighted by an exponential + * half-life (`0.5 ** (age / halfLifeMs)`, same shape as the pricing + * oracle's wash discount), so stale activity cannot pin a wallet to the + * top of the table forever. + * 3. Sybil discount — the caller supplies the band from + * `lib/sybil-resistance.ts:scoreWalletHistory`; suspect wallets keep a + * fixed fraction of their points. A missing band is treated as + * `caution` so an unscored wallet never gets trusted-rate points. + * + * All scoring is pure and deterministic (given `now`), so the leaderboard + * route and tests share one code path. Flag: phase-178. + * Rollback: unset NEXT_PUBLIC_FEATURE_PHASE_178 / FEATURE_PHASE_178 — routes + * 404 and no persisted state is required to revert. + */ + +import { isFeatureEnabled } from "@/lib/feature-flags" + +export const LEADERBOARD_SCHEMA_VERSION = "1.0.0" + +export function isLeaderboardIntegrityEnabled(): boolean { + return isFeatureEnabled("phase-178") +} + +export function leaderboard178RollbackNote(): string { + return "Rollback phase-178: unset NEXT_PUBLIC_FEATURE_PHASE_178 / FEATURE_PHASE_178 or set 0/false and restart. leaderboard-entries.json stays inert." +} + +export type SybilBand = "trusted" | "caution" | "suspect" + +export type LeaderboardEvent = { + points: number + at: number + counterparty?: string +} + +export type LeaderboardCandidate = { + wallet: string + events: LeaderboardEvent[] + sybilBand?: SybilBand +} + +export const LEADERBOARD_DEFAULTS = { + halfLifeMs: 7 * 24 * 60 * 60 * 1000, + burstWindowMs: 60 * 60 * 1000, + burstMaxEvents: 12, + maxCounterpartyShare: 0.8, + minConcentrationEvents: 6, + velocityMaxPointsPerDay: 1000, + farmPenaltyFloor: 0.15, +} as const + +export const SYBIL_BAND_DISCOUNT: Record = { + trusted: 1, + caution: 0.7, + suspect: 0.35, +} + +/** No band supplied means unknown — unknown wallets never score at trusted rate. */ +export const UNKNOWN_SYBIL_BAND: SybilBand = "caution" + +export type LeaderboardIntegrityOptions = Partial + +export class LeaderboardIntegrityError extends Error { + readonly code: "FLAG_DISABLED" | "VALIDATION_FAILED" + readonly details?: unknown + constructor(code: LeaderboardIntegrityError["code"], message: string, details?: unknown) { + super(message) + this.name = "LeaderboardIntegrityError" + this.code = code + this.details = details + } +} + +function round2(value: number): number { + return Math.round(value * 100) / 100 +} + +/** Exponential half-life decay: 1 at age 0, 0.5 at one half-life, → 0. */ +export function decayFactor(ageMs: number, halfLifeMs: number): number { + if (halfLifeMs <= 0) return 1 + const age = Math.max(0, ageMs) + return Math.pow(0.5, age / halfLifeMs) +} + +export type FarmDetection = { + signals: string[] + penalty: number + farmed: boolean +} + +/** + * Pure farm-pattern detector. Each matched rule multiplies the penalty + * (`burst` ×0.5, `concentration` ×0.6, `velocity` ×0.7), floored so a wallet + * is never zeroed out — the goal is to compress campaign ROI, not to make a + * single rule irreversible. + */ +export function detectFarmPatterns( + candidate: Pick, + opts: LeaderboardIntegrityOptions = {}, +): FarmDetection { + const o = { ...LEADERBOARD_DEFAULTS, ...opts } + const events = [...candidate.events].sort((a, b) => a.at - b.at) + const signals: string[] = [] + let penalty = 1 + + if (events.length > 0) { + let maxInWindow = 0 + for (let i = 0; i < events.length; i++) { + let count = 0 + for (let j = i; j < events.length; j++) { + if (events[j].at - events[i].at > o.burstWindowMs) break + count++ + } + if (count > maxInWindow) maxInWindow = count + } + if (maxInWindow > o.burstMaxEvents) { + signals.push("event_burst") + penalty *= 0.5 + } + + const withCounterparty = events.filter((e) => Boolean(e.counterparty)) + if (withCounterparty.length >= o.minConcentrationEvents) { + const counts = new Map() + for (const e of withCounterparty) { + const key = e.counterparty as string + counts.set(key, (counts.get(key) ?? 0) + 1) + } + const largest = Math.max(...counts.values()) + if (largest / withCounterparty.length > o.maxCounterpartyShare) { + signals.push("counterparty_concentration") + penalty *= 0.6 + } + } + + const rawPoints = events.reduce((sum, e) => sum + e.points, 0) + const spanMs = events[events.length - 1].at - events[0].at + const spanDays = Math.max(spanMs, 24 * 60 * 60 * 1000) / (24 * 60 * 60 * 1000) + if (rawPoints / spanDays > o.velocityMaxPointsPerDay) { + signals.push("point_velocity") + penalty *= 0.7 + } + } + + return { + signals, + penalty: round2(Math.max(o.farmPenaltyFloor, penalty)), + farmed: signals.length > 0, + } +} + +export function sybilDiscountFor(band?: SybilBand): number { + return SYBIL_BAND_DISCOUNT[band ?? UNKNOWN_SYBIL_BAND] +} + +export type LeaderboardScore = { + wallet: string + events: number + rawPoints: number + decayedPoints: number + farmPenalty: number + sybilDiscount: number + finalScore: number + signals: string[] +} + +export function scoreLeaderboardCandidate( + candidate: LeaderboardCandidate, + opts: LeaderboardIntegrityOptions = {}, + now: number = Date.now(), +): LeaderboardScore { + if (typeof candidate.wallet !== "string" || candidate.wallet.trim().length === 0) { + throw new LeaderboardIntegrityError("VALIDATION_FAILED", "candidate wallet is required") + } + if (!Array.isArray(candidate.events)) { + throw new LeaderboardIntegrityError("VALIDATION_FAILED", "candidate events must be an array") + } + const halfLifeMs = opts.halfLifeMs ?? LEADERBOARD_DEFAULTS.halfLifeMs + + let rawPoints = 0 + let decayedPoints = 0 + for (const event of candidate.events) { + if (!Number.isFinite(event?.points) || !Number.isFinite(event?.at)) { + throw new LeaderboardIntegrityError( + "VALIDATION_FAILED", + "every event needs finite points and an epoch-ms `at` timestamp", + ) + } + rawPoints += event.points + decayedPoints += event.points * decayFactor(now - event.at, halfLifeMs) + } + + const farm = detectFarmPatterns(candidate, opts) + const sybilDiscount = sybilDiscountFor(candidate.sybilBand) + + const signals = [...farm.signals] + const band = candidate.sybilBand ?? UNKNOWN_SYBIL_BAND + if (sybilDiscount < 1) signals.push(`sybil_${band}`) + + return { + wallet: candidate.wallet, + events: candidate.events.length, + rawPoints: round2(rawPoints), + decayedPoints: round2(decayedPoints), + farmPenalty: farm.penalty, + sybilDiscount, + finalScore: round2(decayedPoints * farm.penalty * sybilDiscount), + signals, + } +} + +export function rankLeaderboard( + candidates: LeaderboardCandidate[], + opts: LeaderboardIntegrityOptions = {}, + now: number = Date.now(), +): LeaderboardScore[] { + return candidates + .map((candidate) => scoreLeaderboardCandidate(candidate, opts, now)) + .sort((a, b) => b.finalScore - a.finalScore || a.wallet.localeCompare(b.wallet)) +} + +export function topLeaderboard( + candidates: LeaderboardCandidate[], + limit: number, + opts: LeaderboardIntegrityOptions = {}, + now: number = Date.now(), +): LeaderboardScore[] { + const capped = Math.max(0, Math.min(Math.floor(limit), 200)) + return rankLeaderboard(candidates, opts, now).slice(0, capped) +} diff --git a/lib/retention-erasure.ts b/lib/retention-erasure.ts new file mode 100644 index 00000000..031c3112 --- /dev/null +++ b/lib/retention-erasure.ts @@ -0,0 +1,292 @@ +/** + * Data retention and GDPR erasure pipeline (Article 17) with TTL sweeps. + * Spike 617 / Issue #480. + * + * Two halves share one flag (phase-179): + * + * 1. TTL sweeps — pure prune functions plus `runRetentionSweeps()`, which + * applies them to the stores that accumulate time-shaped data: expired + * SEP-10 sessions, stale consumed challenges and old notifications. + * Sweeps fail safe: a record without a parseable timestamp is kept. + * + * 2. Wallet erasure — `executeWalletErasure()` walks the wallet-keyed + * sidecars (notifications, age declarations, follows, push + * subscriptions) plus the SEP-10 session map, removes every entry the + * wallet owns, and appends a schema-versioned receipt keyed by a wallet + * fingerprint (never the raw address). + * + * Erasure is gated by `assertNoLegalHold` (phase-169): an active hold throws + * `LEGAL_HOLD_ACTIVE` (HTTP 423) and nothing is deleted. The terms-consent + * ledger is deliberately NOT erased — consent records are the proof of legal + * basis kept under Art. 17(3)(b), so withdrawal is recorded, not shredded. + * + * Partial slice: SQLite tables (signals, replies, listings, offers, tax + * events) are not yet walked; erasure covers JSON sidecars only. + * Rollback: unset the flag — routes 404 and stored receipts stay inert. + */ + +import { isFeatureEnabled } from "@/lib/feature-flags" +import { assertNoLegalHold, LegalHoldError } from "@/lib/legal-hold" +import { updateStore } from "@/lib/json-store" +import { fingerprintWallet } from "@/lib/redaction" +import { incSecurityCounter } from "@/lib/security-counters" +import type { ServerDataFile } from "@/lib/server-data-paths" + +export const ERASURE_RECEIPT_SCHEMA = "phase-gdpr-article-17-erasure" +export const ERASURE_RECEIPT_VERSION = "1.0.0" +export const MAX_ERASURE_RECEIPTS = 1000 + +const G_ADDRESS_RE = /^G[A-Z2-7]{55}$/ + +export function isRetentionErasureEnabled(): boolean { + return isFeatureEnabled("phase-179") +} + +export function retention179RollbackNote(): string { + return "Rollback phase-179: unset NEXT_PUBLIC_FEATURE_PHASE_179 / FEATURE_PHASE_179 or set 0/false and restart. Sweeps stop running and the erasure route 404s; receipts stay inert." +} + +export class RetentionErasureError extends Error { + readonly code: "FLAG_DISABLED" | "VALIDATION_FAILED" | "LEGAL_HOLD_ACTIVE" + readonly details?: unknown + constructor(code: RetentionErasureError["code"], message: string, details?: unknown) { + super(message) + this.name = "RetentionErasureError" + this.code = code + this.details = details + } +} + +// ─── TTL sweeps (pure) ─────────────────────────────────────────────────────── + +export const RETENTION_TTL_MS = { + authSession: 30 * 24 * 60 * 60 * 1000, + consumedChallenge: 24 * 60 * 60 * 1000, + notification: 90 * 24 * 60 * 60 * 1000, +} as const + +export type TtlSweepResult = { kept: T[]; removed: T[] } + +export function sweepByTimestamp( + records: T[], + opts: { + now: number + ttlMs: number + timestampOf: (record: T) => number | null | undefined + }, +): TtlSweepResult { + const kept: T[] = [] + const removed: T[] = [] + for (const record of records) { + const ts = opts.timestampOf(record) + if (ts !== null && ts !== undefined && Number.isFinite(ts) && opts.now - ts > opts.ttlMs) { + removed.push(record) + } else { + kept.push(record) + } + } + return { kept, removed } +} + +export type AuthSessionRecord = { wallet?: string; expires_at?: number } +export type AuthSessionStoreShape = { + sessions?: Record + consumed_challenges?: Record +} + +/** SEP-10 stores `expires_at`/challenge timestamps in seconds; normalize to ms. */ +function toEpochMs(value: number): number { + return value < 1e12 ? value * 1000 : value +} + +/** Drops expired sessions and consumed challenges older than their TTL. Mutates `store`. */ +export function sweepAuthSessions(store: AuthSessionStoreShape, now: number): number { + let removed = 0 + const sessions = store.sessions ?? {} + for (const [token, record] of Object.entries(sessions)) { + const expiresAt = typeof record?.expires_at === "number" ? toEpochMs(record.expires_at) : null + if (expiresAt !== null && now > expiresAt) { + delete sessions[token] + removed++ + } + } + store.sessions = sessions + + const challenges = store.consumed_challenges ?? {} + for (const [hash, consumedAt] of Object.entries(challenges)) { + if (Number.isFinite(consumedAt) && now - toEpochMs(consumedAt) > RETENTION_TTL_MS.consumedChallenge) { + delete challenges[hash] + removed++ + } + } + store.consumed_challenges = challenges + return removed +} + +export type Notifiable = { created_at?: number } + +/** Prunes expired notifications and empties wallet buckets that ran dry. Mutates `store`. */ +export function sweepNotificationStore( + store: Record, + now: number, + ttlMs: number = RETENTION_TTL_MS.notification, +): number { + let removed = 0 + for (const wallet of Object.keys(store)) { + const list = Array.isArray(store[wallet]) ? store[wallet] : [] + const { kept, removed: dropped } = sweepByTimestamp(list, { + now, + ttlMs, + timestampOf: (record) => (typeof record?.created_at === "number" ? record.created_at : null), + }) + removed += dropped.length + if (kept.length === 0) delete store[wallet] + else store[wallet] = kept + } + return removed +} + +// ─── Wallet erasure (Article 17) ───────────────────────────────────────────── + +export type ErasureDomainResult = { domain: string; removed: number } + +export type ErasureReceipt = { + schema: typeof ERASURE_RECEIPT_SCHEMA + version: string + wallet_fingerprint: string + domains: ErasureDomainResult[] + total_removed: number + executed_at: number + initiated_by: "wallet" | "admin" +} + +type ErasureTarget = { + domain: string + store: ServerDataFile + shape: "wallet-keyed" | "auth-sessions" +} + +export const ERASURE_TARGETS: ErasureTarget[] = [ + { domain: "notifications", store: "notifications", shape: "wallet-keyed" }, + { domain: "age_declarations", store: "ageDeclarations", shape: "wallet-keyed" }, + { domain: "follows", store: "profileFollows", shape: "wallet-keyed" }, + { domain: "push_subscriptions", store: "pushSubscriptions", shape: "wallet-keyed" }, + { domain: "auth_sessions", store: "authSessions", shape: "auth-sessions" }, +] + +/** + * Deletes the wallet's bucket from a `Record` sidecar and + * returns how many records went away (array buckets count their elements). + * Mutates `store`, matching the `updateStore` callback contract. + */ +export function eraseWalletKeyedRecord(store: Record, wallet: string): number { + if (!Object.prototype.hasOwnProperty.call(store, wallet)) return 0 + const value = store[wallet] + const removed = Array.isArray(value) ? value.length : 1 + delete store[wallet] + return removed +} + +/** Removes every SEP-10 session bound to the wallet. Mutates `store`. */ +export function eraseWalletAuthSessions(store: AuthSessionStoreShape, wallet: string): number { + const sessions = store.sessions ?? {} + let removed = 0 + for (const [token, record] of Object.entries(sessions)) { + if (record?.wallet === wallet) { + delete sessions[token] + removed++ + } + } + store.sessions = sessions + return removed +} + +export type ErasureOptions = { + initiatedBy?: "wallet" | "admin" + now?: number +} + +export async function executeWalletErasure( + wallet: string, + opts: ErasureOptions = {}, +): Promise { + if (!isRetentionErasureEnabled()) { + throw new RetentionErasureError("FLAG_DISABLED", "GDPR erasure is disabled (phase-179 flag off)") + } + const target = wallet.trim() + if (!G_ADDRESS_RE.test(target)) { + throw new RetentionErasureError("VALIDATION_FAILED", "a Stellar wallet address (G...) is required") + } + + try { + await assertNoLegalHold("wallet", target, "erasure", target) + } catch (error) { + if (error instanceof LegalHoldError) { + incSecurityCounter("gdpr_erasure_blocked") + throw new RetentionErasureError("LEGAL_HOLD_ACTIVE", error.message) + } + throw error + } + + const domains: ErasureDomainResult[] = [] + for (const erasure of ERASURE_TARGETS) { + const removed = await updateStore(erasure.store, (data) => { + if (erasure.shape === "wallet-keyed") { + return eraseWalletKeyedRecord((data ?? {}) as Record, target) + } + return eraseWalletAuthSessions((data ?? {}) as AuthSessionStoreShape, target) + }) + if (removed > 0) { + domains.push({ domain: erasure.domain, removed }) + incSecurityCounter("gdpr_erasure_records", removed) + } + } + + const receipt: ErasureReceipt = { + schema: ERASURE_RECEIPT_SCHEMA, + version: ERASURE_RECEIPT_VERSION, + wallet_fingerprint: fingerprintWallet(target), + domains, + total_removed: domains.reduce((sum, d) => sum + d.removed, 0), + executed_at: opts.now ?? Date.now(), + initiated_by: opts.initiatedBy ?? "wallet", + } + + await updateStore<{ receipts: ErasureReceipt[] }>("gdprErasureLog", (store) => { + store.receipts = [receipt, ...(store.receipts ?? [])].slice(0, MAX_ERASURE_RECEIPTS) + }) + + incSecurityCounter("gdpr_erasure_completed") + return receipt +} + +// ─── Orchestration ─────────────────────────────────────────────────────────── + +export type SweepOutcome = { store: string; removed: number } + +export async function runRetentionSweeps( + now: number = Date.now(), +): Promise<{ ok: true; sweeps: SweepOutcome[]; total_removed: number }> { + if (!isRetentionErasureEnabled()) { + throw new RetentionErasureError("FLAG_DISABLED", "Retention sweeps are disabled (phase-179 flag off)") + } + + const sweeps: SweepOutcome[] = [] + + const authRemoved = await updateStore("authSessions", (store) => + sweepAuthSessions(store, now), + ) + sweeps.push({ store: "authSessions", removed: authRemoved }) + + const notificationRemoved = await updateStore, number>( + "notifications", + (store) => sweepNotificationStore(store, now), + ) + sweeps.push({ store: "notifications", removed: notificationRemoved }) + + const total_removed = sweeps.reduce((sum, s) => sum + s.removed, 0) + incSecurityCounter("retention_sweep_runs") + if (total_removed > 0) incSecurityCounter("retention_records_expired", total_removed) + + return { ok: true, sweeps, total_removed } +} diff --git a/lib/security-counters.ts b/lib/security-counters.ts index 071d160d..78d3a8b9 100644 --- a/lib/security-counters.ts +++ b/lib/security-counters.ts @@ -135,7 +135,24 @@ export type SecurityCounterName = | "db_dual_write_drift" | "db_dual_write_shed" | "preview_isolation_violation" - | "preview_cron_blocked"; + | "preview_cron_blocked" + | "leaderboard_entries_ingested" + | "leaderboard_scored" + | "leaderboard_farm_penalized" + | "leaderboard_sybil_discounted" + | "retention_sweep_runs" + | "retention_records_expired" + | "gdpr_erasure_completed" + | "gdpr_erasure_blocked" + | "gdpr_erasure_records" + | "warehouse_export_runs" + | "warehouse_records_redacted" + | "warehouse_export_failed" + | "anomaly_checks_run" + | "anomaly_volume_breaches" + | "anomaly_counter_breaches" + | "anomaly_balance_breaches" + | "anomaly_alerts_routed"; const counters = new Map() diff --git a/lib/server-data-paths.ts b/lib/server-data-paths.ts index f65249cd..079e1ffd 100644 --- a/lib/server-data-paths.ts +++ b/lib/server-data-paths.ts @@ -50,6 +50,9 @@ const FILES = { legalHolds: "legal-holds.json", termsConsents: "terms-consents.json", legalDocuments: "legal-documents.json", + leaderboardEntries: "leaderboard-entries.json", + gdprErasureLog: "gdpr-erasure-log.json", + anomalyBaselines: "anomaly-baselines.json", } as const diff --git a/package-lock.json b/package-lock.json index 26b012b9..44a911ff 100644 --- a/package-lock.json +++ b/package-lock.json @@ -50,6 +50,8 @@ "embla-carousel-react": "8.5.1", "framer-motion": "12.23.26", "gsap": "3.14.1", + "hyparquet": "^1.31.5", + "hyparquet-writer": "^0.17.0", "input-otp": "1.4.1", "lenis": "1.3.15", "lucide-react": "^0.454.0", @@ -13491,6 +13493,21 @@ "ms": "^2.0.0" } }, + "node_modules/hyparquet": { + "version": "1.31.5", + "resolved": "https://registry.npmjs.org/hyparquet/-/hyparquet-1.31.5.tgz", + "integrity": "sha512-KPVfaxW8xOiO2/4fjeiBfFNEXhGVw9tfcb9ha2iNZdJ3J/yBjEwIeAnTyyuhllFFTb9YO9wg0DfRAOnm6KCvBg==", + "license": "MIT" + }, + "node_modules/hyparquet-writer": { + "version": "0.17.0", + "resolved": "https://registry.npmjs.org/hyparquet-writer/-/hyparquet-writer-0.17.0.tgz", + "integrity": "sha512-GZ4QeopztjSli/Sz1dSOP/TrSKtwG2ydP2yuPMxeiZnho+PCXF7x+wx8OIVtpQn8rcv559IH1EJMHnTbv0sn+Q==", + "license": "MIT", + "dependencies": { + "hyparquet": "1.31.5" + } + }, "node_modules/idb-keyval": { "version": "6.2.1", "resolved": "https://registry.npmjs.org/idb-keyval/-/idb-keyval-6.2.1.tgz", diff --git a/package.json b/package.json index a8506a67..576db30b 100644 --- a/package.json +++ b/package.json @@ -67,6 +67,8 @@ "embla-carousel-react": "8.5.1", "framer-motion": "12.23.26", "gsap": "3.14.1", + "hyparquet": "^1.31.5", + "hyparquet-writer": "^0.17.0", "input-otp": "1.4.1", "lenis": "1.3.15", "lucide-react": "^0.454.0", diff --git a/vercel.json b/vercel.json index a3ece80d..be616bbd 100644 --- a/vercel.json +++ b/vercel.json @@ -3,6 +3,14 @@ { "path": "/api/cron/faucet-health", "schedule": "0 * * * *" + }, + { + "path": "/api/cron/anomaly-detect", + "schedule": "5 * * * *" + }, + { + "path": "/api/cron/retention-sweep", + "schedule": "15 3 * * *" } ], "headers": [