fix: reject replayed webhook deliveries

.github/workflows/ci.yml

@@ -39,3 +39,4 @@ jobs:
           npm init -y >/dev/null && npm pkg set type=module >/dev/null
           npm install "$TARBALL"
           node -e 'import("@corbits/webhooks").then((m) => { for (const n of ["createHookRoutes", "createRunTriggerDeliverer", "createTenantSystemSender", "isRunTriggerUnroutable"]) if (typeof m[n] !== "function") throw new Error("missing export: " + n); })'
+          node -e 'import("@corbits/webhooks/migrations").then((m) => { if (typeof m.runWebhookMigrations !== "function") throw new Error("missing export: runWebhookMigrations"); })'

README.md

@@ -4,7 +4,7 @@ Signed webhook ingress for Interchange workflows: verifies a Slack, Standard Web
 
 ## Why @corbits/webhooks?
 
-1. **Secrets stay in the hub.** Each hook is an ordinary Interchange credential with `metadata.webhook`; this package stores nothing of its own.
+1. **Secrets stay in the hub.** Each hook is an ordinary Interchange credential with `metadata.webhook`. The only thing this package stores is a short-lived table of seen delivery ids.
 2. **Runs never borrow a person's authority.** The workflow runs as its own run principal (the identity Interchange derives for each run) through the hub's mail-triggered grant path, never as the credential owner.
 3. **Three verifiers, no unsigned mode.** Slack signatures, Standard Webhooks HMAC and bearer tokens. A credential without a verifier matches no hook.
 
@@ -14,10 +14,10 @@ Use it to start workflows from third-party events. It does not fire on a schedul
 
 ```bash
 bun add @corbits/webhooks \
-  @intx/crypto @intx/db @intx/hub-api @intx/hub-common @intx/mime @intx/types hono
+  @intx/crypto @intx/db @intx/hub-api @intx/hub-common @intx/mime @intx/types drizzle-orm hono postgres
 ```
 
-The `@intx/*` peers are `^0.4.0`; `hono` is `^4.11.9`.
+The `@intx/*` peers are `^0.4.0`, `drizzle-orm` is `^0.45.1`, `hono` is `^4.11.9` and `postgres` is `^3.4.8`.
 
 ## Where it fits
 
@@ -49,6 +49,8 @@ With neither set, the credential name is matched against definition and asset na
 
 With `standard-webhooks`, the secret is base64-decoded after stripping an optional `whsec_` prefix, as the spec requires. An unprefixed secret also verifies when the sender used it as raw bytes, which is how 0.1 read it. `bearer` rejects an empty secret.
 
