From b63ecdadb84ea5f630474961f901ff004af71305 Mon Sep 17 00:00:00 2001 From: kurehkim334-lang Date: Tue, 20 Oct 2026 07:45:42 +0000 Subject: [PATCH] feat: SSRF guard DNS resolution, webhook encryption, outbox worker, admin key config (#46 #47 #48 #49) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #49 — ADMIN_API_KEY through config layer - Add ADMIN_API_KEY (optional) to envSchema in config.ts - adminAuth reads config.ADMIN_API_KEY instead of process.env - Log boot warning when key is unset; admin routes return 401 - Update admin.test.ts to mock config instead of process.env #47 — Encrypt webhook secrets at rest - Add WEBHOOK_ENCRYPTION_KEY (optional) to envSchema; boot warning when unset - Add encryptWebhookSecret / decryptWebhookSecret helpers (reuse AES-256-GCM) - publishers.ts encrypts secret before storing; decrypts before delivering/testing - webhookService.ts resolveTarget() decrypts secret on load - Migration 0016: idempotent marker (re-encryption happens at app layer) #46 — SSRF guard resolves DNS at delivery time - Add isPrivateIp() covering IPv4 private/loopback/link-local + IPv6 equivalents - Add resolveHostname() wrapping dns.promises.lookup - deliver() resolves hostname before each attempt; blocks on private IPs - Guards against DNS rebinding: re-resolves on every retry attempt - Update webhookService.test.ts: mock node:dns, add SSRF + rebinding tests #48 — Persistent webhook outbox - Add webhook_outbox table to schema + migration 0015 - Add enqueueWebhookEvent() — writes to outbox before delivery - Replace void emitToPublisher() with void enqueueWebhookEvent() in all callers - New webhookOutboxWorker.ts: polls pending rows, exponential backoff, marks delivered/failed, survives process restarts - Wire worker start/stop into index.ts with the other background workers - Add GET /publishers/me/webhooks/deliveries endpoint for delivery history .gitignore: add *.snap, *.tmp, *.bak, npm-debug.log* --- .env.example | 10 ++ .gitignore | 10 +- drizzle/0015_webhook_outbox.sql | 22 ++++ drizzle/0016_encrypt_webhook_secrets.sql | 9 ++ src/config.ts | 36 ++++++ src/db/schema.ts | 39 +++++++ src/index.ts | 3 + src/routes/admin.test.ts | 118 ++++++++++--------- src/routes/admin.ts | 3 +- src/routes/publishers.ts | 51 +++++++- src/routes/resources.ts | 6 +- src/routes/verify.test.ts | 2 +- src/routes/verify.ts | 4 +- src/services/resourceService.ts | 4 +- src/services/webhookService.test.ts | 89 +++++++++++++- src/services/webhookService.ts | 129 +++++++++++++++++++- src/utils/crypto.ts | 18 +++ src/workers/webhookOutboxWorker.ts | 143 +++++++++++++++++++++++ 18 files changed, 627 insertions(+), 69 deletions(-) create mode 100644 drizzle/0015_webhook_outbox.sql create mode 100644 drizzle/0016_encrypt_webhook_secrets.sql create mode 100644 src/workers/webhookOutboxWorker.ts diff --git a/.env.example b/.env.example index 56e4d80..d007f0e 100644 --- a/.env.example +++ b/.env.example @@ -110,11 +110,21 @@ PUBLISHER_RATE_LIMIT_RPM_DEFAULT=60 PUBLISHER_RATE_LIMIT_MAX_RPM=10000 PUBLISHER_RATE_LIMIT_WINDOW_MS=60000 +# ── Admin panel ───────────────────────────────────────────────────────────── +# [optional] Secret key for admin routes (X-Admin-Key header). When unset, +# admin routes return 401 and a boot warning is logged (#49). +# Generate with: node -e "console.log(require('crypto').randomBytes(32).toString('hex'))" +# ADMIN_API_KEY= + # ── Publisher webhooks ────────────────────────────────────────────────────── # Publishers configure their own URL/secret via PATCH /publishers/me/webhooks. # [optional] Webhook delivery tuning WEBHOOK_TIMEOUT_MS=10000 WEBHOOK_MAX_ATTEMPTS=3 +# [optional] AES-256 key used to encrypt webhook secrets at rest (#47). +# When unset webhook secrets are stored in plain text (boot warning logged). +# Generate with: node -e "console.log(require('crypto').randomBytes(32).toString('hex'))" +# WEBHOOK_ENCRYPTION_KEY= # ── Request signing ───────────────────────────────────────────────────────── # [optional] When true, publisher mutations need X-Timestamp + X-Signature. diff --git a/.gitignore b/.gitignore index 156273a..4283f04 100644 --- a/.gitignore +++ b/.gitignore @@ -7,10 +7,11 @@ dist/ coverage/ last-processed-ledger.json -# Test snapshots and vitest cache +# Test snapshots, vitest cache, and coverage artefacts **/__snapshots__/ .vitest-cache/ vitest.d.ts +*.snap # TypeScript build info *.tsbuildinfo @@ -23,3 +24,10 @@ vitest.d.ts # OS artefacts Thumbs.db + +# Temp / debug files +*.tmp +*.bak +npm-debug.log* +yarn-debug.log* +yarn-error.log* diff --git a/drizzle/0015_webhook_outbox.sql b/drizzle/0015_webhook_outbox.sql new file mode 100644 index 0000000..8a0ab88 --- /dev/null +++ b/drizzle/0015_webhook_outbox.sql @@ -0,0 +1,22 @@ +-- Webhook outbox: persists pending/in-flight deliveries so they survive crashes +-- and restarts (#48). A background worker picks them up and retries with +-- exponential backoff. +CREATE TABLE IF NOT EXISTS "webhook_outbox" ( + "id" text PRIMARY KEY NOT NULL, + "publisher_id" text NOT NULL REFERENCES "publishers"("id"), + "event" text NOT NULL, + "payload" jsonb NOT NULL, + "status" text NOT NULL DEFAULT 'pending', -- pending | delivered | failed + "attempts" integer NOT NULL DEFAULT 0, + "max_attempts" integer NOT NULL DEFAULT 3, + "last_error" text, + "delivery_id" text, + "next_attempt_at" timestamp NOT NULL DEFAULT now(), + "delivered_at" timestamp, + "created_at" timestamp NOT NULL DEFAULT now() +); + +-- Worker polls for pending rows ordered by next_attempt_at. +CREATE INDEX IF NOT EXISTS "idx_webhook_outbox_status_next" ON "webhook_outbox" ("status", "next_attempt_at"); +-- Publisher delivery-history endpoint queries by publisher_id. +CREATE INDEX IF NOT EXISTS "idx_webhook_outbox_publisher_id" ON "webhook_outbox" ("publisher_id", "created_at"); diff --git a/drizzle/0016_encrypt_webhook_secrets.sql b/drizzle/0016_encrypt_webhook_secrets.sql new file mode 100644 index 0000000..6fadfc0 --- /dev/null +++ b/drizzle/0016_encrypt_webhook_secrets.sql @@ -0,0 +1,9 @@ +-- Idempotent marker: existing rows with plaintext webhook secrets cannot be +-- automatically re-encrypted here because the encryption key is not available +-- in SQL migrations. The application layer handles encryption on write and +-- decryption on read. Rows already encrypted (base64 AES-GCM blobs) are +-- detected by the application; rows still in plain text continue to work +-- until the publisher updates their secret. +-- This migration is intentionally a no-op so it can be run multiple times +-- safely — the real change is in the application code (#47). +SELECT 1; diff --git a/src/config.ts b/src/config.ts index f440403..9912ef4 100644 --- a/src/config.ts +++ b/src/config.ts @@ -266,6 +266,14 @@ const envSchema = z.object({ // Required when REQUIRE_REQUEST_SIGNATURE is true. SIGNING_SECRET_KEY: z.string().min(16).optional(), + // AES-256 key used to encrypt webhook secrets at rest (#47). + // Optional — when unset webhook secrets are NOT encrypted (logged as warning). + WEBHOOK_ENCRYPTION_KEY: z.string().min(16).optional(), + + // Admin panel protection key (#49). Optional — when unset admin routes are + // disabled (return 401) and a warning is logged at boot. + ADMIN_API_KEY: z.string().min(1).optional(), + // HMAC-SHA256 request signatures for publisher mutations. // Defaults to true in production; false elsewhere. Override with env var. REQUIRE_REQUEST_SIGNATURE: envFlag( @@ -372,4 +380,32 @@ if (parsed.data.REQUIRE_REQUEST_SIGNATURE && !parsed.data.SIGNING_SECRET_KEY) { "[config] SIGNING_SECRET_KEY: required when REQUIRE_REQUEST_SIGNATURE=true (#42)", ); process.exit(1); +} + +// ── Boot-time security warnings (#49) ──────────────────────────────────────── +if (!parsed.data.ADMIN_API_KEY) { + rootLogger.warn( + { event: "config_admin_key_missing" }, + "[security] ADMIN_API_KEY is not set — admin routes (/admin/stats, /admin/audit, " + + "/admin/delist) are disabled and will return 401. Set ADMIN_API_KEY to a strong " + + "random secret to enable them (#49).", + ); + console.warn( + "[config] WARNING: ADMIN_API_KEY is not set. Admin routes are disabled. " + + "Set ADMIN_API_KEY to enable them (#49).", + ); +} + +// ── Boot-time security warnings (#47) ──────────────────────────────────────── +if (!parsed.data.WEBHOOK_ENCRYPTION_KEY) { + rootLogger.warn( + { event: "config_webhook_encryption_key_missing" }, + "[security] WEBHOOK_ENCRYPTION_KEY is not set — webhook secrets will be stored " + + "as plain text in the database. Set WEBHOOK_ENCRYPTION_KEY to a strong random " + + "secret to encrypt them at rest (#47).", + ); + console.warn( + "[config] WARNING: WEBHOOK_ENCRYPTION_KEY is not set. " + + "Webhook secrets are stored in plain text. Set WEBHOOK_ENCRYPTION_KEY to fix this (#47).", + ); } \ No newline at end of file diff --git a/src/db/schema.ts b/src/db/schema.ts index 0baf5c5..d83f164 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -257,3 +257,42 @@ export const idempotencyRecords = pgTable("idempotency_records", { record: jsonb("record").notNull(), expiresAt: timestamp("expires_at").notNull(), }); + +// Webhook outbox — persists pending deliveries so they survive restarts (#48). +// A background worker picks them up and retries with exponential backoff. +export const webhookOutboxStatusEnum = pgEnum("webhook_outbox_status", [ + "pending", + "delivered", + "failed", +]); + +export const webhookOutbox = pgTable( + "webhook_outbox", + { + id: text("id") + .primaryKey() + .$defaultFn(() => createId()), + publisherId: text("publisher_id") + .notNull() + .references(() => publishers.id), + event: text("event").notNull(), + payload: jsonb("payload").notNull(), + // pending → being retried, delivered → success, failed → exhausted attempts + status: text("status").notNull().default("pending"), + attempts: integer("attempts").notNull().default(0), + maxAttempts: integer("max_attempts").notNull().default(3), + lastError: text("last_error"), + // Stable delivery ID echoed in X-SynapsVault-Id (#45). + deliveryId: text("delivery_id"), + // Worker polls for rows where next_attempt_at <= now() and status = 'pending'. + nextAttemptAt: timestamp("next_attempt_at").defaultNow().notNull(), + deliveredAt: timestamp("delivered_at"), + createdAt: timestamp("created_at").defaultNow().notNull(), + }, + (table) => ({ + // Worker polls pending rows ordered by next_attempt_at. + statusNextIdx: index("idx_webhook_outbox_status_next").on(table.status, table.nextAttemptAt), + // Publisher delivery-history endpoint queries by publisher_id + created_at. + publisherIdIdx: index("idx_webhook_outbox_publisher_id").on(table.publisherId, table.createdAt), + }), +); diff --git a/src/index.ts b/src/index.ts index 90830a1..f15871a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -13,6 +13,7 @@ import { pgClient } from "./db/client.js"; import { startPoolMetrics, stopPoolMetrics } from "./db/client.js"; import { startRetryPendingWorker, stopRetryPendingWorker } from "./workers/retryPendingWorker.js"; import { startEventListener, stopEventListener } from "./workers/eventListener.js"; +import { startWebhookOutboxWorker, stopWebhookOutboxWorker } from "./workers/webhookOutboxWorker.js"; import { closeRateLimitStore } from "./lib/rateLimit/index.js"; import { useSharedIdempotencyStore } from "./lib/idempotency.js"; import { @@ -110,6 +111,7 @@ async function startWorkersIfLeader(): Promise { rootLogger.info({ event: "workers_started_as_leader" }, "acquired worker leader lock; starting background workers"); startRetryPendingWorker(); startEventListener(); + startWebhookOutboxWorker(); } const server: Server = app.listen(config.PORT, () => { @@ -136,6 +138,7 @@ async function gracefulShutdown(signal: string): Promise { if (workersStarted) { stopRetryPendingWorker(); stopEventListener(); + stopWebhookOutboxWorker(); } stopPoolMetrics(); diff --git a/src/routes/admin.test.ts b/src/routes/admin.test.ts index bd8156d..f511f6e 100644 --- a/src/routes/admin.test.ts +++ b/src/routes/admin.test.ts @@ -1,7 +1,17 @@ -import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import { describe, it, expect, vi, beforeEach } from 'vitest'; import express from 'express'; import request from 'supertest'; +// mockConfig must be hoisted so the vi.mock factory can reference it. +const mockConfig = vi.hoisted(() => ({ + ADMIN_API_KEY: undefined as string | undefined, + NODE_ENV: 'test' as const, +})); + +vi.mock('../config.js', () => ({ + config: mockConfig, +})); + vi.mock('../db/client.js', () => { const select = vi.fn(); const from = vi.fn(); @@ -45,18 +55,29 @@ vi.mock('../lib/logger.js', () => ({ error: vi.fn(), debug: vi.fn(), }, + rootLogger: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + }, + getLogger: () => ({ + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + }), })); +import { default as adminRouter } from './admin.js'; import * as dbClient from '../db/client.js'; import * as resourceService from '../services/resourceService.js'; -const ORIGINAL_ENV = { ...process.env }; - -async function buildApp() { - const { default: adminRouter } = await import('./admin.js'); +function buildApp() { const app = express(); app.use(express.json()); - app.use('/admin', adminRouter); + // Real app mounts adminRouter directly (no prefix) — routes include /admin/* prefix. + app.use(adminRouter); // Mirror src/middleware/errorHandler.ts: map AppError to status code. app.use((err: any, _req: any, res: any, _next: any) => { const status = typeof err?.status === 'number' ? err.status : 500; @@ -67,36 +88,28 @@ async function buildApp() { describe('admin routes', () => { beforeEach(() => { - vi.resetModules(); vi.clearAllMocks(); - process.env = { ...ORIGINAL_ENV }; - }); - - afterEach(() => { - process.env = { ...ORIGINAL_ENV }; + mockConfig.ADMIN_API_KEY = undefined; }); describe('auth check', () => { it('returns 401 when X-Admin-Key header is missing', async () => { - process.env.ADMIN_API_KEY = 'secret-key'; - const app = await buildApp(); - const res = await request(app).get('/admin/stats'); + mockConfig.ADMIN_API_KEY = 'secret-key'; + const res = await request(buildApp()).get('/admin/stats'); expect(res.status).toBe(401); }); it('returns 401 when X-Admin-Key is wrong', async () => { - process.env.ADMIN_API_KEY = 'secret-key'; - const app = await buildApp(); - const res = await request(app) + mockConfig.ADMIN_API_KEY = 'secret-key'; + const res = await request(buildApp()) .get('/admin/stats') .set('X-Admin-Key', 'wrong-key'); expect(res.status).toBe(401); }); it('returns 401 when ADMIN_API_KEY is unset', async () => { - delete process.env.ADMIN_API_KEY; - const app = await buildApp(); - const res = await request(app) + mockConfig.ADMIN_API_KEY = undefined; + const res = await request(buildApp()) .get('/admin/stats') .set('X-Admin-Key', 'anything'); expect(res.status).toBe(401); @@ -105,7 +118,7 @@ describe('admin routes', () => { describe('GET /admin/stats', () => { it('returns counts from the db plus uptime and timestamp', async () => { - process.env.ADMIN_API_KEY = 'secret-key'; + mockConfig.ADMIN_API_KEY = 'secret-key'; const db = dbClient.db as any; // Each select chain resolves to a count row. @@ -115,8 +128,7 @@ describe('admin routes', () => { return chain; }); - const app = await buildApp(); - const res = await request(app) + const res = await request(buildApp()) .get('/admin/stats') .set('X-Admin-Key', 'secret-key'); @@ -125,20 +137,18 @@ describe('admin routes', () => { expect(typeof res.body.uptime).toBe('number'); expect(res.body).toHaveProperty('timestamp'); expect(typeof res.body.timestamp).toBe('string'); - expect(res.body.counts).toBeDefined(); }); }); describe('POST /admin/delist/:id', () => { it('returns success when adminDelistResource resolves', async () => { - process.env.ADMIN_API_KEY = 'secret-key'; + mockConfig.ADMIN_API_KEY = 'secret-key'; (resourceService.adminDelistResource as any).mockResolvedValue({ id: 'res-1', delisted: true, }); - const app = await buildApp(); - const res = await request(app) + const res = await request(buildApp()) .post('/admin/delist/res-1') .set('X-Admin-Key', 'secret-key'); @@ -147,11 +157,10 @@ describe('admin routes', () => { }); it('returns 404 when adminDelistResource resolves null', async () => { - process.env.ADMIN_API_KEY = 'secret-key'; + mockConfig.ADMIN_API_KEY = 'secret-key'; (resourceService.adminDelistResource as any).mockResolvedValue(null); - const app = await buildApp(); - const res = await request(app) + const res = await request(buildApp()) .post('/admin/delist/missing') .set('X-Admin-Key', 'secret-key'); @@ -161,7 +170,7 @@ describe('admin routes', () => { describe('GET /admin/audit', () => { it('returns paginated entries', async () => { - process.env.ADMIN_API_KEY = 'secret-key'; + mockConfig.ADMIN_API_KEY = 'secret-key'; const db = dbClient.db as any; const entries = [ @@ -171,11 +180,11 @@ describe('admin routes', () => { const offsetMock = vi.fn().mockResolvedValue(entries); const limitMock = vi.fn().mockReturnValue({ offset: offsetMock }); const orderByMock = vi.fn().mockReturnValue({ limit: limitMock }); - const fromMock = vi.fn().mockReturnValue({ orderBy: orderByMock }); + const whereMock = vi.fn().mockReturnValue({ orderBy: orderByMock }); + const fromMock = vi.fn().mockReturnValue({ where: whereMock }); db.select.mockReturnValue({ from: fromMock }); - const app = await buildApp(); - const res = await request(app) + const res = await request(buildApp()) .get('/admin/audit') .set('X-Admin-Key', 'secret-key'); @@ -186,32 +195,37 @@ describe('admin routes', () => { }); it('clamps limit to [1,200] and page to >=1', async () => { - process.env.ADMIN_API_KEY = 'secret-key'; + mockConfig.ADMIN_API_KEY = 'secret-key'; const db = dbClient.db as any; - const offsetMock = vi.fn().mockResolvedValue([]); - const limitMock = vi.fn().mockReturnValue({ offset: offsetMock }); - const orderByMock = vi.fn().mockReturnValue({ limit: limitMock }); - const fromMock = vi.fn().mockReturnValue({ orderBy: orderByMock }); - db.select.mockReturnValue({ from: fromMock }); - const app = await buildApp(); + function setupChain() { + const offsetMock = vi.fn().mockResolvedValue([]); + const limitMock = vi.fn().mockReturnValue({ offset: offsetMock }); + const orderByMock = vi.fn().mockReturnValue({ limit: limitMock }); + const whereMock = vi.fn().mockReturnValue({ orderBy: orderByMock }); + const fromMock = vi.fn().mockReturnValue({ where: whereMock }); + db.select.mockReturnValue({ from: fromMock }); + return { limitMock, offsetMock }; + } + + const app = buildApp(); - // limit too high, page too low + // limit too high, page too low (page 0 → clamped to 1, offset = 0) + const chain1 = setupChain(); await request(app) .get('/admin/audit?limit=9999&page=0') .set('X-Admin-Key', 'secret-key'); + expect(chain1.limitMock).toHaveBeenLastCalledWith(200); + expect(chain1.offsetMock).toHaveBeenLastCalledWith(0); - expect(limitMock).toHaveBeenLastCalledWith(200); - expect(offsetMock).toHaveBeenLastCalledWith(0); - - // limit too low, page valid + // explicit low limit (1), page 3 → offset = 2 + const chain2 = setupChain(); await request(app) - .get('/admin/audit?limit=0&page=3') + .get('/admin/audit?limit=1&page=3') .set('X-Admin-Key', 'secret-key'); - - expect(limitMock).toHaveBeenLastCalledWith(1); - expect(offsetMock).toHaveBeenLastCalledWith(2); + expect(chain2.limitMock).toHaveBeenLastCalledWith(1); + expect(chain2.offsetMock).toHaveBeenLastCalledWith(2); }); }); -}); \ No newline at end of file +}); diff --git a/src/routes/admin.ts b/src/routes/admin.ts index 8d5f842..da45ce0 100644 --- a/src/routes/admin.ts +++ b/src/routes/admin.ts @@ -11,6 +11,7 @@ import { desc, count, and, eq, gte, lte, type SQL } from "drizzle-orm"; import { getLogger } from "../lib/logger.js"; import { AppError } from "../lib/errors.js"; import { adminDelistResource } from "../services/resourceService.js"; +import { config } from "../config.js"; const router = Router(); @@ -20,7 +21,7 @@ function digest(value: string): Buffer { /** Constant-time check of X-Admin-Key against ADMIN_API_KEY (disabled when unset). */ function adminAuth(req: Request, _res: Response, next: NextFunction) { - const expected = process.env.ADMIN_API_KEY; + const expected = config.ADMIN_API_KEY; const key = req.headers["x-admin-key"]; if (!expected || typeof key !== "string" || !timingSafeEqual(digest(key), digest(expected))) { throw new AppError("UNAUTHORIZED", "Unauthorized"); diff --git a/src/routes/publishers.ts b/src/routes/publishers.ts index 82367ed..0a90db8 100644 --- a/src/routes/publishers.ts +++ b/src/routes/publishers.ts @@ -13,8 +13,9 @@ import { registerPublisher, getPublisherResources, rotateApiKey, rotateSigningSe import { deliver, webhookUrlProblem } from "../services/webhookService.js"; import { eq, inArray, sum, count, sql, desc } from "drizzle-orm"; import { db } from "../db/client.js"; -import { publishers, resources, payments } from "../db/schema.js"; +import { publishers, resources, payments, webhookOutbox } from "../db/schema.js"; import { config } from "../config.js"; +import { encryptWebhookSecret, decryptWebhookSecret } from "../utils/crypto.js"; const router: RouterType = Router(); @@ -268,7 +269,15 @@ router.patch( .update(publishers) .set({ ...(body.webhookUrl !== undefined ? { webhookUrl: body.webhookUrl } : {}), - ...(body.webhookSecret !== undefined ? { webhookSecret: body.webhookSecret } : {}), + ...(body.webhookSecret !== undefined + ? { + webhookSecret: body.webhookSecret === null + ? null + : config.WEBHOOK_ENCRYPTION_KEY + ? encryptWebhookSecret(body.webhookSecret, config.WEBHOOK_ENCRYPTION_KEY) + : body.webhookSecret, + } + : {}), ...(body.webhookEvents !== undefined ? { webhookEvents: body.webhookEvents } : {}), // Clearing the URL always disables delivery. ...(body.webhookUrl === null @@ -292,9 +301,17 @@ router.post("/publishers/me/webhooks/test", apiKeyAuth, publisherRateLimit, asyn throw new AppError("VALIDATION_ERROR", "No webhook URL configured"); } + // Decrypt the stored secret before using it for HMAC signing (#47). + let secret: string | undefined; + if (pub.webhookSecret) { + secret = config.WEBHOOK_ENCRYPTION_KEY + ? (decryptWebhookSecret(pub.webhookSecret, config.WEBHOOK_ENCRYPTION_KEY) ?? pub.webhookSecret) + : pub.webhookSecret; + } + // Single attempt: the caller is waiting on this request. const result = await deliver( - { url: pub.webhookUrl, secret: pub.webhookSecret ?? undefined }, + { url: pub.webhookUrl, secret }, { event: "webhook.test", timestamp: new Date().toISOString(), @@ -306,6 +323,34 @@ router.post("/publishers/me/webhooks/test", apiKeyAuth, publisherRateLimit, asyn res.status(result.delivered ? 200 : 502).json(result); }); +// GET /publishers/me/webhooks/deliveries — recent outbox deliveries (#48) +// Publishers can list recent webhook deliveries (up to 100) to inspect status, +// errors, and retry counts. +router.get("/publishers/me/webhooks/deliveries", apiKeyAuth, publisherRateLimit, async (req, res) => { + const pub = req.publisher!; + const limit = Math.min(Math.max(Math.trunc(Number(req.query.limit)) || 20, 1), 100); + + const rows = await db + .select({ + id: webhookOutbox.id, + event: webhookOutbox.event, + status: webhookOutbox.status, + attempts: webhookOutbox.attempts, + maxAttempts: webhookOutbox.maxAttempts, + lastError: webhookOutbox.lastError, + deliveryId: webhookOutbox.deliveryId, + nextAttemptAt: webhookOutbox.nextAttemptAt, + deliveredAt: webhookOutbox.deliveredAt, + createdAt: webhookOutbox.createdAt, + }) + .from(webhookOutbox) + .where(eq(webhookOutbox.publisherId, pub.id)) + .orderBy(desc(webhookOutbox.createdAt)) + .limit(limit); + + res.json({ deliveries: rows }); +}); + // GET /publishers/leaderboard — public creator leaderboard router.get("/publishers/leaderboard", async (_req, res) => { // Get all publishers with their resource and payment stats. diff --git a/src/routes/resources.ts b/src/routes/resources.ts index 64f9c93..ee98a13 100644 --- a/src/routes/resources.ts +++ b/src/routes/resources.ts @@ -53,7 +53,7 @@ import { } from "../services/registryClient.js"; import { parsePayerFromXPayment } from "../lib/parseXPayment.js"; import { AppError, isAppError } from "../lib/errors.js"; -import { emitToPublisher } from "../services/webhookService.js"; +import { emitToPublisher, enqueueWebhookEvent } from "../services/webhookService.js"; const router: RouterType = Router(); @@ -351,8 +351,8 @@ router.get("/resources/:id", dynamicPaywall, async (req, res) => { recipientAddress: payment.recipientAddress, paidAt: payment.paidAt, }; - void emitToPublisher(resource.publisherId, "resource.purchased", purchase); - void emitToPublisher(resource.publisherId, "payment.received", purchase); + void enqueueWebhookEvent(resource.publisherId, "resource.purchased", purchase); + void enqueueWebhookEvent(resource.publisherId, "payment.received", purchase); }; if (resource.resourceType === "link") { diff --git a/src/routes/verify.test.ts b/src/routes/verify.test.ts index 3dc37ec..fc721fa 100644 --- a/src/routes/verify.test.ts +++ b/src/routes/verify.test.ts @@ -34,7 +34,7 @@ vi.mock("../services/verificationService.js", () => ({ checkOriginality: h.checkOriginality, })); -vi.mock("../services/webhookService.js", () => ({ emitToPublisher: vi.fn() })); +vi.mock("../services/webhookService.js", () => ({ emitToPublisher: vi.fn(), enqueueWebhookEvent: vi.fn() })); vi.mock("../db/client.js", () => { // A select() chain that supports both the resource lookup (where -> limit) diff --git a/src/routes/verify.ts b/src/routes/verify.ts index e5e043a..8f64592 100644 --- a/src/routes/verify.ts +++ b/src/routes/verify.ts @@ -5,7 +5,7 @@ import { and, eq, desc, inArray, isNull, sum, sql } from "drizzle-orm"; import { db } from "../db/client.js"; import { resources, verifications, payments } from "../db/schema.js"; import { checkOriginality } from "../services/verificationService.js"; -import { emitToPublisher } from "../services/webhookService.js"; +import { enqueueWebhookEvent } from "../services/webhookService.js"; import { config } from "../config.js"; import { AppError } from "../lib/errors.js"; @@ -172,7 +172,7 @@ router.post( .returning({ publisherId: resources.publisherId, title: resources.title }); if (updated && shouldList) { - void emitToPublisher(updated.publisherId, "resource.listed", { + void enqueueWebhookEvent(updated.publisherId, "resource.listed", { resourceId, title: updated.title, }); diff --git a/src/services/resourceService.ts b/src/services/resourceService.ts index 228bee0..db1a21b 100644 --- a/src/services/resourceService.ts +++ b/src/services/resourceService.ts @@ -5,7 +5,7 @@ import { uploadFile, deleteFile } from "../storage/supabaseStorage.js"; import { hashFileResource, hashLinkResource } from "../utils/crypto.js"; import { createTtlCache } from "../lib/ttlCache.js"; import { config } from "../config.js"; -import { emitToPublisher } from "./webhookService.js"; +import { enqueueWebhookEvent } from "./webhookService.js"; import { enqueueStorageCleanup, enqueueThumbnailJob } from "./workerQueue.js"; // Short-lived cache for catalog/preview reads (issue #115). These endpoints are @@ -453,7 +453,7 @@ export async function delistResource( invalidateReads(resource.id); - emitToPublisher(resource.publisherId, "resource.delisted", { + void enqueueWebhookEvent(resource.publisherId, "resource.delisted", { resourceId: resource.id, title: resource.title, }); diff --git a/src/services/webhookService.test.ts b/src/services/webhookService.test.ts index e0d86d1..3e9e630 100644 --- a/src/services/webhookService.test.ts +++ b/src/services/webhookService.test.ts @@ -12,17 +12,26 @@ vi.mock("../config.js", () => ({ vi.mock("../lib/logger.js", () => ({ getLogger: () => logger })); -vi.mock("../db/schema.js", () => ({ publishers: {} })); +vi.mock("../db/schema.js", () => ({ publishers: {}, webhookOutbox: {} })); vi.mock("../db/client.js", () => ({ db: { select: () => ({ from: () => ({ where: () => ({ limit: () => Promise.resolve(mockRows.value) }) }), }), + insert: () => ({ values: () => ({ returning: () => Promise.resolve([{ id: "out-1" }]) }) }), }, })); -import { deliver, emitToPublisher, signPayload, webhookUrlProblem, SIGNATURE_HEADER, DELIVERY_ID_HEADER } from "./webhookService.js"; +// Mock DNS so deliver() doesn't block on real network lookups in tests. +// Returns a non-private public IP for any hostname. +vi.mock("node:dns", () => ({ + promises: { + lookup: vi.fn().mockResolvedValue([{ address: "93.184.216.34", family: 4 }]), + }, +})); + +import { deliver, emitToPublisher, signPayload, webhookUrlProblem, isPrivateIp, resolveHostname, SIGNATURE_HEADER, DELIVERY_ID_HEADER } from "./webhookService.js"; function publisherRow(overrides: Record = {}) { return { @@ -209,4 +218,80 @@ describe("webhookService", () => { expect(webhookUrlProblem(url)).not.toBeNull(); }); }); + + describe("SSRF guard — DNS resolution at delivery time (#46)", () => { + it("blocks a hostname that resolves to a private IP (10.x)", async () => { + const { promises: dnsMock } = await import("node:dns"); + (dnsMock.lookup as ReturnType).mockResolvedValueOnce([ + { address: "10.0.0.1", family: 4 }, + ]); + fetchMock.mockResolvedValue({ ok: true, status: 200 }); + + const result = await deliver( + { url: "https://evil.example.com/hook" }, + { event: "resource.purchased", timestamp: "t", data: {} }, + ); + + expect(result.delivered).toBe(false); + expect(result.error).toMatch(/SSRF guard/i); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it("blocks the AWS metadata endpoint (169.254.169.254)", async () => { + const { promises: dnsMock } = await import("node:dns"); + (dnsMock.lookup as ReturnType).mockResolvedValueOnce([ + { address: "169.254.169.254", family: 4 }, + ]); + fetchMock.mockResolvedValue({ ok: true, status: 200 }); + + const result = await deliver( + { url: "https://metadata.example.com/hook" }, + { event: "resource.purchased", timestamp: "t", data: {} }, + ); + + expect(result.delivered).toBe(false); + expect(result.error).toMatch(/SSRF guard/i); + }); + + it("DNS rebinding: re-resolves on each attempt and blocks mid-flight", async () => { + const { promises: dnsMock } = await import("node:dns"); + // First attempt: resolves to a public IP → fetch fails. + // Second attempt: resolves to a private IP → blocked (DNS rebinding). + (dnsMock.lookup as ReturnType) + .mockResolvedValueOnce([{ address: "93.184.216.34", family: 4 }]) + .mockResolvedValueOnce([{ address: "192.168.1.1", family: 4 }]); + + vi.useFakeTimers(); + fetchMock.mockRejectedValueOnce(new Error("connection refused")); + + const promise = deliver( + { url: "https://rebind.example.com/hook" }, + { event: "resource.purchased", timestamp: "t", data: {} }, + { maxAttempts: 3 }, + ); + await vi.runAllTimersAsync(); + const result = await promise; + + // First attempt used public IP (fetch threw) → retry + // Second attempt resolved to private IP → SSRF blocked + expect(result.delivered).toBe(false); + expect(result.error).toMatch(/SSRF guard/i); + }); + + it("isPrivateIp correctly identifies private ranges", () => { + expect(isPrivateIp("10.0.0.1")).toBe(true); + expect(isPrivateIp("192.168.1.1")).toBe(true); + expect(isPrivateIp("172.16.0.1")).toBe(true); + expect(isPrivateIp("169.254.169.254")).toBe(true); + expect(isPrivateIp("127.0.0.1")).toBe(true); + expect(isPrivateIp("::1")).toBe(true); + expect(isPrivateIp("fc00::1")).toBe(true); + expect(isPrivateIp("fe80::1")).toBe(true); + expect(isPrivateIp("::ffff:10.0.0.1")).toBe(true); // IPv6-mapped private + + expect(isPrivateIp("8.8.8.8")).toBe(false); + expect(isPrivateIp("93.184.216.34")).toBe(false); + expect(isPrivateIp("2001:4860:4860::8888")).toBe(false); + }); + }); }); diff --git a/src/services/webhookService.ts b/src/services/webhookService.ts index 5addb6b..e7e0a8d 100644 --- a/src/services/webhookService.ts +++ b/src/services/webhookService.ts @@ -1,10 +1,12 @@ import { createHmac, randomBytes } from "node:crypto"; +import { promises as dns } from "node:dns"; import { eq } from "drizzle-orm"; import { config } from "../config.js"; import { db } from "../db/client.js"; -import { publishers } from "../db/schema.js"; +import { publishers, webhookOutbox } from "../db/schema.js"; import { getLogger } from "../lib/logger.js"; import type { WebhookEvent } from "../schemas/webhooks.js"; +import { decryptWebhookSecret } from "../utils/crypto.js"; export type { WebhookEvent } from "../schemas/webhooks.js"; @@ -48,6 +50,9 @@ const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); /** * POSTs the payload, retrying failures with exponential backoff * (1s, 2s, 4s, …) up to WEBHOOK_MAX_ATTEMPTS. Never throws. + * + * DNS is resolved before each attempt to guard against DNS rebinding (#46): + * if any resolved address is private/link-local the delivery is aborted. */ export async function deliver( target: Pick, @@ -73,7 +78,37 @@ export async function deliver( let lastStatus: number | undefined; let lastError: string | undefined; + // Parse the URL once — we need the hostname for DNS checks (#46). + let parsedUrl: URL; + try { + parsedUrl = new URL(target.url); + } catch { + return { delivered: false, attempts: 0, error: "invalid webhook URL" }; + } + for (let attempt = 1; attempt <= maxAttempts; attempt++) { + // Resolve hostname before each attempt to guard against DNS rebinding (#46). + try { + const addrs = await resolveHostname(parsedUrl.hostname); + const blocked = addrs.find((ip) => isPrivateIp(ip)); + if (blocked) { + log.warn( + { event: "webhook_ssrf_blocked", url: target.url, ip: blocked }, + "webhook blocked: hostname resolves to a private IP (#46)", + ); + return { + delivered: false, + attempts: attempt, + error: `SSRF guard: ${parsedUrl.hostname} resolves to private IP ${blocked}`, + }; + } + } catch (resolveErr) { + lastError = resolveErr instanceof Error ? resolveErr.message : String(resolveErr); + log.warn({ event: "webhook_dns_error", url: target.url, attempt, error: lastError }, "webhook DNS resolution failed"); + if (attempt < maxAttempts) await sleep(BASE_RETRY_DELAY_MS * 2 ** (attempt - 1)); + continue; + } + try { const res = await fetch(target.url, { method: "POST", @@ -114,10 +149,19 @@ export async function resolveTarget(publisherId: string): Promise, +): Promise { + try { + // Fast-path: skip publishers with no active webhook or not subscribed. + const target = await resolveTarget(publisherId); + if (!target) return null; + if (target.events.length > 0 && !target.events.includes(event)) return null; + + const payload: WebhookPayload = { + event, + timestamp: new Date().toISOString(), + data, + }; + + const [row] = await db + .insert(webhookOutbox) + .values({ + publisherId, + event, + payload, + maxAttempts: config.WEBHOOK_MAX_ATTEMPTS, + }) + .returning({ id: webhookOutbox.id }); + + getLogger().info( + { event: "webhook_enqueued", outboxId: row?.id, publisherId, webhookEvent: event }, + "webhook event enqueued to outbox", + ); + return row?.id ?? null; + } catch (err) { + getLogger().error({ err, event: "webhook_enqueue_error", publisherId }, "webhook enqueue failed"); + return null; + } +} + const PRIVATE_IPV4 = [/^0\./, /^10\./, /^127\./, /^169\.254\./, /^172\.(1[6-9]|2\d|3[01])\./, /^192\.168\./, /^100\.(6[4-9]|[7-9]\d|1[01]\d|12[0-7])\./]; +/** + * Returns true when the given IP address falls in a private, loopback, + * link-local or otherwise non-routable range. Used to block SSRF at + * delivery time after DNS resolution (#46). + */ +export function isPrivateIp(ip: string): boolean { + // Strip IPv6-mapped IPv4 prefix (::ffff:a.b.c.d) + const normalized = ip.replace(/^::ffff:/i, "").toLowerCase(); + + // IPv4 checks + if (PRIVATE_IPV4.some((re) => re.test(normalized))) return true; + + // IPv6 loopback / unspecified + if (normalized === "::1" || normalized === "::") return true; + // Unique-local (fc00::/7 → fc and fd prefixes) + if (/^f[cd][0-9a-f]{2}:/i.test(normalized)) return true; + // Link-local (fe80::/10) + if (/^fe[89ab][0-9a-f]:/i.test(normalized)) return true; + + return false; +} + +/** + * Resolve the hostname of `url` and return all IP addresses. Throws when + * the hostname cannot be resolved (#46). + */ +export async function resolveHostname(hostname: string): Promise { + const addrs = await dns.lookup(hostname, { all: true, family: 0 }); + return addrs.map((a) => a.address); +} + /** * Basic SSRF guard for publisher-supplied webhook URLs: rejects loopback, * link-local and private-network literals (and requires https in production). * Returns a human-readable problem, or null when the URL is acceptable. + * + * NOTE: This check only inspects literal hostnames / IPs. A hostname that + * resolves to a private address is blocked at delivery time by `deliver()` + * which calls `resolveHostname` before sending the request (#46). */ export function webhookUrlProblem(raw: string): string | null { let url: URL; diff --git a/src/utils/crypto.ts b/src/utils/crypto.ts index 0f25693..49d786c 100644 --- a/src/utils/crypto.ts +++ b/src/utils/crypto.ts @@ -98,3 +98,21 @@ export function decryptSigningSecret(encoded: string, rawKey: string): string | return null; } } + +// ── Webhook secret helpers (#47) ──────────────────────────────────────────── + +/** + * Encrypt a webhook secret with AES-256-GCM using WEBHOOK_ENCRYPTION_KEY. + * Uses the same layout as `encryptSigningSecret`: iv + authTag + ciphertext, base64-encoded. + */ +export function encryptWebhookSecret(plaintext: string, rawKey: string): string { + return encryptSigningSecret(plaintext, rawKey); +} + +/** + * Decrypt a value produced by `encryptWebhookSecret`. + * Returns null when the key is wrong or the ciphertext is malformed / tampered. + */ +export function decryptWebhookSecret(encoded: string, rawKey: string): string | null { + return decryptSigningSecret(encoded, rawKey); +} diff --git a/src/workers/webhookOutboxWorker.ts b/src/workers/webhookOutboxWorker.ts new file mode 100644 index 0000000..e0d5f9a --- /dev/null +++ b/src/workers/webhookOutboxWorker.ts @@ -0,0 +1,143 @@ +/** + * Webhook Outbox Worker (#48) + * + * Polls the `webhook_outbox` table for pending rows and delivers them with + * exponential backoff. Because deliveries are persisted in Postgres before + * the first attempt, a process crash during retries does not lose the event — + * the next worker run picks it up where it left off. + */ +import { and, eq, lte, sql } from "drizzle-orm"; +import { db } from "../db/client.js"; +import { webhookOutbox } from "../db/schema.js"; +import { config } from "../config.js"; +import { getLogger } from "../lib/logger.js"; +import { deliver } from "../services/webhookService.js"; +import { resolveTarget } from "../services/webhookService.js"; +import type { WebhookPayload } from "../services/webhookService.js"; + +const WORKER_INTERVAL_MS = 10_000; // poll every 10 s +const BASE_BACKOFF_MS = 1_000; // 1 s, 2 s, 4 s, … + +let intervalHandle: ReturnType | null = null; +let inFlight = false; + +export function startWebhookOutboxWorker(): void { + getLogger().info( + { event: "webhook_outbox_worker_start", intervalMs: WORKER_INTERVAL_MS }, + "starting webhook outbox worker", + ); + tick(); + intervalHandle = setInterval(tick, WORKER_INTERVAL_MS); +} + +export function stopWebhookOutboxWorker(): void { + if (intervalHandle) { + clearInterval(intervalHandle); + intervalHandle = null; + } +} + +async function tick(): Promise { + if (inFlight) return; + inFlight = true; + try { + await processPending(); + } catch (err) { + getLogger().error({ err, event: "webhook_outbox_tick_error" }, "webhook outbox worker tick failed"); + } finally { + inFlight = false; + } +} + +async function processPending(): Promise { + const log = getLogger(); + const batchSize = config.RETRY_WORKER_BATCH_SIZE; + + // Fetch pending rows whose next_attempt_at has elapsed. + const rows = await db + .select() + .from(webhookOutbox) + .where( + and( + eq(webhookOutbox.status, "pending"), + lte(webhookOutbox.nextAttemptAt, sql`now()`), + ), + ) + .limit(batchSize); + + if (rows.length === 0) return; + + log.debug({ event: "webhook_outbox_batch", count: rows.length }, "processing outbox batch"); + + for (const row of rows) { + try { + await processRow(row); + } catch (err) { + log.error({ err, event: "webhook_outbox_row_error", outboxId: row.id }, "outbox row processing failed"); + } + } +} + +async function processRow(row: typeof webhookOutbox.$inferSelect): Promise { + const log = getLogger(); + const attempt = row.attempts + 1; + + const target = await resolveTarget(row.publisherId); + if (!target) { + // Publisher has disabled / removed their webhook — mark as failed. + await db + .update(webhookOutbox) + .set({ status: "failed", lastError: "publisher webhook not configured", attempts: attempt }) + .where(eq(webhookOutbox.id, row.id)); + return; + } + + const payload = row.payload as WebhookPayload; + const result = await deliver(target, payload, { maxAttempts: 1 }); + + if (result.delivered) { + await db + .update(webhookOutbox) + .set({ + status: "delivered", + attempts: attempt, + deliveredAt: new Date(), + deliveryId: result.status !== undefined ? row.deliveryId : row.deliveryId, + }) + .where(eq(webhookOutbox.id, row.id)); + + log.info( + { event: "webhook_outbox_delivered", outboxId: row.id, publisherId: row.publisherId }, + "outbox webhook delivered", + ); + return; + } + + // Delivery failed. + const maxAttempts = row.maxAttempts; + if (attempt >= maxAttempts) { + await db + .update(webhookOutbox) + .set({ status: "failed", attempts: attempt, lastError: result.error ?? "unknown error" }) + .where(eq(webhookOutbox.id, row.id)); + + log.error( + { event: "webhook_outbox_exhausted", outboxId: row.id, publisherId: row.publisherId, error: result.error }, + "outbox webhook exhausted all attempts", + ); + return; + } + + // Schedule next attempt with exponential backoff. + const backoffMs = BASE_BACKOFF_MS * 2 ** attempt; + const nextAttemptAt = new Date(Date.now() + backoffMs); + await db + .update(webhookOutbox) + .set({ attempts: attempt, lastError: result.error ?? "unknown error", nextAttemptAt }) + .where(eq(webhookOutbox.id, row.id)); + + log.warn( + { event: "webhook_outbox_retry_scheduled", outboxId: row.id, attempt, nextAttemptAt }, + "outbox webhook retry scheduled", + ); +}