import { createHmac, randomBytes, timingSafeEqual } from "crypto" import { and, asc, eq, lte } from "drizzle-orm" import { db } from "@/lib/db" import { webhook_deliveries, webhook_endpoints } from "@/lib/db/schema" import { isSafeWebhookUrl } from "./ssrf" // ============================================================================ // Webhook delivery: signing + HTTP POST + retry bookkeeping. // // Signature scheme (Stripe-style, HMAC-SHA256): // header X-PMN-Signature: t=,v1= // signed `${t}.${rawBody}` // Receivers recompute the HMAC with their endpoint secret and compare. A short // timestamp lets them reject replays. // ============================================================================ const SECRET_PREFIX = "whsec_" const DELIVERY_TIMEOUT_MS = 10_000 const MAX_RESPONSE_CHARS = 2_000 // Retry backoff (minutes) indexed by the attempt number just completed. const BACKOFF_MINUTES = [1, 5, 15, 60, 180] type EndpointRow = typeof webhook_endpoints.$inferSelect type DeliveryRow = typeof webhook_deliveries.$inferSelect /** Generate a new endpoint signing secret, e.g. `whsec_<64 hex>`. */ export function generateWebhookSecret(): string { return `${SECRET_PREFIX}${randomBytes(32).toString("hex")}` } /** Compute the `X-PMN-Signature` header value for a raw body + secret. */ export function signaturePayload(secret: string, timestamp: number, body: string): string { return createHmac("sha256", secret).update(`${timestamp}.${body}`).digest("hex") } function signatureHeader(secret: string, body: string): { header: string; timestamp: number } { const timestamp = Math.floor(Date.now() / 1000) const v1 = signaturePayload(secret, timestamp, body) return { header: `t=${timestamp},v1=${v1}`, timestamp } } /** * Verify an inbound signature header against a body + secret. Exposed so that a * receiver built on this codebase (and our own tests) can validate deliveries. * Tolerates clock skew up to `toleranceSeconds` (default 5 min). */ export function verifySignature( secret: string, header: string, body: string, toleranceSeconds = 300 ): boolean { const parts = Object.fromEntries( header.split(",").map((kv) => { const [k, v] = kv.split("=") return [k?.trim(), v?.trim()] }) ) const t = Number(parts.t) const given = parts.v1 if (!Number.isFinite(t) || !given) return false if (Math.abs(Math.floor(Date.now() / 1000) - t) > toleranceSeconds) return false const expected = signaturePayload(secret, t, body) const a = Buffer.from(expected) const b = Buffer.from(given) return a.length === b.length && timingSafeEqual(a, b) } function backoffIso(attemptsCompleted: number): string { const minutes = BACKOFF_MINUTES[attemptsCompleted - 1] ?? BACKOFF_MINUTES[BACKOFF_MINUTES.length - 1] return new Date(Date.now() + minutes * 60_000).toISOString() } /** * Attempt to deliver a single delivery row to its endpoint, then persist the * outcome (status, attempt count, next_attempt_at) and update the endpoint's * health fields. Never throws — returns whether the POST succeeded. */ export async function attemptDelivery(delivery: DeliveryRow, endpoint: EndpointRow): Promise { const body = JSON.stringify(delivery.payload) const nowIso = new Date().toISOString() const attemptsNow = delivery.attempts + 1 let ok = false let responseStatus: number | null = null let responseBody: string | null = null let error: string | null = null try { // Re-check for SSRF at delivery time — DNS may have changed since creation. if (!(await isSafeWebhookUrl(endpoint.url))) { throw new Error("Destination blocked by SSRF protection") } const { header } = signatureHeader(endpoint.secret, body) const controller = new AbortController() const timer = setTimeout(() => controller.abort(), DELIVERY_TIMEOUT_MS) try { const res = await fetch(endpoint.url, { method: "POST", headers: { "content-type": "application/json", "user-agent": "PMN-Webhooks/1.0", "x-pmn-event": delivery.event, "x-pmn-delivery": delivery.id, "x-pmn-webhook-id": endpoint.id, "x-pmn-signature": header, }, body, signal: controller.signal, redirect: "manual", }) responseStatus = res.status responseBody = (await res.text().catch(() => "")).slice(0, MAX_RESPONSE_CHARS) ok = res.status >= 200 && res.status < 300 if (!ok) error = `Endpoint returned HTTP ${res.status}` } finally { clearTimeout(timer) } } catch (e) { error = e instanceof Error ? (e.name === "AbortError" ? "Request timed out" : e.message) : "Delivery failed" } const exhausted = !ok && attemptsNow >= delivery.max_attempts await db .update(webhook_deliveries) .set({ status: ok ? "success" : exhausted ? "failed" : "pending", attempts: attemptsNow, response_status: responseStatus, response_body: responseBody, error: ok ? null : error, delivered_at: ok ? nowIso : delivery.delivered_at, next_attempt_at: ok || exhausted ? delivery.next_attempt_at : backoffIso(attemptsNow), updated_at: nowIso, }) .where(eq(webhook_deliveries.id, delivery.id)) await db .update(webhook_endpoints) .set( ok ? { last_success_at: nowIso, failure_count: 0, last_error: null } : { last_error_at: nowIso, last_error: error, failure_count: endpoint.failure_count + 1 } ) .where(eq(webhook_endpoints.id, endpoint.id)) return ok } /** * Cron drain: deliver every pending delivery whose retry time has come, for all * accounts. Returns counts for the cron response. Endpoints that are disabled * are left untouched (their pending deliveries resume if re-enabled). */ export async function processDueDeliveries(limit = 100): Promise<{ processed: number; delivered: number }> { const nowIso = new Date().toISOString() const due = await db.query.webhook_deliveries.findMany({ where: and( eq(webhook_deliveries.status, "pending"), lte(webhook_deliveries.next_attempt_at, nowIso) ), with: { endpoint: true }, orderBy: asc(webhook_deliveries.next_attempt_at), limit, }) let delivered = 0 let processed = 0 for (const row of due) { const { endpoint, ...delivery } = row if (!endpoint || endpoint.status !== "active") continue processed++ if (await attemptDelivery(delivery as DeliveryRow, endpoint)) delivered++ } return { processed, delivered } }