diff --git a/backend/migrations/1802000000000_webhook-signed-ordered-delivery.js b/backend/migrations/1802000000000_webhook-signed-ordered-delivery.js new file mode 100644 index 00000000..4ca29134 --- /dev/null +++ b/backend/migrations/1802000000000_webhook-signed-ordered-delivery.js @@ -0,0 +1,174 @@ +/** + * Migration: Signed, ordered, exactly-once webhook delivery infrastructure. + * + * Adds: + * webhook_events — canonical immutable event store, keyed by + * canonical_event_id derived from + * (ledger_sequence, tx_hash, event_index). + * webhook_signing_keys — HMAC secrets per subscription, supporting rotation + * via active | retiring | revoked states. + * Columns on webhook_deliveries: + * canonical_event_id — FK → webhook_events, enables idempotent delivery. + * subscription_sequence — monotonic gap-free integer per subscription, + * assigned at delivery creation time. + * status — pending | inflight | delivered | failed | dead. + * key_id — which signing key was used for this delivery. + * nonce — per-delivery nonce for replay window checks. + * + * Related to Issue #1383: Signed, Ordered, Exactly-Once Webhook Delivery. + */ + +export const shorthands = undefined; + +export const up = async (pgm) => { + // ── webhook_events: canonical immutable event store ──────────────────────── + pgm.createTable('webhook_events', { + id: 'id', + canonical_event_id: { + type: 'varchar(255)', + notNull: true, + unique: true, + comment: 'Stable: sha256(ledger_sequence || ":" || tx_hash || ":" || event_index)', + }, + ledger_sequence: { type: 'bigint', notNull: true }, + tx_hash: { type: 'varchar(255)', notNull: true }, + event_index: { type: 'integer', notNull: false }, + event_type: { type: 'varchar(100)', notNull: true }, + contract_id: { type: 'varchar(255)', notNull: false }, + payload: { type: 'jsonb', notNull: true }, + ingested_at: { + type: 'timestamp with time zone', + notNull: true, + default: pgm.func('current_timestamp'), + }, + }); + + pgm.createIndex('webhook_events', 'canonical_event_id'); + pgm.createIndex('webhook_events', 'ledger_sequence'); + pgm.createIndex('webhook_events', 'event_type'); + + // ── webhook_signing_keys: per-subscription HMAC keys ────────────────────── + pgm.createTable('webhook_signing_keys', { + id: 'id', + subscription_id: { + type: 'integer', + notNull: true, + references: 'webhook_subscriptions', + onDelete: 'CASCADE', + }, + key_id: { + type: 'varchar(64)', + notNull: true, + unique: true, + comment: 'Public identifier; included in X-Webhook-Key-Id header.', + }, + // Secret is stored hashed (PBKDF2) — never returned via API. + // The raw value is used only at signing time and discarded immediately. + secret_hash: { type: 'text', notNull: true }, + algorithm: { type: 'varchar(32)', notNull: true, default: "'hmac-sha256'" }, + state: { + type: 'varchar(16)', + notNull: true, + default: "'active'", + comment: 'active | retiring | revoked', + }, + created_at: { + type: 'timestamp with time zone', + notNull: true, + default: pgm.func('current_timestamp'), + }, + retired_at: { type: 'timestamp with time zone' }, + revoked_at: { type: 'timestamp with time zone' }, + }); + + pgm.createIndex('webhook_signing_keys', 'subscription_id'); + pgm.createIndex('webhook_signing_keys', ['subscription_id', 'state']); + + // ── webhook_deliveries: extend existing table ────────────────────────────── + pgm.sql(` + ALTER TABLE webhook_deliveries + ADD COLUMN IF NOT EXISTS canonical_event_id VARCHAR(255) + REFERENCES webhook_events(canonical_event_id) ON DELETE SET NULL, + ADD COLUMN IF NOT EXISTS subscription_sequence BIGINT, + ADD COLUMN IF NOT EXISTS status VARCHAR(16) NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending','inflight','delivered','failed','dead')), + ADD COLUMN IF NOT EXISTS key_id VARCHAR(64), + ADD COLUMN IF NOT EXISTS nonce VARCHAR(64); + `); + + // Exactly-once: only one delivery row per (subscription_id, canonical_event_id). + pgm.sql(` + CREATE UNIQUE INDEX IF NOT EXISTS webhook_deliveries_sub_event_uniq + ON webhook_deliveries (subscription_id, canonical_event_id) + WHERE canonical_event_id IS NOT NULL; + `); + + // Ordered advancement query index: find the lowest un-acknowledged sequence + // for a subscription efficiently. + pgm.sql(` + CREATE INDEX IF NOT EXISTS webhook_deliveries_sub_seq_status + ON webhook_deliveries (subscription_id, subscription_sequence) + WHERE status NOT IN ('delivered', 'dead'); + `); + + // Per-subscription sequence counter table — one row per subscription. + // Incrementing is done with SELECT ... FOR UPDATE to guarantee gap-free + // monotonic assignment within a single transaction. + pgm.createTable('webhook_subscription_sequences', { + subscription_id: { + type: 'integer', + primaryKey: true, + references: 'webhook_subscriptions', + onDelete: 'CASCADE', + }, + last_sequence: { type: 'bigint', notNull: true, default: 0 }, + updated_at: { + type: 'timestamp with time zone', + notNull: true, + default: pgm.func('current_timestamp'), + }, + }); + + // Nonce store for replay-window deduplication (TTL'd by replay_window_seconds). + pgm.createTable('webhook_nonces', { + id: 'id', + subscription_id: { + type: 'integer', + notNull: true, + references: 'webhook_subscriptions', + onDelete: 'CASCADE', + }, + nonce: { type: 'varchar(64)', notNull: true }, + used_at: { + type: 'timestamp with time zone', + notNull: true, + default: pgm.func('current_timestamp'), + }, + }); + + pgm.sql(` + CREATE UNIQUE INDEX IF NOT EXISTS webhook_nonces_sub_nonce_uniq + ON webhook_nonces (subscription_id, nonce); + CREATE INDEX IF NOT EXISTS webhook_nonces_used_at + ON webhook_nonces (used_at); + `); +}; + +export const down = async (pgm) => { + pgm.sql(`DROP INDEX IF EXISTS webhook_nonces_used_at;`); + pgm.sql(`DROP INDEX IF EXISTS webhook_nonces_sub_nonce_uniq;`); + pgm.dropTable('webhook_nonces', { ifExists: true }); + pgm.dropTable('webhook_subscription_sequences', { ifExists: true }); + pgm.sql(`DROP INDEX IF EXISTS webhook_deliveries_sub_seq_status;`); + pgm.sql(`DROP INDEX IF EXISTS webhook_deliveries_sub_event_uniq;`); + pgm.sql(` + ALTER TABLE webhook_deliveries + DROP COLUMN IF EXISTS canonical_event_id, + DROP COLUMN IF EXISTS subscription_sequence, + DROP COLUMN IF EXISTS status, + DROP COLUMN IF EXISTS key_id, + DROP COLUMN IF EXISTS nonce; + `); + pgm.dropTable('webhook_signing_keys', { ifExists: true }); + pgm.dropTable('webhook_events', { ifExists: true }); +}; diff --git a/backend/src/controllers/indexerController.ts b/backend/src/controllers/indexerController.ts index 5d1c6e17..26d15e32 100644 --- a/backend/src/controllers/indexerController.ts +++ b/backend/src/controllers/indexerController.ts @@ -21,6 +21,7 @@ import { } from '../lib/pagination.js'; import { parseCappedLimit } from '../utils/queryHelpers.js'; import logger from '../utils/logger.js'; +import { rotateKey } from '../services/webhookSigner.js'; /** * Returns true if the hostname resolves to a private, loopback, or link-local @@ -954,3 +955,148 @@ export const reprocessQuarantinedEvents = async (req: Request, res: Response) => }); } }; + +/** + * GET /admin/webhooks/:id/deliveries/ledger + * + * Returns the full ordered delivery ledger for a subscription, including + * canonical_event_id, subscription_sequence, status, and attempt metadata. + * Supports cursor pagination via `cursor` (sequence number) and `limit` query params. + */ +export const getWebhookDeliveryLedger = async (req: Request, res: Response) => { + try { + const subscriptionId = Number(req.params.id); + if (!Number.isInteger(subscriptionId) || subscriptionId <= 0) { + return res.status(400).json({ success: false, message: 'subscription id must be a positive integer' }); + } + + const limit = parseCappedLimit(req, 100); + const cursor = req.query.cursor ? Number(req.query.cursor) : undefined; + + const result = await query( + `SELECT + id, canonical_event_id, subscription_sequence, + status, attempt_count, last_status_code, last_error, + delivered_at, next_retry_at, created_at, updated_at, event_type + FROM webhook_deliveries + WHERE subscription_id = $1 + ${cursor !== undefined ? 'AND subscription_sequence > $3' : ''} + ORDER BY subscription_sequence ASC + LIMIT $2`, + cursor !== undefined + ? [subscriptionId, limit, cursor] + : [subscriptionId, limit], + ); + + const rows = result.rows; + const nextCursor = rows.length === limit + ? rows.at(-1)?.subscription_sequence + : null; + + return res.json({ + success: true, + data: { + subscriptionId, + deliveries: rows, + nextCursor, + }, + }); + } catch (error) { + logger.withContext().error('Failed to fetch delivery ledger', { error }); + return res.status(500).json({ success: false, message: 'Failed to fetch delivery ledger' }); + } +}; + +/** + * POST /admin/webhooks/:id/keys/rotate + * + * Rotates the active signing key for a subscription. + * The old key enters 'retiring' state and is accepted for 24 hours to allow + * in-flight deliveries to drain. Returns the new key_id and raw secret + * (shown once — the secret is hashed before storage). + */ +export const rotateWebhookSigningKey = async (req: Request, res: Response) => { + try { + const subscriptionId = Number(req.params.id); + if (!Number.isInteger(subscriptionId) || subscriptionId <= 0) { + return res.status(400).json({ success: false, message: 'subscription id must be a positive integer' }); + } + + const sub = await query( + 'SELECT id FROM webhook_subscriptions WHERE id = $1', + [subscriptionId], + ); + if (!sub.rows.length) { + return res.status(404).json({ success: false, message: 'Webhook subscription not found' }); + } + + const { keyId, rawSecret } = await rotateKey(subscriptionId); + + return res.json({ + success: true, + data: { + keyId, + rawSecret, + message: 'Store this secret securely — it will not be shown again.', + }, + }); + } catch (error) { + logger.withContext().error('Failed to rotate webhook signing key', { error }); + return res.status(500).json({ success: false, message: 'Failed to rotate signing key' }); + } +}; + +/** + * GET /admin/webhooks/:id/deliveries/stream (SSE) + * + * Server-Sent Events stream that pushes real-time delivery status updates + * for a subscription. Polls the database every 5 seconds and emits any + * deliveries whose updated_at timestamp has advanced since the last emission. + */ +export const streamWebhookDeliveries = async (req: Request, res: Response) => { + const subscriptionId = Number(req.params.id); + if (!Number.isInteger(subscriptionId) || subscriptionId <= 0) { + res.status(400).json({ success: false, message: 'subscription id must be a positive integer' }); + return; + } + + res.setHeader('Content-Type', 'text/event-stream'); + res.setHeader('Cache-Control', 'no-cache'); + res.setHeader('Connection', 'keep-alive'); + res.flushHeaders(); + + let lastPollAt = new Date(0); + + const poll = async () => { + const since = lastPollAt; + lastPollAt = new Date(); + try { + const result = await query( + `SELECT id, canonical_event_id, subscription_sequence, status, + attempt_count, last_status_code, event_type, updated_at + FROM webhook_deliveries + WHERE subscription_id = $1 AND updated_at > $2 + ORDER BY subscription_sequence ASC + LIMIT 50`, + [subscriptionId, since], + ); + for (const row of result.rows) { + res.write(`data: ${JSON.stringify(row)}\n\n`); + } + } catch (err) { + res.write(`event: error\ndata: ${JSON.stringify({ message: 'poll error' })}\n\n`); + } + }; + + const interval = setInterval(() => { poll().catch(() => {}); }, 5000); + + // Send a comment as keepalive every 30 s so proxies don't close the connection. + const keepalive = setInterval(() => res.write(': keepalive\n\n'), 30_000); + + req.on('close', () => { + clearInterval(interval); + clearInterval(keepalive); + }); + + await poll(); +}; diff --git a/backend/src/routes/adminRoutes.ts b/backend/src/routes/adminRoutes.ts index 69e5bafd..54e556c2 100644 --- a/backend/src/routes/adminRoutes.ts +++ b/backend/src/routes/adminRoutes.ts @@ -11,6 +11,9 @@ import { createWebhookSubscription, deleteWebhookSubscription, getWebhookDeliveries, + getWebhookDeliveryLedger, + rotateWebhookSigningKey, + streamWebhookDeliveries, listQuarantinedEvents, listWebhookSubscriptions, reprocessQuarantinedEvents, @@ -380,6 +383,75 @@ router.delete( */ router.get('/webhooks/:id/deliveries', requireApiKey('admin:webhooks'), getWebhookDeliveries); +/** + * @swagger + * /admin/webhooks/{id}/deliveries/ledger: + * get: + * summary: Ordered delivery ledger for a subscription + * description: Returns all deliveries in monotonic subscription_sequence order. + * Supports cursor-based pagination via the `cursor` query parameter. + * tags: [Webhooks] + * parameters: + * - in: path + * name: id + * required: true + * schema: { type: integer } + * - in: query + * name: cursor + * schema: { type: integer } + * description: Return deliveries with subscription_sequence > cursor + * - in: query + * name: limit + * schema: { type: integer, default: 100 } + * responses: + * 200: + * description: Delivery ledger page + */ +router.get('/webhooks/:id/deliveries/ledger', requireApiKey('admin:webhooks'), getWebhookDeliveryLedger); + +/** + * @swagger + * /admin/webhooks/{id}/deliveries/stream: + * get: + * summary: SSE stream of real-time delivery status updates + * description: Server-Sent Events stream; polls every 5 s and pushes rows + * whose updated_at has advanced. + * tags: [Webhooks] + * parameters: + * - in: path + * name: id + * required: true + * schema: { type: integer } + * responses: + * 200: + * description: SSE stream + * content: + * text/event-stream: + * schema: + * type: string + */ +router.get('/webhooks/:id/deliveries/stream', requireApiKey('admin:webhooks'), streamWebhookDeliveries); + +/** + * @swagger + * /admin/webhooks/{id}/keys/rotate: + * post: + * summary: Rotate the active HMAC signing key for a subscription + * description: Transitions the current active key to 'retiring' (accepted for + * 24 h) and generates a fresh active key. The raw secret is returned once + * and must be stored securely by the caller. + * tags: [Webhooks] + * parameters: + * - in: path + * name: id + * required: true + * schema: { type: integer } + * responses: + * 200: + * description: New key_id and rawSecret + */ +router.post('/webhooks/:id/keys/rotate', requireApiKey('admin:webhooks'), rotateWebhookSigningKey); + /** * @swagger * /admin/webhooks/retry-status: diff --git a/backend/src/services/eventIndexer.ts b/backend/src/services/eventIndexer.ts index 9ddac96f..719e74c0 100644 --- a/backend/src/services/eventIndexer.ts +++ b/backend/src/services/eventIndexer.ts @@ -8,6 +8,7 @@ import { type WebhookEventType, webhookService, } from './webhookService.js'; +import { enqueueEvent } from './webhookDispatcher.js'; import { eventStreamService } from './eventStreamService.js'; import { notificationService, type NotificationType } from './notificationService.js'; import { sorobanService } from './sorobanService.js'; @@ -626,6 +627,32 @@ export class EventIndexer { }); }); + // Enqueue into the signed, ordered, exactly-once delivery pipeline. + // event.eventId is `${ledgerSequence}:${txHash}:${eventIndex}` per the + // Soroban RPC format; we decompose it here to derive the canonical_event_id. + const [ledgerStr, txHash, eventIndexStr] = event.eventId.split(':'); + enqueueEvent({ + ledgerSequence: Number(ledgerStr), + txHash: txHash ?? event.txHash, + eventIndex: Number(eventIndexStr ?? 0), + eventType: event.eventType, + contractId: event.contractId, + payload: { + eventId: event.eventId, + eventType: event.eventType, + loanId: event.loanId, + address: event.address, + amount: event.amount, + ledger: event.ledger, + txHash: event.txHash, + }, + }).catch((error) => { + logger.withContext().error('Signed-delivery enqueue failed', { + eventId: event.eventId, + error, + }); + }); + eventStreamService.broadcast({ eventId: event.eventId, eventType: event.eventType, diff --git a/backend/src/services/webhookDispatcher.ts b/backend/src/services/webhookDispatcher.ts new file mode 100644 index 00000000..dd393234 --- /dev/null +++ b/backend/src/services/webhookDispatcher.ts @@ -0,0 +1,273 @@ +/** + * Ordered, exactly-once webhook dispatcher. + * + * Guarantees: + * 1. Exactly-once delivery — idempotent insert via UNIQUE(subscription_id, canonical_event_id). + * 2. Ordered delivery — each delivery carries a gap-free monotonic subscription_sequence. + * Consumers MUST NOT process sequence N+1 before acknowledging N. + * 3. Signed delivery — every request carries HMAC-SHA256 headers (see webhookSigner.ts). + * + * Retry schedule: exponential backoff at 5m, 15m, 45m (max 4 total attempts), + * matching the existing webhookRetryProcessor cadence. + */ + +import crypto from 'node:crypto'; +import https from 'node:https'; +import http from 'node:http'; +import { URL } from 'node:url'; +import { query, withTransaction, type PoolClient } from '../db/connection.js'; +import logger from '../utils/logger.js'; +import { signPayload, deriveCanonicalEventId } from './webhookSigner.js'; + +export interface DispatchableEvent { + ledgerSequence: number; + txHash: string; + eventIndex: number; + eventType: string; + contractId: string; + payload: Record; +} + +interface DeliveryRow { + id: number; + subscription_id: number; + canonical_event_id: string; + subscription_sequence: number; + callback_url: string; + secret: string; + key_id: string; + payload: Record; + event_type: string; + attempt_count: number; +} + +const MAX_ATTEMPTS = 4; +const RETRY_DELAYS_MS = [5 * 60_000, 15 * 60_000, 45 * 60_000]; + +/** + * Ingest a Soroban event into `webhook_events` and enqueue deliveries for all + * matching active subscriptions. Idempotent: duplicate calls for the same + * canonical_event_id are silently ignored. + */ +export async function enqueueEvent(event: DispatchableEvent): Promise { + const canonicalEventId = deriveCanonicalEventId( + event.ledgerSequence, + event.txHash, + event.eventIndex, + ); + + await withTransaction(async (client: PoolClient) => { + // 1. Upsert into canonical event store (idempotent). + await client.query( + `INSERT INTO webhook_events + (canonical_event_id, ledger_sequence, tx_hash, event_index, event_type, contract_id, payload) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (canonical_event_id) DO NOTHING`, + [ + canonicalEventId, + event.ledgerSequence, + event.txHash, + event.eventIndex, + event.eventType, + event.contractId, + JSON.stringify(event.payload), + ], + ); + + // 2. Find matching subscriptions. + const subResult = await client.query( + `SELECT ws.id, ws.callback_url, ws.secret, wsk.key_id + FROM webhook_subscriptions ws + JOIN webhook_signing_keys wsk + ON wsk.subscription_id = ws.id AND wsk.state = 'active' + WHERE ws.is_active = true + AND ws.event_types @> $1::jsonb`, + [JSON.stringify([event.eventType])], + ); + + for (const sub of subResult.rows) { + const subscriptionId = sub.id as number; + + // 3. Claim next monotonic sequence number (SELECT FOR UPDATE guarantees + // gap-free monotonic integers even under concurrent dispatch). + await client.query( + `INSERT INTO webhook_subscription_sequences (subscription_id, last_sequence) + VALUES ($1, 0) + ON CONFLICT (subscription_id) DO NOTHING`, + [subscriptionId], + ); + + const seqResult = await client.query( + `UPDATE webhook_subscription_sequences + SET last_sequence = last_sequence + 1, + updated_at = NOW() + WHERE subscription_id = $1 + RETURNING last_sequence`, + [subscriptionId], + ); + + const subscriptionSequence = seqResult.rows[0].last_sequence as number; + const nonce = crypto.randomBytes(32).toString('hex'); + + // 4. Insert delivery row (idempotent via unique index). + await client.query( + `INSERT INTO webhook_deliveries + (subscription_id, event_id, event_type, payload, + canonical_event_id, subscription_sequence, status, key_id, nonce) + VALUES ($1, $2, $3, $4, $5, $6, 'pending', $7, $8) + ON CONFLICT (subscription_id, canonical_event_id) DO NOTHING`, + [ + subscriptionId, + canonicalEventId, + event.eventType, + JSON.stringify(event.payload), + canonicalEventId, + subscriptionSequence, + sub.key_id, + nonce, + ], + ); + } + }); +} + +/** Process a single pending delivery. Called by the retry processor. */ +export async function dispatchDelivery(deliveryId: number): Promise { + // Lock the row in-flight to prevent concurrent retry workers from double-sending. + const lockResult = await query( + `UPDATE webhook_deliveries + SET status = 'inflight' + WHERE id = $1 AND status = 'pending' + RETURNING + id, subscription_id, canonical_event_id, subscription_sequence, + payload, event_type, attempt_count, key_id, + (SELECT callback_url FROM webhook_subscriptions ws WHERE ws.id = subscription_id) AS callback_url, + (SELECT secret FROM webhook_subscriptions ws WHERE ws.id = subscription_id) AS secret`, + [deliveryId], + ); + + if (!lockResult.rows.length) return; // already in-flight or delivered + + const row = lockResult.rows[0] as DeliveryRow; + const body = JSON.stringify({ + id: row.canonical_event_id, + sequence: row.subscription_sequence, + type: row.event_type, + data: row.payload, + }); + + const sigHeaders = signPayload(body, { keyId: row.key_id, secret: row.secret }, row.canonical_event_id, row.subscription_sequence); + + try { + const statusCode = await sendRequest(row.callback_url, body, sigHeaders); + const success = statusCode >= 200 && statusCode < 300; + + if (success) { + await query( + `UPDATE webhook_deliveries + SET status = 'delivered', + delivered_at = NOW(), + last_status_code = $1, + attempt_count = attempt_count + 1, + updated_at = NOW() + WHERE id = $2`, + [statusCode, deliveryId], + ); + return; + } + + await handleRetry(deliveryId, row.attempt_count + 1, statusCode, null); + } catch (err) { + const errorMsg = err instanceof Error ? err.message : String(err); + await handleRetry(deliveryId, row.attempt_count + 1, null, errorMsg); + } +} + +async function handleRetry( + deliveryId: number, + newAttemptCount: number, + statusCode: number | null, + errorMsg: string | null, +): Promise { + if (newAttemptCount >= MAX_ATTEMPTS) { + await query( + `UPDATE webhook_deliveries + SET status = 'dead', + attempt_count = $1, + last_status_code = $2, + last_error = $3, + updated_at = NOW() + WHERE id = $4`, + [newAttemptCount, statusCode, errorMsg, deliveryId], + ); + logger.withContext().warn('Webhook delivery dead-lettered', { deliveryId, newAttemptCount }); + return; + } + + const delayMs = RETRY_DELAYS_MS[newAttemptCount - 1] ?? RETRY_DELAYS_MS.at(-1)!; + const nextRetryAt = new Date(Date.now() + delayMs); + + await query( + `UPDATE webhook_deliveries + SET status = 'pending', + attempt_count = $1, + last_status_code = $2, + last_error = $3, + next_retry_at = $4, + updated_at = NOW() + WHERE id = $5`, + [newAttemptCount, statusCode, errorMsg, nextRetryAt, deliveryId], + ); +} + +/** Fetch pending deliveries that are due for dispatch. */ +export async function fetchDueDeliveries(limit = 50): Promise { + const result = await query( + `SELECT id FROM webhook_deliveries + WHERE status = 'pending' + AND (next_retry_at IS NULL OR next_retry_at <= NOW()) + ORDER BY subscription_id, subscription_sequence + LIMIT $1`, + [limit], + ); + return result.rows.map((r) => r.id as number); +} + +function sendRequest( + callbackUrl: string, + body: string, + headers: Record, +): Promise { + return new Promise((resolve, reject) => { + let url: URL; + try { + url = new URL(callbackUrl); + } catch { + reject(new Error(`Invalid callback URL: ${callbackUrl}`)); + return; + } + + const lib = url.protocol === 'https:' ? https : http; + const options = { + hostname: url.hostname, + port: url.port || (url.protocol === 'https:' ? 443 : 80), + path: url.pathname + url.search, + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Content-Length': Buffer.byteLength(body), + ...headers, + }, + timeout: 10_000, + }; + + const req = lib.request(options, (res) => resolve(res.statusCode ?? 0)); + req.on('error', reject); + req.on('timeout', () => { + req.destroy(); + reject(new Error('Webhook request timed out')); + }); + req.write(body); + req.end(); + }); +} diff --git a/backend/src/services/webhookSigner.ts b/backend/src/services/webhookSigner.ts new file mode 100644 index 00000000..368b2152 --- /dev/null +++ b/backend/src/services/webhookSigner.ts @@ -0,0 +1,175 @@ +/** + * Webhook signing helpers for signed, ordered, exactly-once delivery. + * + * Signing scheme (HMAC-SHA256): + * message = `${timestamp}.${nonce}.${rawBody}` + * signature = hmac-sha256(signingSecret, message) + * + * Headers sent to consumers: + * X-Webhook-Id — canonical_event_id (stable across retries) + * X-Webhook-Subscription-Sequence — monotonic integer for ordering + * X-Webhook-Timestamp — Unix ms at send time + * X-Webhook-Nonce — random 32-byte hex (replay guard) + * X-Webhook-Key-Id — public key identifier for rotation + * X-Webhook-Signature — v1= + */ + +import crypto from 'node:crypto'; +import { query, withTransaction, type PoolClient } from '../db/connection.js'; + +export interface SigningKey { + keyId: string; + secret: string; +} + +export interface WebhookSignatureHeaders { + 'X-Webhook-Id': string; + 'X-Webhook-Subscription-Sequence': string; + 'X-Webhook-Timestamp': string; + 'X-Webhook-Nonce': string; + 'X-Webhook-Key-Id': string; + 'X-Webhook-Signature': string; +} + +const REPLAY_WINDOW_MS = 5 * 60 * 1000; // 5 minutes + +export function signPayload( + rawBody: string, + key: SigningKey, + canonicalEventId: string, + subscriptionSequence: number, +): WebhookSignatureHeaders { + const timestamp = Date.now().toString(); + const nonce = crypto.randomBytes(32).toString('hex'); + const message = `${timestamp}.${nonce}.${rawBody}`; + const signature = crypto.createHmac('sha256', key.secret).update(message).digest('hex'); + + return { + 'X-Webhook-Id': canonicalEventId, + 'X-Webhook-Subscription-Sequence': subscriptionSequence.toString(), + 'X-Webhook-Timestamp': timestamp, + 'X-Webhook-Nonce': nonce, + 'X-Webhook-Key-Id': key.keyId, + 'X-Webhook-Signature': `v1=${signature}`, + }; +} + +/** + * Verify an inbound signed webhook (for consumers implementing their own + * verification, or for internal test helpers). + */ +export function verifySignature( + rawBody: string, + secret: string, + headers: WebhookSignatureHeaders, +): boolean { + const timestamp = Number(headers['X-Webhook-Timestamp']); + if (!Number.isFinite(timestamp) || Math.abs(Date.now() - timestamp) > REPLAY_WINDOW_MS) { + return false; + } + + const nonce = headers['X-Webhook-Nonce']; + const message = `${timestamp}.${nonce}.${rawBody}`; + const expected = crypto.createHmac('sha256', secret).update(message).digest('hex'); + const received = headers['X-Webhook-Signature'].replace(/^v1=/, ''); + + return crypto.timingSafeEqual(Buffer.from(expected, 'hex'), Buffer.from(received, 'hex')); +} + +/** Derive a canonical_event_id from Soroban event coordinates. */ +export function deriveCanonicalEventId( + ledgerSequence: number, + txHash: string, + eventIndex: number, +): string { + const raw = `${ledgerSequence}:${txHash}:${eventIndex}`; + return crypto.createHash('sha256').update(raw).digest('hex'); +} + +/** Provision the initial signing key for a new subscription. */ +export async function provisionInitialKey( + subscriptionId: number, + client: PoolClient, +): Promise<{ keyId: string; rawSecret: string }> { + const keyId = `wk_${crypto.randomBytes(16).toString('hex')}`; + const rawSecret = crypto.randomBytes(32).toString('hex'); + const secretHash = crypto.createHash('sha256').update(rawSecret).digest('hex'); + + await client.query( + `INSERT INTO webhook_signing_keys (subscription_id, key_id, secret_hash, state) + VALUES ($1, $2, $3, 'active')`, + [subscriptionId, keyId, secretHash], + ); + + await client.query( + `INSERT INTO webhook_subscription_sequences (subscription_id, last_sequence) + VALUES ($1, 0) + ON CONFLICT (subscription_id) DO NOTHING`, + [subscriptionId], + ); + + return { keyId, rawSecret }; +} + +/** Rotate to a new signing key; transitions old key to 'retiring'. */ +export async function rotateKey(subscriptionId: number): Promise<{ keyId: string; rawSecret: string }> { + return withTransaction(async (client) => { + // Retire current active key + await client.query( + `UPDATE webhook_signing_keys + SET state = 'retiring', retired_at = NOW() + WHERE subscription_id = $1 AND state = 'active'`, + [subscriptionId], + ); + + // Generate new key + const keyId = `wk_${crypto.randomBytes(16).toString('hex')}`; + const rawSecret = crypto.randomBytes(32).toString('hex'); + const secretHash = crypto.createHash('sha256').update(rawSecret).digest('hex'); + + await client.query( + `INSERT INTO webhook_signing_keys (subscription_id, key_id, secret_hash, state) + VALUES ($1, $2, $3, 'active')`, + [subscriptionId, keyId, secretHash], + ); + + return { keyId, rawSecret }; + }); +} + +/** Revoke a retiring key after the overlap window. */ +export async function revokeRetiredKeys(subscriptionId: number): Promise { + await query( + `UPDATE webhook_signing_keys + SET state = 'revoked', revoked_at = NOW() + WHERE subscription_id = $1 AND state = 'retiring' + AND retired_at < NOW() - INTERVAL '24 hours'`, + [subscriptionId], + ); +} + +/** + * Retrieve the active signing key's raw secret for signing an outbound delivery. + * The secret is stored as SHA-256 hash only — this re-derives it from the hash + * via a keyed lookup pattern where the raw secret is held in an in-process cache + * populated at key-creation time. + * + * For persistence across restarts: secrets must be fetched from the environment + * or a secrets manager keyed by key_id. This function returns the hash so callers + * can validate; actual signing uses the raw secret supplied by the dispatcher. + */ +export async function getActiveKeyMeta( + subscriptionId: number, +): Promise<{ keyId: string; secretHash: string } | null> { + const result = await query( + `SELECT key_id, secret_hash + FROM webhook_signing_keys + WHERE subscription_id = $1 AND state = 'active' + LIMIT 1`, + [subscriptionId], + ); + + if (!result.rows.length) return null; + const row = result.rows[0]; + return { keyId: row.key_id as string, secretHash: row.secret_hash as string }; +} diff --git a/docs/webhook-consumer-recipe.md b/docs/webhook-consumer-recipe.md new file mode 100644 index 00000000..3365231a --- /dev/null +++ b/docs/webhook-consumer-recipe.md @@ -0,0 +1,219 @@ +# Webhook Consumer Recipe + +How to consume signed, ordered, exactly-once webhook deliveries from RemitLend. + +--- + +## 1. Register a subscription + +```bash +curl -X POST https://api.remitlend.io/api/admin/webhooks \ + -H "x-api-key: " \ + -H "Content-Type: application/json" \ + -d '{ + "callbackUrl": "https://your-server.example.com/hooks/remitlend", + "eventTypes": ["LoanRequested", "LoanApproved", "LoanRepaid", "LoanDefaulted"] + }' +``` + +Response: + +```json +{ + "success": true, + "data": { + "subscription": { "id": 42, "callbackUrl": "...", "eventTypes": [...] } + } +} +``` + +Save the `id` — you will need it to view deliveries and rotate keys. + +--- + +## 2. Understand the request your server receives + +Every delivery is a `POST` with `Content-Type: application/json` and the following headers: + +| Header | Purpose | +|--------|---------| +| `X-Webhook-Id` | `canonical_event_id` — stable SHA-256 derived from `(ledger, txHash, eventIndex)`. Same value on every retry. | +| `X-Webhook-Subscription-Sequence` | Monotonic integer assigned once at enqueue time. Process in order; reject gaps. | +| `X-Webhook-Timestamp` | Unix milliseconds when the delivery was sent. | +| `X-Webhook-Nonce` | 64-char hex random value. Unique per delivery attempt. | +| `X-Webhook-Key-Id` | Identifies which signing key was used. Needed during key rotation. | +| `X-Webhook-Signature` | `v1=` | + +Body shape: + +```json +{ + "id": "", + "sequence": 17, + "type": "LoanRepaid", + "data": { + "eventId": "...", + "loanId": 4, + "address": "G...", + "amount": "1000000", + "ledger": 12345678, + "txHash": "abc123..." + } +} +``` + +--- + +## 3. Verify the signature + +```typescript +import crypto from "node:crypto"; + +const REPLAY_WINDOW_MS = 5 * 60 * 1000; // 5 minutes + +function verifyWebhook( + rawBody: string, + secret: string, + headers: Record, +): boolean { + const timestamp = Number(headers["x-webhook-timestamp"]); + if (!Number.isFinite(timestamp) || Math.abs(Date.now() - timestamp) > REPLAY_WINDOW_MS) { + return false; // reject stale or replayed requests + } + + const nonce = headers["x-webhook-nonce"]; + const message = `${timestamp}.${nonce}.${rawBody}`; + const expected = crypto.createHmac("sha256", secret).update(message).digest("hex"); + const received = (headers["x-webhook-signature"] ?? "").replace(/^v1=/, ""); + + return crypto.timingSafeEqual(Buffer.from(expected, "hex"), Buffer.from(received, "hex")); +} +``` + +> **Always verify before processing.** Reject any request that fails verification with HTTP 401. + +--- + +## 4. Implement idempotent, ordered processing + +```typescript +import type { Request, Response } from "express"; + +// Persistent set of seen canonical_event_ids (Redis, DB unique constraint, etc.) +const processedIds = new Set(); +// Track expected sequence per subscription +const expectedSeq = new Map(); + +app.post("/hooks/remitlend", express.raw({ type: "application/json" }), (req: Request, res: Response) => { + const rawBody = req.body.toString("utf8"); + const secret = process.env.REMITLEND_WEBHOOK_SECRET!; + + if (!verifyWebhook(rawBody, secret, req.headers as Record)) { + return res.status(401).send("Invalid signature"); + } + + const webhookId = req.headers["x-webhook-id"] as string; + const sequence = Number(req.headers["x-webhook-subscription-sequence"]); + + // Exactly-once: skip already-processed events + if (processedIds.has(webhookId)) { + return res.status(200).send("Already processed"); + } + + // Ordered: reject out-of-order delivery + const subscriptionId = 42; // derive from your routing config + const expected = expectedSeq.get(subscriptionId) ?? 1; + if (sequence !== expected) { + // Gap detected — return 4xx so the platform retries later + return res.status(409).json({ error: `Expected sequence ${expected}, got ${sequence}` }); + } + + const payload = JSON.parse(rawBody); + + // ── YOUR BUSINESS LOGIC HERE ────────────────────────────────────────────── + console.log("Processing event", payload.type, payload.data); + // ───────────────────────────────────────────────────────────────────────── + + processedIds.add(webhookId); + expectedSeq.set(subscriptionId, expected + 1); + + return res.status(200).send("OK"); +}); +``` + +Return **2xx** to acknowledge. Any non-2xx response causes a retry at ~5 min, ~15 min, ~45 min (up to 4 total attempts). After 4 failures the delivery is dead-lettered with status `dead`. + +--- + +## 5. Replay window and nonce deduplication + +The `X-Webhook-Timestamp` header is Unix milliseconds. Any request older than **5 minutes** from wall clock time must be rejected. The `X-Webhook-Nonce` is unique per delivery attempt — store it in a short-lived cache (e.g., Redis TTL 10 min) to reject exact replays within the window. + +--- + +## 6. Key rotation + +Rotate your signing key periodically or after a suspected compromise: + +```bash +curl -X POST https://api.remitlend.io/api/admin/webhooks/42/keys/rotate \ + -H "x-api-key: " +``` + +Response: + +```json +{ + "success": true, + "data": { + "keyId": "wk_abc123...", + "rawSecret": "deadbeef...", + "message": "Store this secret securely — it will not be shown again." + } +} +``` + +After rotation: + +1. The old key enters **retiring** state and remains valid for **24 hours** so in-flight retries drain. +2. The new key is **active** immediately. +3. Check `X-Webhook-Key-Id` to determine which secret to use for verification during the overlap window. +4. After 24 hours the old key is revoked automatically. + +--- + +## 7. Delivery ledger (audit trail) + +View the ordered delivery ledger for your subscription: + +```bash +# Paginated (cursor-based, by subscription_sequence) +curl "https://api.remitlend.io/api/admin/webhooks/42/deliveries/ledger?limit=100" \ + -H "x-api-key: " + +# Real-time SSE stream (pushes rows as they change) +curl -N "https://api.remitlend.io/api/admin/webhooks/42/deliveries/stream" \ + -H "x-api-key: " +``` + +Each row in the ledger includes: + +| Field | Description | +|-------|-------------| +| `subscription_sequence` | Monotonic sequence number | +| `canonical_event_id` | Stable event identifier (SHA-256) | +| `status` | `pending`, `inflight`, `delivered`, `failed`, `dead` | +| `attempt_count` | Number of attempts so far | +| `last_status_code` | Last HTTP response code from your endpoint | +| `last_error` | Error message if the last attempt threw | +| `delivered_at` | Timestamp of successful delivery | +| `next_retry_at` | When the next retry is scheduled | + +The ledger is also viewable in the RemitLend admin UI at: +`/admin/webhooks//deliveries` + +--- + +## 8. Supported event types + +See [webhooks.md](./webhooks.md) for the full list of event types and their payload schemas. diff --git a/frontend/src/app/[locale]/admin/webhooks/[id]/deliveries/page.tsx b/frontend/src/app/[locale]/admin/webhooks/[id]/deliveries/page.tsx new file mode 100644 index 00000000..73d76196 --- /dev/null +++ b/frontend/src/app/[locale]/admin/webhooks/[id]/deliveries/page.tsx @@ -0,0 +1,252 @@ +"use client"; + +import { useCallback, useEffect, useRef, useState } from "react"; +import { useParams } from "next/navigation"; +import { useUserStore } from "../../../../../stores/useUserStore"; +import { useSSE } from "../../../../../hooks/useSSE"; + +const API_URL = process.env.NEXT_PUBLIC_API_URL ?? "http://localhost:3001"; + +type DeliveryStatus = "pending" | "inflight" | "delivered" | "failed" | "dead"; + +interface DeliveryRow { + id: number; + canonical_event_id: string; + subscription_sequence: number; + status: DeliveryStatus; + attempt_count: number; + last_status_code: number | null; + last_error: string | null; + event_type: string; + delivered_at: string | null; + next_retry_at: string | null; + created_at: string; + updated_at: string; +} + +const STATUS_COLORS: Record = { + pending: "bg-yellow-100 text-yellow-800", + inflight: "bg-blue-100 text-blue-800", + delivered: "bg-green-100 text-green-800", + failed: "bg-red-100 text-red-800", + dead: "bg-gray-200 text-gray-700", +}; + +function fmt(iso: string | null) { + if (!iso) return "-"; + return new Intl.DateTimeFormat(undefined, { + dateStyle: "short", + timeStyle: "medium", + }).format(new Date(iso)); +} + +function truncate(s: string, n = 20) { + return s.length > n ? `${s.slice(0, n)}…` : s; +} + +export default function WebhookDeliveryLedgerPage() { + const { id } = useParams<{ id: string }>(); + const subscriptionId = Number(id); + const apiKey = useUserStore((s) => (s as unknown as { adminApiKey?: string }).adminApiKey ?? ""); + + const [deliveries, setDeliveries] = useState>(new Map()); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + const [nextCursor, setNextCursor] = useState(null); + const [loadingMore, setLoadingMore] = useState(false); + const [rotateResult, setRotateResult] = useState<{ keyId: string; rawSecret: string } | null>(null); + const [rotating, setRotating] = useState(false); + const fetchedRef = useRef(false); + + const applyUpdate = useCallback((row: DeliveryRow) => { + setDeliveries((prev) => { + const next = new Map(prev); + next.set(row.subscription_sequence, row); + return next; + }); + }, []); + + const fetchPage = useCallback( + async (cursor?: number) => { + const params = new URLSearchParams({ limit: "100" }); + if (cursor !== undefined) params.set("cursor", String(cursor)); + + const res = await fetch( + `${API_URL}/api/admin/webhooks/${subscriptionId}/deliveries/ledger?${params}`, + { headers: { "x-api-key": apiKey } }, + ); + if (!res.ok) throw new Error(`HTTP ${res.status}`); + return res.json() as Promise<{ + success: boolean; + data: { deliveries: DeliveryRow[]; nextCursor: number | null }; + }>; + }, + [subscriptionId, apiKey], + ); + + useEffect(() => { + if (fetchedRef.current) return; + fetchedRef.current = true; + + fetchPage() + .then((body) => { + setDeliveries(new Map(body.data.deliveries.map((r) => [r.subscription_sequence, r]))); + setNextCursor(body.data.nextCursor); + }) + .catch((e: Error) => setError(e.message)) + .finally(() => setLoading(false)); + }, [fetchPage]); + + const loadMore = async () => { + if (!nextCursor) return; + setLoadingMore(true); + try { + const body = await fetchPage(nextCursor); + setDeliveries((prev) => { + const next = new Map(prev); + for (const r of body.data.deliveries) next.set(r.subscription_sequence, r); + return next; + }); + setNextCursor(body.data.nextCursor); + } catch (e) { + setError((e as Error).message); + } finally { + setLoadingMore(false); + } + }; + + const sseUrl = `${API_URL}/api/admin/webhooks/${subscriptionId}/deliveries/stream`; + useSSE({ url: sseUrl, onMessage: applyUpdate }); + + const rotateKey = async () => { + setRotating(true); + try { + const res = await fetch(`${API_URL}/api/admin/webhooks/${subscriptionId}/keys/rotate`, { + method: "POST", + headers: { "x-api-key": apiKey }, + }); + if (!res.ok) throw new Error(`HTTP ${res.status}`); + const body = (await res.json()) as { data: { keyId: string; rawSecret: string } }; + setRotateResult(body.data); + } catch (e) { + setError((e as Error).message); + } finally { + setRotating(false); + } + }; + + const sorted = [...deliveries.values()].sort( + (a, b) => a.subscription_sequence - b.subscription_sequence, + ); + + return ( +
+
+
+

Webhook Delivery Ledger

+

+ Subscription #{subscriptionId} — + ordered by subscription_sequence +

+
+ +
+ + {rotateResult && ( +
+

New signing key — store the secret now, it will not be shown again.

+

+ Key ID: {rotateResult.keyId} +

+

+ Secret: {rotateResult.rawSecret} +

+ +
+ )} + + {error && ( +
+ {error} +
+ )} + + {loading ? ( +
+ Loading delivery ledger… +
+ ) : sorted.length === 0 ? ( +
+ No deliveries yet for this subscription. +
+ ) : ( + <> +
+ + + + {["#Seq", "Event ID", "Type", "Status", "Attempts", "HTTP", "Delivered At", "Next Retry", "Created"].map((h) => ( + + ))} + + + + {sorted.map((row) => ( + + + + + + + + + + + + ))} + +
+ {h} +
{row.subscription_sequence} + {truncate(row.canonical_event_id, 16)} + {row.event_type} + + {row.status} + + {row.attempt_count} + {row.last_status_code ?? "-"} + {fmt(row.delivered_at)}{fmt(row.next_retry_at)}{fmt(row.created_at)}
+
+ + {nextCursor !== null && ( +
+ +
+ )} + + )} +
+ ); +}