diff --git a/src/__tests__/webhookDelivery.test.ts b/src/__tests__/webhookDelivery.test.ts new file mode 100644 index 0000000..5ef4cee --- /dev/null +++ b/src/__tests__/webhookDelivery.test.ts @@ -0,0 +1,263 @@ +import { + signPayload, + verifySignature, + backoffDelay, + deliverWebhook, + listDeadLetter, + clearDeadLetter, + replayDeadLetter, +} from "../services/webhookDelivery"; + +const SECRET = "test-secret"; + +// ─── Mock fetch ─────────────────────────────────────────────────────────────── + +function mockFetch( + responses: Array<{ status: number; ok: boolean }>, + delay = 0, +): typeof fetch { + let callIndex = 0; + return (async (_url: URL | string, _init: RequestInit) => { + if (delay > 0) await new Promise((r) => setTimeout(r, delay)); + const resp = responses[Math.min(callIndex, responses.length - 1)]; + callIndex++; + return { + ok: resp.ok, + status: resp.status, + statusText: `Status ${resp.status}`, + headers: new Map(), + json: async () => ({}), + text: async () => "", + } as Response; + }) as typeof fetch; +} + +// ─── Signing ────────────────────────────────────────────────────────────────── + +describe("signPayload", () => { + it("produces a t=...,v1=... header", () => { + const sig = signPayload(1234567890, '{"test":true}', SECRET); + expect(sig).toMatch(/^t=1234567890,v1=[0-9a-f]{64}$/); + }); + + it("produces deterministic signatures for the same input", () => { + const sig1 = signPayload(1234567890, '{"test":true}', SECRET); + const sig2 = signPayload(1234567890, '{"test":true}', SECRET); + expect(sig1).toBe(sig2); + }); + + it("produces different signatures for different payloads", () => { + const sig1 = signPayload(1234567890, '{"a":1}', SECRET); + const sig2 = signPayload(1234567890, '{"a":2}', SECRET); + expect(sig1).not.toBe(sig2); + }); +}); + +describe("verifySignature", () => { + const body = '{"event":"test","payload":{"foo":"bar"}}'; + const timestamp = Math.floor(Date.now() / 1000); + + it("verifies a valid signature", () => { + const sig = signPayload(timestamp, body, SECRET); + expect(verifySignature(sig, body, SECRET)).toBe(true); + }); + + it("rejects a tampered body", () => { + const sig = signPayload(timestamp, body, SECRET); + expect(verifySignature(sig, '{"event":"test","payload":{"foo":"HACKED"}}', SECRET)).toBe(false); + }); + + it("rejects a wrong secret", () => { + const sig = signPayload(timestamp, body, SECRET); + expect(verifySignature(sig, body, "wrong-secret")).toBe(false); + }); + + it("rejects an expired timestamp", () => { + const oldTs = Math.floor(Date.now() / 1000) - 600; // 10 min ago + const sig = signPayload(oldTs, body, SECRET); + expect(verifySignature(sig, body, SECRET, 300)).toBe(false); + }); + + it("rejects a malformed header", () => { + expect(verifySignature("garbage", body, SECRET)).toBe(false); + expect(verifySignature("t=123", body, SECRET)).toBe(false); + expect(verifySignature("v1=abc", body, SECRET)).toBe(false); + expect(verifySignature("", body, SECRET)).toBe(false); + }); +}); + +// ─── Backoff ────────────────────────────────────────────────────────────────── + +describe("backoffDelay", () => { + it("increases exponentially with attempt number", () => { + const d1 = backoffDelay(1); + const d2 = backoffDelay(2); + const d3 = backoffDelay(3); + // Due to jitter, just check rough ordering + expect(d1).toBeLessThanOrEqual(d2 + 5000); // jitter can cause overlap + expect(d2).toBeLessThanOrEqual(d3 + 10000); + }); + + it("caps at MAX_BACKOFF_MS", () => { + const d = backoffDelay(20); // very high attempt + expect(d).toBeLessThanOrEqual(37500); // 30s + 25% jitter + }); +}); + +// ─── Delivery ───────────────────────────────────────────────────────────────── + +beforeEach(() => { + clearDeadLetter(); +}); + +describe("deliverWebhook", () => { + const origBackoff = process.env.WEBHOOK_INITIAL_BACKOFF_MS; + beforeEach(() => { process.env.WEBHOOK_INITIAL_BACKOFF_MS = "0"; }); + afterEach(() => { if (origBackoff === undefined) delete process.env.WEBHOOK_INITIAL_BACKOFF_MS; else process.env.WEBHOOK_INITIAL_BACKOFF_MS = origBackoff; }); + it("delivers successfully on first attempt (200)", async () => { + const fetchImpl = mockFetch([{ status: 200, ok: true }]); + const attempts = await deliverWebhook( + "wh_test", + "https://example.com/hook", + "pair.registered", + { pair: "XLM/USDC" }, + fetchImpl, + ); + expect(attempts).toHaveLength(1); + expect(attempts[0].status).toBe("success"); + expect(attempts[0].statusCode).toBe(200); + }); + + it("retries on 5xx then succeeds", async () => { + const fetchImpl = mockFetch([ + { status: 503, ok: false }, + { status: 200, ok: true }, + ]); + const attempts = await deliverWebhook( + "wh_test", + "https://example.com/hook", + "pair.registered", + { pair: "XLM/USDC" }, + fetchImpl, + ); + expect(attempts.length).toBe(2); + expect(attempts[0].status).toBe("retry"); + expect(attempts[0].statusCode).toBe(503); + expect(attempts[1].status).toBe("success"); + }); + + it("does not retry on 4xx (dead-letters immediately)", async () => { + const fetchImpl = mockFetch([{ status: 404, ok: false }]); + const attempts = await deliverWebhook( + "wh_test", + "https://example.com/hook", + "pair.registered", + {}, + fetchImpl, + ); + expect(attempts).toHaveLength(1); + expect(attempts[0].status).toBe("dead_letter"); + expect(attempts[0].statusCode).toBe(404); + expect(listDeadLetter()).toHaveLength(1); + expect(listDeadLetter()[0].failureReason).toContain("4xx"); + }); + + it("retries on timeout/error then dead-letters after max attempts", async () => { + process.env.WEBHOOK_MAX_ATTEMPTS = "2"; + // Always throws (simulating network failure) + const fetchImpl = (async () => { + throw new Error("ECONNREFUSED"); + }) as typeof fetch; + const attempts = await deliverWebhook( + "wh_test", + "https://example.com/hook", + "pair.registered", + {}, + fetchImpl, + ); + delete process.env.WEBHOOK_MAX_ATTEMPTS; + expect(attempts.length).toBe(2); + expect(attempts[attempts.length - 1].status).toBe("dead_letter"); + expect(listDeadLetter()).toHaveLength(1); + expect(listDeadLetter()[0].failureReason).toContain("exhausted"); + }); + + it("signs each delivery attempt with X-Signature header", async () => { + let receivedHeaders: Record = {}; + const fetchImpl = (async (_url: URL | string, init: RequestInit) => { + receivedHeaders = init.headers; + return { ok: true, status: 200, statusText: "OK", headers: new Map(), json: async () => ({}), text: async () => "" } as Response; + }) as typeof fetch; + + await deliverWebhook( + "wh_test", + "https://example.com/hook", + "pair.registered", + { data: 123 }, + fetchImpl, + ); + + expect(receivedHeaders["X-Signature"]).toMatch(/^t=\d+,v1=[0-9a-f]{64}$/); + expect(receivedHeaders["X-Webhook-Id"]).toBe("wh_test"); + expect(receivedHeaders["X-Webhook-Event"]).toBe("pair.registered"); + }); + + it("dead-letters oversized payloads without sending", async () => { + let fetchCalled = false; + const fetchImpl = (async () => { + fetchCalled = true; + return { ok: true, status: 200 } as Response; + }) as typeof fetch; + + const largePayload = { data: "x".repeat(200000) }; + await deliverWebhook("wh_test", "https://example.com/hook", "test.event", largePayload, fetchImpl); + + expect(fetchCalled).toBe(false); + expect(listDeadLetter()).toHaveLength(1); + }); +}); + +// ─── Dead-letter queue ──────────────────────────────────────────────────────── + +describe("dead-letter queue", () => { + it("lists dead-lettered entries", async () => { + const fetchImpl = mockFetch([{ status: 500, ok: false }]); + // Override max attempts to 1 for speed + process.env.WEBHOOK_MAX_ATTEMPTS = "1"; + await deliverWebhook("wh_1", "https://example.com/hook", "test.event", {}, fetchImpl); + + const dlq = listDeadLetter(); + expect(dlq.length).toBeGreaterThan(0); + expect(dlq[0].webhookId).toBe("wh_1"); + expect(dlq[0].eventType).toBe("test.event"); + }); + + it("replays a dead-lettered entry successfully", async () => { + // First, dead-letter it + const failFetch = (async () => { throw new Error("fail"); }) as typeof fetch; + await deliverWebhook("wh_2", "https://example.com/hook", "test.event", { x: 1 }, failFetch); + + const dlq = listDeadLetter(); + expect(dlq).toHaveLength(1); + const dlqId = dlq[0].id; + + // Now replay with a working endpoint + const successFetch = mockFetch([{ status: 200, ok: true }]); + const attempts = await replayDeadLetter(dlqId, successFetch); + + expect(attempts).not.toBeNull(); + expect(attempts![attempts!.length - 1].status).toBe("success"); + + // Entry should be removed from DLQ after successful replay + expect(listDeadLetter().find((e) => e.id === dlqId)).toBeUndefined(); + }); + + it("clears the dead-letter queue", async () => { + const failFetch = (async () => { throw new Error("fail"); }) as typeof fetch; + await deliverWebhook("wh_3", "https://example.com/hook", "test.event", {}, failFetch); + + expect(listDeadLetter().length).toBeGreaterThan(0); + clearDeadLetter(); + expect(listDeadLetter()).toHaveLength(0); + }); +}); diff --git a/src/index.ts b/src/index.ts index 16017bd..ce763cf 100644 --- a/src/index.ts +++ b/src/index.ts @@ -41,6 +41,7 @@ import { type EventType, } from "./stores"; import { applySlippage, checkQuoteBounds, priceQuote, priceReverseQuote } from "./pricing"; +import { listDeadLetter, replayDeadLetter, clearDeadLetter } from "./services/webhookDelivery"; interface CacheEntry { value: { @@ -2012,6 +2013,34 @@ app.post( * * @route GET /api/v1/webhooks/:id */ +// ─── Dead-letter queue routes (#552) ──────────────────────────────────────── + +app.get("/api/v1/webhooks/dead-letter", (req: Request, res: Response) => { + const items = listDeadLetter(); + const rawLimit = parseIntegerQueryParam(req.query.limit, 100); + if (rawLimit === null) { + sendError(res, req, 400, "invalid_request", "limit must be a single integer"); + return; + } + const limit = Math.min(500, Math.max(1, rawLimit)); + res.json({ items: items.slice(0, limit), total: items.length }); +}); + +app.post("/api/v1/webhooks/dead-letter/:id/replay", async (req: Request, res: Response) => { + const id = req.params.id ?? ""; + const attempts = await replayDeadLetter(id); + if (attempts === null) { + sendError(res, req, 404, "not_found", `dead-letter entry ${id} not found`); + return; + } + res.json({ id, attempts }); +}); + +app.delete("/api/v1/webhooks/dead-letter", (_req: Request, res: Response) => { + clearDeadLetter(); + res.status(204).send(); +}); + app.get("/api/v1/webhooks/:id", (req: Request, res: Response) => { const id = req.params.id ?? ""; const record = webhookStore.get(id); diff --git a/src/services/webhookDelivery.ts b/src/services/webhookDelivery.ts new file mode 100644 index 0000000..403946c --- /dev/null +++ b/src/services/webhookDelivery.ts @@ -0,0 +1,334 @@ +import { createHmac, timingSafeEqual } from "node:crypto"; +import { logger } from "../logger"; + +// ─── Types ──────────────────────────────────────────────────────────────────── + +export interface DeliveryAttempt { + attempt: number; + status: "success" | "retry" | "dead_letter"; + statusCode: number; + timestamp: number; + error?: string; +} + +export interface DeadLetterEntry { + id: string; + webhookId: string; + webhookUrl: string; + eventType: string; + payload: Record; + attempts: DeliveryAttempt[]; + createdAt: number; + deadLetteredAt: number; + failureReason: string; +} + +// ─── Configuration ──────────────────────────────────────────────────────────── + +const WEBHOOK_SECRET = process.env.WEBHOOK_SIGNING_SECRET || "stableroute-default-webhook-secret"; + +function getMaxAttempts() { return Number(process.env.WEBHOOK_MAX_ATTEMPTS) || 5; } +function getInitialBackoffMs() { return Number(process.env.WEBHOOK_INITIAL_BACKOFF_MS) || 1000; } +function getMaxBackoffMs() { return Number(process.env.WEBHOOK_MAX_BACKOFF_MS) || 30000; } +function getRequestTimeoutMs() { return Number(process.env.WEBHOOK_TIMEOUT_MS) || 10000; } +function getMaxPayloadBytes() { return Number(process.env.WEBHOOK_MAX_PAYLOAD_BYTES) || 65536; } + +// ─── Dead-letter queue (in-memory; would use Redis/DB in production) ───────── + +const deadLetterQueue: DeadLetterEntry[] = []; + +// ─── HMAC Signing ───────────────────────────────────────────────────────────── + +/** + * Sign a webhook payload with HMAC-SHA256. + * Returns a header string: `t=,v1=`. + */ +export function signPayload( + timestamp: number, + body: string, + secret: string = WEBHOOK_SECRET, +): string { + const signedPayload = `${timestamp}.${body}`; + const signature = createHmac("sha256", secret) + .update(signedPayload) + .digest("hex"); + return `t=${timestamp},v1=${signature}`; +} + +/** + * Verify a webhook signature (for subscribers to validate). + * Returns true if the signature is valid and within the replay window. + */ +export function verifySignature( + signatureHeader: string, + body: string, + secret: string = WEBHOOK_SECRET, + toleranceSeconds = 300, +): boolean { + const parts = signatureHeader.split(","); + const tsPart = parts.find((p) => p.startsWith("t=")); + const sigPart = parts.find((p) => p.startsWith("v1=")); + if (!tsPart || !sigPart) return false; + + const timestamp = Number(tsPart.slice(2)); + if (!Number.isFinite(timestamp)) return false; + + // Replay protection + const now = Math.floor(Date.now() / 1000); + if (Math.abs(now - timestamp) > toleranceSeconds) return false; + + const expectedSig = createHmac("sha256", secret) + .update(`${timestamp}.${body}`) + .digest("hex"); + + const providedSig = sigPart.slice(3); + if (providedSig.length !== expectedSig.length) return false; + + try { + return timingSafeEqual( + Buffer.from(providedSig, "hex"), + Buffer.from(expectedSig, "hex"), + ); + } catch { + return false; + } +} + +// ─── Delivery ────────────────────────────────────────────────────────────────── + +/** + * Calculate exponential backoff delay for a given attempt number. + * Uses jitter to avoid thundering herd. + */ +export function backoffDelay(attempt: number): number { + const base = getInitialBackoffMs() * Math.pow(2, attempt - 1); + const capped = Math.min(base, getMaxBackoffMs()); + // Add up to 25% jitter + const jitter = Math.random() * capped * 0.25; + return Math.floor(capped + jitter); +} + +/** + * Deliver a webhook event to a subscriber URL with retries and dead-lettering. + * + * This function: + * 1. Signs the payload with HMAC + * 2. POSTs to the webhook URL + * 3. On 5xx/timeout, retries with exponential backoff + * 4. After MAX_ATTEMPTS, moves to the dead-letter queue + */ +export async function deliverWebhook( + webhookId: string, + webhookUrl: string, + eventType: string, + payload: Record, + fetchImpl: typeof fetch = fetch, +): Promise { + const body = JSON.stringify({ event: eventType, payload, timestamp: Date.now() }); + + // Payload size guard + if (Buffer.byteLength(body) > getMaxPayloadBytes()) { + logger.warn( + { webhookId, size: Buffer.byteLength(body) }, + "webhook payload exceeds max size, skipping delivery", + ); + const oversizeAttempts: DeliveryAttempt[] = [ + { + attempt: 1, + status: "dead_letter" as const, + statusCode: 0, + timestamp: Date.now(), + error: "payload exceeds maximum size", + }, + ]; + moveToDeadLetter( + webhookId, + webhookUrl, + eventType, + payload, + oversizeAttempts, + "payload exceeds maximum size", + ); + return oversizeAttempts; + } + + const attempts: DeliveryAttempt[] = []; + + const maxAttempts = getMaxAttempts(); + for (let attempt = 1; attempt <= maxAttempts; attempt++) { + const timestamp = Math.floor(Date.now() / 1000); + const signature = signPayload(timestamp, body); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), getRequestTimeoutMs()); + + try { + const response = await fetchImpl(webhookUrl, { + method: "POST", + headers: { + "Content-Type": "application/json", + "X-Signature": signature, + "X-Webhook-Id": webhookId, + "X-Webhook-Event": eventType, + "User-Agent": "StableRoute-Webhook/1.0", + }, + body, + signal: controller.signal, + }); + + clearTimeout(timer); + + if (response.ok) { + const result: DeliveryAttempt = { + attempt, + status: "success", + statusCode: response.status, + timestamp: Date.now(), + }; + attempts.push(result); + logger.info( + { webhookId, eventType, attempt, status: response.status }, + "webhook delivered successfully", + ); + return attempts; + } + + // 4xx — do not retry (client error, payload is correct but subscriber rejected) + if (response.status >= 400 && response.status < 500) { + const result: DeliveryAttempt = { + attempt, + status: "dead_letter", + statusCode: response.status, + timestamp: Date.now(), + error: `4xx client error: ${response.status}`, + }; + attempts.push(result); + moveToDeadLetter( + webhookId, + webhookUrl, + eventType, + payload, + attempts, + `4xx response: ${response.status}`, + ); + return attempts; + } + + // 5xx — retryable + const result: DeliveryAttempt = { + attempt, + status: attempt < maxAttempts ? "retry" : "dead_letter", + statusCode: response.status, + timestamp: Date.now(), + error: `5xx response: ${response.status}`, + }; + attempts.push(result); + + if (attempt < maxAttempts) { + await sleep(backoffDelay(attempt)); + } + } catch (err) { + clearTimeout(timer); + const errorMsg = err instanceof Error ? err.message : String(err); + const result: DeliveryAttempt = { + attempt, + status: attempt < maxAttempts ? "retry" : "dead_letter", + statusCode: 0, + timestamp: Date.now(), + error: errorMsg, + }; + attempts.push(result); + + if (attempt < maxAttempts) { + await sleep(backoffDelay(attempt)); + } + } + } + + // All attempts exhausted — move to dead-letter queue + moveToDeadLetter( + webhookId, + webhookUrl, + eventType, + payload, + attempts, + "all retry attempts exhausted", + ); + + return attempts; +} + +// ─── Dead-letter queue management ────────────────────────────────────────────── + +function moveToDeadLetter( + webhookId: string, + webhookUrl: string, + eventType: string, + payload: Record, + attempts: DeliveryAttempt[], + reason: string, +): void { + const entry: DeadLetterEntry = { + id: `dlq_${Date.now()}_${Math.random().toString(36).slice(2, 10)}`, + webhookId, + webhookUrl, + eventType, + payload, + attempts, + createdAt: attempts[0]?.timestamp ?? Date.now(), + deadLetteredAt: Date.now(), + failureReason: reason, + }; + deadLetterQueue.push(entry); + logger.warn( + { webhookId, eventType, reason, dlqId: entry.id }, + "webhook moved to dead-letter queue", + ); +} + +export function listDeadLetter(): DeadLetterEntry[] { + return [...deadLetterQueue]; +} + +export function getDeadLetterEntry(id: string): DeadLetterEntry | undefined { + return deadLetterQueue.find((e) => e.id === id); +} + +/** + * Replay a dead-lettered webhook by re-attempting delivery. + * Returns the new delivery attempts. Removes from DLQ on success. + */ +export async function replayDeadLetter( + id: string, + fetchImpl: typeof fetch = fetch, +): Promise { + const entry = deadLetterQueue.find((e) => e.id === id); + if (!entry) return null; + + const attempts = await deliverWebhook( + entry.webhookId, + entry.webhookUrl, + entry.eventType, + entry.payload, + fetchImpl, + ); + + // If the last attempt was successful, remove from DLQ + if (attempts[attempts.length - 1]?.status === "success") { + const idx = deadLetterQueue.findIndex((e) => e.id === id); + if (idx !== -1) deadLetterQueue.splice(idx, 1); + } + + return attempts; +} + +export function clearDeadLetter(): void { + deadLetterQueue.length = 0; +} + +// ─── Utilities ──────────────────────────────────────────────────────────────── + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +export { deadLetterQueue, WEBHOOK_SECRET };