+A replayed delivery gets `409`. Standard Webhooks deliveries are keyed on the credential and `webhook-id`, Slack on the credential and signature, each kept until its timestamp leaves the ±300s window. The seen-set is the `replay` table in the `webhooks` schema, created by `runWebhookMigrations` (see [Using with Interchange](#using-with-interchange)), so every replica sharing the database rejects the replay. A failed delivery releases its key so the sender's retry goes through. Bearer requests carry no id and are not deduplicated.
+
 Signatures are checked over the raw request bytes, and the body is forwarded unchanged. Bodies over 1 MiB get `413`; a verified body that is not UTF-8 gets `415`. Bot tokens for chat integrations belong in the workflow's `credentialBindings`, not in the signing secret.
 
 ### Exports
@@ -64,23 +66,32 @@ Signatures are checked over the raw request bytes, and the body is forwarded unc
 | `HookMailRouter`                                      | Router type the host passes in: `routeMail` and `sendRunGrants`.           |
 | `MailDeliverer`                                       | `{ to(address, content, tenantId, subject) }`, what the deliverer returns. |
 
+### `runWebhookMigrations(dbConfig, { schema })`
+
+From `@corbits/webhooks/migrations`. Run it after Interchange's `runMigrations`, with the same config and the same `schema`: the host schema that holds the `credential` table. The `replay` table always lives in the `webhooks` schema. It is idempotent and takes an advisory lock, so several replicas can start at once.
+
 ## Using with Interchange
 
-Build the routes from the hub's database, credential cipher and principal key store:
+Run `runWebhookMigrations` at hub start, after `runMigrations`, then build the routes from the hub's database, credential cipher and principal key store:
 
 ```ts
 import { createEnvKeyCredentialCipher } from "@intx/crypto";
-import { createDB, createPrincipalKeyStore } from "@intx/db";
+import { createDB, createPrincipalKeyStore, runMigrations } from "@intx/db";
 import { hexDecode } from "@intx/types";
 import { createHookRoutes, type HookMailRouter } from "@corbits/webhooks";
+import { runWebhookMigrations } from "@corbits/webhooks/migrations";
 
-const { db } = createDB({
+const dbConfig = {
   host: "localhost",
   port: 5432,
   user: "postgres",
   password: "postgres",
   database: "interchange",
-});
+};
+await runMigrations(dbConfig, { schema: "public" });
+await runWebhookMigrations(dbConfig, { schema: "public" });
+
+const { db } = createDB(dbConfig);
 
 export const hookRoutes = (router: HookMailRouter) =>
   createHookRoutes({
@@ -136,6 +147,7 @@ The hub answers `202 { "ok": true, "to": "run_…@acme.example" }` and the run s
 
 - `installWebhooks(opts)` is removed. Call `app.route("/api/hooks", createHookRoutes(deps))` with the same `db`, `credentialCipher`, `principalKeyStore` and `router`; drop `app` from the options.
 - `@intx/*` and `hono` moved to peer dependencies. Add them to the host.
+- Run `runWebhookMigrations` at hub start, after `runMigrations`; replay protection needs its table.
 - Existing hook credentials and URLs keep working unchanged when the routes stay mounted at `/api/hooks`.
 
 ## License

e2e/helpers.ts

@@ -11,6 +11,7 @@ import { createEnvKeyCredentialCipher } from "@intx/crypto";
 import { principal, tenant } from "@intx/db/schema";
 
 import { createHookRoutes, type HookMailRouter } from "../src/index.js";
+import { runWebhookMigrations } from "../src/migrations.js";
 
 const PG_ENV = [
   "PGHOST",
@@ -31,7 +32,7 @@ function requireEnv(key: (typeof PG_ENV)[number]): string {
   return value;
 }
 
-function harnessDbConfig(): DBConfig {
+export function harnessDbConfig(): DBConfig {
   return {
     host: requireEnv("PGHOST"),
     port: Number(requireEnv("PGPORT")),
@@ -57,7 +58,9 @@ export async function createTestDb(): Promise<TestDb> {
   const schema = `wh_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 8)}`;
   try {
     await runMigrations(config, { schema });
+    await runWebhookMigrations(config, { schema });
   } catch (error) {
+    await dropSchema(config, { schema: "webhooks" });
     await dropSchema(config, { schema });
     throw error;
   }
@@ -82,6 +85,7 @@ export async function createTestDb(): Promise<TestDb> {
     },
     close: async () => {
       await handle.close();
+      await dropSchema(config, { schema: "webhooks" });
       await dropSchema(config, { schema });
     },
   };

e2e/hooks.test.ts

@@ -106,8 +106,8 @@ async function seedDeployment(db: TestDb["db"]): Promise<void> {
 
 async function standardWebhooksHeaders(
   body: string,
+  id = `msg_${crypto.randomUUID()}`,
 ): Promise<Record<string, string>> {
-  const id = "msg_1";
   const timestamp = String(Math.floor(Date.now() / 1000));
   const key = await crypto.subtle.importKey(
     "raw",
@@ -204,6 +204,29 @@ describe.skipIf(!harnessDbAvailable())(
       expect(recorded.mail).toHaveLength(1);
     });
 
+    test("a replayed delivery is 409 on every replica", async () => {
+      const recorded = recordingRouter();
+      const replicaA = mountHookRoutes({ db: db(), router: recorded.router });
+      const replicaB = mountHookRoutes({ db: db(), router: recorded.router });
+      const body = `{"text":"hi"}`;
+      const headers = await standardWebhooksHeaders(body);
+
+      const first = await replicaA.request(`/api/hooks/${HOOK}`, {
+        method: "POST",
+        headers,
+        body,
+      });
+      const replay = await replicaB.request(`/api/hooks/${HOOK}`, {
+        method: "POST",
+        headers,
+        body,
+      });
+
+      expect(first.status).toBe(202);
+      expect(replay.status).toBe(409);
+      expect(recorded.mail).toHaveLength(1);
+    });
+
     test("the tenant-scoped name path resolves only inside that tenant", async () => {
       const recorded = recordingRouter();
       const app = mountHookRoutes({ db: db(), router: recorded.router });

e2e/migrations.test.ts

@@ -0,0 +1,60 @@
+import { afterAll, beforeAll, describe, expect, test } from "bun:test";
+import { randomUUID } from "node:crypto";
+import { runMigrations } from "@intx/db";
+import postgres from "postgres";
+
+import { runWebhookMigrations } from "../src/migrations.js";
+import { harnessDbAvailable, harnessDbConfig } from "./helpers.js";
+
+const describeIfDb = harnessDbAvailable() ? describe : describe.skip;
+
+describeIfDb("runWebhookMigrations", () => {
+  const admin = harnessDbAvailable()
+    ? postgres({ ...harnessDbConfig(), max: 1, onnotice: () => undefined })
+    : undefined;
+  const target = harnessDbAvailable()
+    ? {
+        ...harnessDbConfig(),
+        database: `webhooks_migrations_${randomUUID().slice(0, 8)}`,
+      }
+    : undefined;
+
+  beforeAll(async () => {
+    if (admin === undefined || target === undefined) return;
+    await admin.unsafe(`CREATE DATABASE "${target.database}"`);
+    await runMigrations(target, { schema: "public" });
+  }, 120_000);
+
+  afterAll(async () => {
+    if (admin === undefined || target === undefined) return;
+    await admin.unsafe(
+      `DROP DATABASE IF EXISTS "${target.database}" WITH (FORCE)`,
+    );
+    await admin.end({ timeout: 5 });
+  });
+
+  test("three concurrent runners converge on one replay table", async () => {
+    if (target === undefined) return;
+    await Promise.all([
+      runWebhookMigrations(target, { schema: "public" }),
+      runWebhookMigrations(target, { schema: "public" }),
+      runWebhookMigrations(target, { schema: "public" }),
+    ]);
+
+    const client = postgres({ ...target, max: 1 });
+    try {
+      const [fk] = await client`
+        SELECT confrelid::regclass::text AS target
+        FROM pg_constraint
+        WHERE conrelid = 'webhooks.replay'::regclass AND contype = 'f'`;
+      expect(fk?.["target"]).toBe("credential");
+      const [index] = await client`
+        SELECT indexname FROM pg_indexes
+        WHERE schemaname = 'webhooks' AND tablename = 'replay'
+          AND indexname = 'webhooks_replay_expires_at_idx'`;
+      expect(index).toBeDefined();
+    } finally {
+      await client.end({ timeout: 5 });
+    }
+  });
+});

migrations/0001_webhooks.sql

@@ -0,0 +1,10 @@
+CREATE SCHEMA IF NOT EXISTS "webhooks";
+--> statement-breakpoint
+CREATE TABLE IF NOT EXISTS "webhooks"."replay" (
+  "credential_id" text NOT NULL REFERENCES "public"."credential"("id") ON DELETE CASCADE,
+  "nonce" text NOT NULL,
+  "expires_at" timestamptz NOT NULL,
+  PRIMARY KEY ("credential_id", "nonce")
+);
+--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "webhooks_replay_expires_at_idx" ON "webhooks"."replay" ("expires_at");

package.json

@@ -22,6 +22,7 @@
   },
   "files": [
     "dist",
+    "migrations",
     "README.md",
     "LICENSE"
   ],
@@ -31,6 +32,10 @@
     ".": {
       "types": "./dist/index.d.ts",
       "default": "./dist/index.js"
+    },
+    "./migrations": {
+      "types": "./dist/migrations.d.ts",
+      "default": "./dist/migrations.js"
     }
   },
   "publishConfig": {
@@ -59,6 +64,7 @@
     "hono": "4.11.9",
     "oxfmt": "0.70.0",
     "oxlint": "1.85.0",
+    "postgres": "3.4.9",
     "typescript": "5.9.3"
   },
   "peerDependencies": {
@@ -68,7 +74,9 @@
     "@intx/hub-common": "^0.4.0",
     "@intx/mime": "^0.4.0",
     "@intx/types": "^0.4.0",
-    "hono": "^4.11.9"
+    "drizzle-orm": "^0.45.1",
+    "hono": "^4.11.9",
+    "postgres": "^3.4.8"
   },
   "engines": {
     "bun": ">=1.2.0",

src/hooks.test.ts

@@ -22,6 +22,41 @@ function hook(over: Partial<LoadedHook> = {}): LoadedHook {
   };
 }
 
+function memoryReplay() {
+  const seen = new Set<string>();
+  return {
+    claim: async (credentialId: string, nonce: string) =>
+      !seen.has(`${credentialId}:${nonce}`) &&
+      !!seen.add(`${credentialId}:${nonce}`),
+    release: async (credentialId: string, nonce: string) =>
+      void seen.delete(`${credentialId}:${nonce}`),
+  };
+}
+
+async function slackRequest(secret: string, body: string) {
+  const timestamp = String(Math.floor(Date.now() / 1000));
+  const key = await crypto.subtle.importKey(
+    "raw",
+    new TextEncoder().encode(secret),
+    { name: "HMAC", hash: "SHA-256" },
+    false,
+    ["sign"],
+  );
+  const mac = await crypto.subtle.sign(
+    "HMAC",
+    key,
+    new TextEncoder().encode(`v0:${timestamp}:${body}`),
+  );
+  return {
+    method: "POST",
+    body,
+    headers: {
+      "x-slack-request-timestamp": timestamp,
+      "x-slack-signature": `v0=${Buffer.from(mac).toString("hex")}`,
+    },
+  };
+}
+
 function mount(opts?: {
   loaded?: LoadedHook | "ambiguous" | undefined;
   runs?: LiveRun[];
@@ -33,6 +68,7 @@ function mount(opts?: {
     "/api/hooks",
     createHookApp({
       loadHook: async () => opts?.loaded,
+      replay: memoryReplay(),
       listRuns: async () => opts?.runs ?? [],
       deliver: {
         to: async (to, content, tenantId) => {
@@ -252,6 +288,40 @@ describe("createHookApp", () => {
     expect(delivered).toEqual([]);
   });
 
+  test("409s a replayed Slack delivery", async () => {
+    const { app, delivered } = mount({
+      loaded: hook({
+        secret: "signing-secret",
+        meta: { verify: "slack", to: JIMMY.address },
+      }),
+      runs: [JIMMY],
+    });
+    const init = await slackRequest("signing-secret", "{}");
+    const first = await app.request("/api/hooks/slack", init);
+    const again = await app.request("/api/hooks/slack", init);
+    expect(first.status).toBe(202);
+    expect(again.status).toBe(409);
+    expect(delivered).toHaveLength(1);
+  });
+
+  test("accepts a sender retry after a failed delivery", async () => {
+    let fail = true;
+    const { app } = mount({
+      loaded: hook({
+        secret: "signing-secret",
+        meta: { verify: "slack", to: JIMMY.address },
+      }),
+      runs: [JIMMY],
+      deliver: async () => {
+        if (fail) throw new Error("agent is unreachable");
+      },
+    });
+    const init = await slackRequest("signing-secret", "{}");
+    expect((await app.request("/api/hooks/slack", init)).status).toBe(503);
+    fail = false;
+    expect((await app.request("/api/hooks/slack", init)).status).toBe(202);
+  });
+
   test("404 when the hook is ambiguous", async () => {
     const { app } = mount({ loaded: "ambiguous" });
     const res = await app.request("/api/hooks/slack", {