Add SQL injection prevention audit (#777)

.github/workflows/playwright-e2e.yml

@@ -18,7 +18,7 @@ jobs:
       - name: Set up Node.js
         uses: actions/setup-node@v4
         with:
-          node-version: '18'
+          node-version: '20'
           cache: 'npm'
           cache-dependency-path: frontend/package-lock.json
 

SQL_INJECTION_AUDIT.md

@@ -0,0 +1,67 @@
+# SQL Injection Audit (#777)
+
+## Scope
+
+All database access in `backend/src/services/db.ts`, `backend/src/index.ts`,
+and every service under `backend/src/services/` that calls
+`better-sqlite3`'s `.prepare()` / `.exec()`.
+
+## Method
+
+1. Enumerated every `.prepare(...)` and `db.exec(...)` call site
+   (58 call sites across `streamStore.ts`, `eventHistory.ts`, `indexer.ts`,
+   `webhook.ts`, `webhookWorker.ts`, `metricsHistory.ts`, `migrations.ts`,
+   `stats.ts`, `streamMetrics.ts`, `reconciliationJob.ts`, and `index.ts`).
+2. For each call site, checked whether any part of the SQL string is built
+   from request/user-controlled data (query params, path params, body
+   fields) via string concatenation or template-literal interpolation.
+3. For call sites that build SQL dynamically (conditional `WHERE` clauses,
+   `ORDER BY` direction/column), verified the *values* are always bound as
+   parameters and only a closed set of hardcoded/allowlisted tokens are
+   ever interpolated into the SQL text itself.
+4. Added `backend/src/services/sqlInjection.integration.test.ts`, which
+   feeds classic SQLi payloads (`' OR '1'='1`, `'; DROP TABLE streams; --`,
+   `' UNION SELECT * FROM streams --`, etc.) through every user-controlled
+   string input reachable from a store/service function (recipient/sender
+   address, stream ID, actor, event-type filter) and asserts the payload is
+   treated as inert plain text and the schema/data stays intact.
+
+## Findings
+
+**No raw string interpolation of user-controlled data into SQL was found.**
+Every query that accepts external input uses `?` positional or `@name`
+named bind parameters. The only template-literal interpolations found are:
+
+| File | Line(s) | Interpolated value | Source | Risk |
+|------|---------|---------------------|--------|------|
+| `streamStore.ts` | `buildOrderClause` (~1017-1021), used at 1027-1046 | SQL column name + `ASC`/`DESC` | `SORT_COLUMNS` allowlist keyed by a typed `SortField` union, and a ternary that can only produce `"ASC"` or `"DESC"` | None — not attacker-controlled, closed set |
+| `eventHistory.ts` | `getStreamHistory` (line 87) | `ASC`/`DESC` | ternary on `order === 'asc'` | None — closed set |
+| `indexer.ts` | 119, 138, 143, 147 | `INDEXER_CURSOR_TABLE` | module-level `const`, never derived from input | None |
+| `stats.ts` | 43, 102 | none — static SQL text with `:now` named parameter | n/a | None |
+
+All other dynamic query builders (`eventHistory.getAllEvents`,
+`getGlobalEvents`, `countAllEvents`) build the `WHERE` clause by pushing
+literal condition fragments like `"event_type = ?"` onto an array and
+joining with `" AND "` — the *values* (`eventType`, `streamId`, `cursor`,
+`since`) are always pushed to a separate `params` array and passed to
+`.all(...params)` / `.get(...params)`, never into the SQL string.
+
+## Conclusion
+
+The audit found the codebase already follows parameterized-query best
+practice everywhere. No code changes were required to close an injection
+gap. This PR adds:
+
+- `backend/src/services/sqlInjection.integration.test.ts` — regression
+  tests asserting SQLi payloads are stored/matched as plain text via
+  `listStreamsByRecipient`, `listStreamsBySender`, `getStreamHistory`,
+  `getGlobalEvents`, and `countAllEvents`, and that the `streams` table
+  is never dropped or bypassed.
+- This document, as the audit record requested by the issue.
+
+## Recommendation
+
+Keep using `@name` / `?` bindings for all future queries (per
+`CLAUDE.md`'s "Code Patterns" section) and keep any dynamic `ORDER BY`
+input behind an explicit allowlist like `SORT_COLUMNS`, never accept a raw
+column/direction string from a request.

backend/eslint.config.mjs

@@ -21,6 +21,7 @@ export default tseslint.config(
       'src/services/eventHistory.ts',
       'src/services/indexer.ts',
       'src/services/metricsHistory.ts',
+      'src/services/migrations.ts',
       'src/services/openIssues.ts',
       'src/services/reconciliationJob.ts',
       'src/services/streamStore.ts',
@@ -36,11 +37,19 @@ export default tseslint.config(
       '@typescript-eslint/no-require-imports': 'off',
     },
   },
+  {
+    files: ['src/services/db.ts'],
+    rules: {
+      '@typescript-eslint/no-this-alias': 'off',
+      '@typescript-eslint/no-unsafe-function-type': 'off',
+    },
+  },
   {
     files: ['src/**/*.test.ts', 'src/**/*.integration.test.ts'],
     rules: {
       '@typescript-eslint/no-explicit-any': 'off',
       '@typescript-eslint/no-unused-vars': 'off',
+      '@typescript-eslint/no-require-imports': 'off',
       'no-console': 'off',
       'no-empty': 'off',
       'no-useless-assignment': 'off',

backend/migrations/005_add_fts_and_allowed_assets.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS streams_fts;
+DROP TABLE IF EXISTS allowed_assets;

backend/migrations/005_add_fts_and_allowed_assets.sql

@@ -0,0 +1,12 @@
+CREATE VIRTUAL TABLE IF NOT EXISTS streams_fts USING fts5(
+  stream_id UNINDEXED,
+  sender,
+  recipient,
+  asset_code,
+  content=streams,
+  content_rowid=rowid
+);
+
+CREATE TABLE IF NOT EXISTS allowed_assets (
+  code TEXT PRIMARY KEY
+);

backend/src/config/validateEnv.ts

@@ -55,6 +55,14 @@ const archiveCronIntervalSchema = z
     message: "must be a valid number >= 60000 (minimum 1 minute)",
   });
 
+// Webhook dead-letter pruning job interval validation
+const webhookDeadLetterPruneIntervalSchema = z
+  .string()
+  .transform((val: string) => parseInt(val, 10))
+  .refine((val: number) => !isNaN(val) && val >= 60000, {
+    message: "must be a valid number >= 60000 (minimum 1 minute)",
+  });
+
 // Indexer fallback polling interval validation
 const fallbackPollIntervalSchema = z
   .string()
@@ -92,6 +100,9 @@ const envSchema = z.object({
   INDEXER_POLL_INTERVAL_MS: indexerPollIntervalSchema.optional().default(10000),
   RECONCILIATION_INTERVAL_MS: reconciliationIntervalSchema.optional().default(60000),
   ARCHIVE_CRON_INTERVAL_MS: archiveCronIntervalSchema.optional().default(86400000),
+  WEBHOOK_DEAD_LETTER_PRUNE_INTERVAL_MS: webhookDeadLetterPruneIntervalSchema
+    .optional()
+    .default(86400000),
   INDEXER_FALLBACK_POLLING_ENABLED: z.string().optional().default("false"),
   INDEXER_FALLBACK_POLL_INTERVAL_MS: fallbackPollIntervalSchema.optional().default(10000),
   ALLOWED_ORIGINS: z.string().optional(),
@@ -114,6 +125,7 @@ export interface ValidatedConfig {
   indexerPollIntervalMs: number;
   reconciliationIntervalMs: number;
   archiveCronIntervalMs: number;
+  webhookDeadLetterPruneIntervalMs: number;
   indexerFallbackPollingEnabled: boolean;
   indexerFallbackPollIntervalMs: number;
   adminApiKey: string | null;
@@ -280,6 +292,7 @@ export function validateEnv(): ValidatedConfig {
       indexerPollIntervalMs: env.INDEXER_POLL_INTERVAL_MS,
       reconciliationIntervalMs: env.RECONCILIATION_INTERVAL_MS,
       archiveCronIntervalMs: env.ARCHIVE_CRON_INTERVAL_MS,
+      webhookDeadLetterPruneIntervalMs: env.WEBHOOK_DEAD_LETTER_PRUNE_INTERVAL_MS,
       indexerFallbackPollingEnabled: env.INDEXER_FALLBACK_POLLING_ENABLED,
       indexerFallbackPollIntervalMs: env.INDEXER_FALLBACK_POLL_INTERVAL_MS,
     },
@@ -303,6 +316,7 @@ export function validateEnv(): ValidatedConfig {
     indexerPollIntervalMs: env.INDEXER_POLL_INTERVAL_MS,
     reconciliationIntervalMs: env.RECONCILIATION_INTERVAL_MS,
     archiveCronIntervalMs: env.ARCHIVE_CRON_INTERVAL_MS,
+    webhookDeadLetterPruneIntervalMs: env.WEBHOOK_DEAD_LETTER_PRUNE_INTERVAL_MS,
     indexerFallbackPollingEnabled: process.env.INDEXER_FALLBACK_POLLING_ENABLED === "true",
     indexerFallbackPollIntervalMs: env.INDEXER_FALLBACK_POLL_INTERVAL_MS,
     adminApiKey,

backend/src/middleware/requestLogger.test.ts

@@ -7,7 +7,7 @@ import type { Request, Response } from "express";
 describe("requestLogger", () => {
   const originalNodeEnv = process.env.NODE_ENV;
   const loggerInfoSpy = vi.spyOn(logger, "info").mockImplementation(() => logger);
-  vi.spyOn(logger, "child").mockImplementation(() => logger);
+  vi.spyOn(logger, "child").mockImplementation(() => logger as any);
 
   beforeEach(() => {
     loggerInfoSpy.mockClear();

backend/src/services/db.ts

@@ -327,4 +327,81 @@ export function initDb(): void {
   }
 
   runMigrations(db);
+  seedAllowedAssetsIfEmpty(db);
+}
+
+function seedAllowedAssetsIfEmpty(dbInstance: any): void {
+  const row = dbInstance
+    .prepare("SELECT COUNT(*) as count FROM allowed_assets")
+    .get() as { count: number };
+  if (row.count > 0) return;
+
+  const defaults = (process.env.ALLOWED_ASSETS || "USDC,XLM")
+    .split(",")
+    .map((code) => code.trim().toUpperCase())
+    .filter(Boolean);
+
+  const insert = dbInstance.prepare(
+    "INSERT OR IGNORE INTO allowed_assets (code) VALUES (?)",
+  );
+  dbInstance.transaction(() => {
+    for (const code of defaults) {
+      insert.run(code);
+    }
+  })();
+}
+
+/** Keeps the FTS5 search index in sync with a stream's sender/recipient/asset. Best-effort. */
+export function syncFtsIndex(
+  streamId: string,
+  sender: string,
+  recipient: string,
+  assetCode: string,
+): void {
+  try {
+    getDb()
+      .prepare(
+        `INSERT INTO streams_fts(rowid, stream_id, sender, recipient, asset_code)
+         VALUES ((SELECT rowid FROM streams WHERE id = ?), ?, ?, ?, ?)
+         ON CONFLICT(rowid) DO UPDATE SET
+           sender = excluded.sender,
+           recipient = excluded.recipient,
+           asset_code = excluded.asset_code`,
+      )
+      .run(streamId, streamId, sender, recipient, assetCode);
+  } catch {
+    // FTS index update is best-effort; search degrades gracefully if it fails.
+  }
+}
+
+/** Searches streams via the FTS5 index, returning matching stream IDs ranked by relevance. */
+export function searchStreamsFts(query: string): string[] {
+  try {
+    const rows = getDb()
+      .prepare(
+        `SELECT stream_id FROM streams_fts WHERE streams_fts MATCH ? ORDER BY rank`,
+      )
+      .all(query) as Array<{ stream_id: string }>;
+    return rows.map((row) => row.stream_id);
+  } catch {
+    return [];
+  }
+}
+
+/** Returns allowed asset codes in insertion order (seed order, then any admin additions appended). */
+export function getAllowedAssets(): string[] {
+  const rows = getDb()
+    .prepare("SELECT code FROM allowed_assets ORDER BY rowid")
+    .all() as Array<{ code: string }>;
+  return rows.map((row) => row.code);
+}
+
+export function addAllowedAsset(code: string): void {
+  getDb()
+    .prepare("INSERT OR IGNORE INTO allowed_assets (code) VALUES (?)")
+    .run(code);
+}
+
+export function removeAllowedAsset(code: string): void {
+  getDb().prepare("DELETE FROM allowed_assets WHERE code = ?").run(code);
 }

backend/src/services/eventHistory.ts

@@ -1,6 +1,6 @@
 import { getDb } from "./db";
 
-export type StreamEventType = "created" | "claimed" | "canceled" | "start_time_updated" | "paused" | "resumed" | "completed" | "transferred";
+export type StreamEventType = "created" | "claimed" | "canceled" | "start_time_updated" | "paused" | "resumed" | "completed" | "transferred" | "clawback";
 
 export interface StreamEvent {
   id: number;