Вебхуки: приём, подпись, повторы и идемпотентность

mkdir webhook-receiver
cd webhook-receiver
npm init -y
npm pkg set type=module
npm pkg set private=true --json
npm pkg delete main scripts.test
npm pkg set scripts.start="node src/server.js" scripts.test="node --test"
npm install express
mkdir src scripts test

import express from "express";
import { verifyWebhook, WebhookVerificationError } from "./signature.js";

export function createApp({ secrets, store, queue, logger = console, now = () => Date.now() }) {
  const app = express();
  app.disable("x-powered-by");

  // Raw body only on this route: the signature covers the exact bytes
  app.post(
    "/webhooks/payments",
    express.raw({ type: "application/json", limit: "256kb" }),
    async (req, res) => {
      if (!Buffer.isBuffer(req.body)) {
        return res.status(415).json({ error: "Expected application/json" });
      }
      let meta;
      try {
        meta = verifyWebhook({ secrets, headers: req.headers, rawBody: req.body, now: now() });
      } catch (error) {
        if (!(error instanceof WebhookVerificationError)) throw error;
        logger.warn(`webhook rejected: ${error.message}`);
        return res.status(400).json({ error: "Invalid webhook signature" });
      }

      let event;
      try {
        event = JSON.parse(req.body.toString("utf8"));
      } catch {
        return res.status(400).json({ error: "Body is not valid JSON" });
      }

      // Save first, then answer; a repeated delivery is not queued again
      const isNew = await store.insertIfNew({ id: meta.id, type: event.type, payload: event });
      if (isNew) queue.enqueue(meta.id);
      logger.info(`webhook ${meta.id} ${event.type}: ${isNew ? "queued" : "duplicate"}`);
      res.status(200).json({ received: true, duplicate: !isNew });
    },
  );

  // The rest of the API parses JSON as usual
  app.use(express.json());
  return app;
}

// The simplest in-process background queue: the HTTP response does not wait
// for processing. Tasks are lost if the process dies, which is why the event
// is saved to the store first. In production use BullMQ or similar.
export function createBackgroundQueue(handler, logger = console) {
  const running = new Set();
  return {
    enqueue(id) {
      const task = new Promise((resolve) => setImmediate(resolve))
        .then(() => handler(id))
        .catch((error) => logger.error(`event ${id} failed: ${error.message}`))
        .finally(() => running.delete(task));
      running.add(task);
    },
    async idle() {
      while (running.size > 0) await Promise.all(running);
    },
  };
}

import { createHmac, timingSafeEqual } from "node:crypto";

export const TOLERANCE_SECONDS = 5 * 60;

export class WebhookVerificationError extends Error {}

// Secret in the Standard Webhooks format: "whsec_" + base64 of random bytes
function keyFrom(secret) {
  return Buffer.from(secret.replace(/^whsec_/, ""), "base64");
}

// Signed content: "<id>.<timestamp>.<raw body>", HMAC-SHA256, base64
export function sign(secret, id, timestamp, rawBody) {
  return createHmac("sha256", keyFrom(secret))
    .update(`${id}.${timestamp}.`)
    .update(rawBody)
    .digest("base64");
}

export function verifyWebhook({ secrets, headers, rawBody, now = Date.now() }) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatureHeader = headers["webhook-signature"];
  if (!id || !timestamp || !signatureHeader) {
    throw new WebhookVerificationError("Missing webhook headers");
  }
  if (!/^\d{1,12}$/.test(timestamp)) {
    throw new WebhookVerificationError("Invalid timestamp");
  }
  // Replay protection: an intercepted request is useless after the tolerance window
  const age = Math.floor(now / 1000) - Number(timestamp);
  if (Math.abs(age) > TOLERANCE_SECONDS) {
    throw new WebhookVerificationError(`Timestamp outside tolerance (${age}s)`);
  }

  // The header may hold several signatures separated by spaces (secret rotation)
  const received = signatureHeader
    .split(" ")
    .filter((part) => part.startsWith("v1,"))
    .map((part) => Buffer.from(part.slice(3), "base64"));

  for (const secret of secrets) {
    const expected = Buffer.from(sign(secret, id, timestamp, rawBody), "base64");
    for (const candidate of received) {
      // timingSafeEqual throws on different lengths, so compare lengths first
      if (candidate.length === expected.length && timingSafeEqual(candidate, expected)) {
        return { id, timestamp: Number(timestamp) };
      }
    }
  }
  throw new WebhookVerificationError("No matching signature");
}

// GitHub format: X-Hub-Signature-256: sha256=<hex HMAC of the raw body>
export function verifyGithubSignature(secret, rawBody, header = "") {
  const expected = Buffer.from(
    "sha256=" + createHmac("sha256", secret).update(rawBody).digest("hex"),
  );
  const received = Buffer.from(header);
  return received.length === expected.length && timingSafeEqual(received, expected);
}

// In-memory "inbox" for the demo. In production: a table with a unique
// event id (see the SQL below); the logic stays the same.
export class EventStore {
  #events = new Map();

  // false means the id is already known: a repeated delivery
  async insertIfNew({ id, type, payload }) {
    if (this.#events.has(id)) return false;
    this.#events.set(id, {
      id,
      type,
      payload,
      status: "received",
      attempts: 0,
      receivedAt: new Date().toISOString(),
    });
    return true;
  }

  async get(id) {
    return this.#events.get(id) ?? null;
  }

  async markProcessed(id, result) {
    Object.assign(this.#events.get(id), { status: "processed", result });
  }

  async markFailed(id, error) {
    const event = this.#events.get(id);
    Object.assign(event, { status: "failed", attempts: event.attempts + 1, error: error.message });
  }

  async listUnprocessed() {
    return [...this.#events.values()].filter((event) => event.status !== "processed");
  }
}

CREATE TABLE webhook_events (
  provider     text        NOT NULL,
  id           text        NOT NULL,  -- webhook-id / event id from the provider
  type         text        NOT NULL,
  payload      jsonb       NOT NULL,
  status       text        NOT NULL DEFAULT 'received'
               CHECK (status IN ('received', 'processed', 'failed')),
  attempts     int         NOT NULL DEFAULT 0,
  received_at  timestamptz NOT NULL DEFAULT now(),
  processed_at timestamptz,
  PRIMARY KEY (provider, id)
);

CREATE TABLE orders (
  id               text PRIMARY KEY,
  status           text NOT NULL,
  provider_version int  NOT NULL DEFAULT 0
);
INSERT INTO orders (id, status) VALUES ('ord_1042', 'new');

-- Receiver: one row returned = new event, zero rows = repeated delivery
INSERT INTO webhook_events (provider, id, type, payload)
VALUES ('payments', 'msg_1', 'order.status_changed', '{"data": {"id": "ord_1042"}}')
ON CONFLICT (provider, id) DO NOTHING
RETURNING id;

-- Processor: an older version does not overwrite a newer one
UPDATE orders SET status = 'refunded', provider_version = 2
WHERE id = 'ord_1042' AND provider_version < 2;
UPDATE orders SET status = 'paid', provider_version = 1
WHERE id = 'ord_1042' AND provider_version < 1;

$ docker run -d --rm --name kb-pg-webhooks -e POSTGRES_PASSWORD=test postgres:17-alpine
$ docker exec -i kb-pg-webhooks psql -U postgres < inbox.sql
CREATE TABLE
CREATE TABLE
INSERT 0 1
  id
-------
 msg_1
(1 row)

INSERT 0 1
UPDATE 1
UPDATE 0

$ docker exec kb-pg-webhooks psql -U postgres -c "INSERT INTO webhook_events (provider, id, type, payload) VALUES ('payments', 'msg_1', 'order.status_changed', '{\"data\": {\"id\": \"ord_1042\"}}') ON CONFLICT (provider, id) DO NOTHING RETURNING id;" -c "SELECT * FROM orders;"
 id
----
(0 rows)

INSERT 0 0
    id    |  status  | provider_version
----------+----------+------------------
 ord_1042 | refunded |                2
(1 row)

export function createProcessor({ store, orders, logger = console }) {
  async function processEvent(id) {
    const event = await store.get(id);
    if (!event || event.status === "processed") return;
    try {
      if (event.type === "order.status_changed") {
        const { id: orderId, status, version } = event.payload.data;
        const applied = await orders.applyIfNewer(orderId, status, version);
        await store.markProcessed(id, applied ? "applied" : "stale");
        logger.info(`event ${id}: order ${orderId} -> ${status} v${version} (${applied ? "applied" : "stale, skipped"})`);
      } else {
        await store.markProcessed(id, "ignored"); // a type we do not handle
      }
    } catch (error) {
      await store.markFailed(id, error);
      throw error;
    }
  }

  // Run on a schedule: picks up events that failed or were lost from the queue
  async function reprocessPending() {
    for (const event of await store.listUnprocessed()) await processEvent(event.id);
  }

  return { processEvent, reprocessPending };
}

// Order state on our side. version is the object version from the provider:
// it grows with every change, so an old event never overwrites a newer state.
export class OrderStore {
  #orders = new Map();

  async applyIfNewer(id, status, version) {
    const current = this.#orders.get(id);
    if (current && current.version >= version) return false;
    this.#orders.set(id, { status, version });
    return true;
  }

  async get(id) {
    return this.#orders.get(id) ?? null;
  }
}

import { createApp } from "./app.js";
import { EventStore } from "./event-store.js";
import { OrderStore } from "./orders.js";
import { createProcessor } from "./processor.js";
import { createBackgroundQueue } from "./queue.js";

// Current secret plus, during rotation, the previous one: "whsec_new,whsec_old"
const secrets = (process.env.WEBHOOK_SECRETS ?? "").split(",").filter(Boolean);
if (secrets.length === 0) {
  console.error("Set WEBHOOK_SECRETS: the signing secret from the provider dashboard");
  process.exit(1);
}

const store = new EventStore();
const orders = new OrderStore();
const processor = createProcessor({ store, orders });
const queue = createBackgroundQueue(processor.processEvent);
const app = createApp({ secrets, store, queue });

// Safety net: once a minute, retry events that failed in the background
setInterval(() => processor.reprocessPending().catch(console.error), 60_000).unref();

const port = Number(process.env.PORT ?? 4242);
app.listen(port, "127.0.0.1", (error) => {
  if (error) {
    console.error(`Cannot start server: ${error.code}`);
    process.exit(1);
  }
  console.log(`Webhook receiver: http://127.0.0.1:${port}/webhooks/payments`);
});

import { randomUUID } from "node:crypto";
import { sign } from "./signature.js";

// Test sender: signs the body exactly the way the provider does
export async function sendWebhook({
  url,
  secret,
  event,
  id = `msg_${randomUUID()}`,
  timestamp = Math.floor(Date.now() / 1000),
  body = JSON.stringify(event),
  signature = `v1,${sign(secret, id, timestamp, body)}`,
}) {
  const res = await fetch(url, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      "webhook-id": id,
      "webhook-timestamp": String(timestamp),
      "webhook-signature": signature,
    },
    body,
    signal: AbortSignal.timeout(15_000),
  });
  return { status: res.status, body: await res.json() };
}

import { sendWebhook } from "../src/sender.js";

// node scripts/send-test-event.js <orderId> <status> <version> [eventId]
const [orderId = "ord_1042", status = "paid", version = "1", id] = process.argv.slice(2);
const result = await sendWebhook({
  url: process.env.WEBHOOK_URL ?? "http://127.0.0.1:4242/webhooks/payments",
  secret: process.env.WEBHOOK_SECRET,
  id,
  event: { type: "order.status_changed", data: { id: orderId, status, version: Number(version) } },
});
console.log(result.status, JSON.stringify(result.body));

# terminal 1
$ export WEBHOOK_SECRET="whsec_$(node -e "console.log(require('node:crypto').randomBytes(32).toString('base64'))")"
$ WEBHOOK_SECRETS="$WEBHOOK_SECRET" npm start

> webhook-receiver@1.0.0 start
> node src/server.js

Webhook receiver: http://127.0.0.1:4242/webhooks/payments
webhook msg_002 order.status_changed: queued
event msg_002: order ord_1042 -> refunded v2 (applied)
webhook msg_001 order.status_changed: queued
event msg_001: order ord_1042 -> paid v1 (stale, skipped)
webhook msg_001 order.status_changed: duplicate
webhook rejected: No matching signature

# terminal 2 (the same WEBHOOK_SECRET)
$ node scripts/send-test-event.js ord_1042 refunded 2 msg_002
200 {"received":true,"duplicate":false}
$ node scripts/send-test-event.js ord_1042 paid 1 msg_001
200 {"received":true,"duplicate":false}
$ node scripts/send-test-event.js ord_1042 paid 1 msg_001
200 {"received":true,"duplicate":true}
$ curl -s -X POST http://127.0.0.1:4242/webhooks/payments -H 'content-type: application/json' \
    -H 'webhook-id: msg_fake' -H "webhook-timestamp: $(date +%s)" \
    -H 'webhook-signature: v1,AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=' \
    -d '{"type":"order.status_changed","data":{"id":"ord_1042","status":"paid","version":99}}' -w '  HTTP %{http_code}\n'
{"error":"Invalid webhook signature"}  HTTP 400

import assert from "node:assert/strict";
import { randomBytes } from "node:crypto";
import { afterEach, beforeEach, describe, it } from "node:test";
import { createApp } from "../src/app.js";
import { EventStore } from "../src/event-store.js";
import { OrderStore } from "../src/orders.js";
import { createProcessor } from "../src/processor.js";
import { createBackgroundQueue } from "../src/queue.js";
import { sendWebhook } from "../src/sender.js";
import { sign, verifyGithubSignature } from "../src/signature.js";

const newSecret = () => `whsec_${randomBytes(32).toString("base64")}`;
const SECRET = newSecret();
const silent = { info() {}, warn() {}, error() {} };
const paid = (version = 1) => ({
  type: "order.status_changed",
  data: { id: "ord_1042", status: "paid", version },
});

let server, url, store, orders, queue, processor;

async function start({ secrets = [SECRET], slowMs = 0 } = {}) {
  store = new EventStore();
  orders = new OrderStore();
  processor = createProcessor({ store, orders, logger: silent });
  const handle = async (id) => {
    if (slowMs) await new Promise((resolve) => setTimeout(resolve, slowMs));
    await processor.processEvent(id);
  };
  queue = createBackgroundQueue(handle, silent);
  const app = createApp({ secrets, store, queue, logger: silent });
  await new Promise((resolve) => (server = app.listen(0, "127.0.0.1", resolve)));
  url = `http://127.0.0.1:${server.address().port}/webhooks/payments`;
}

beforeEach(() => start());
afterEach(() => new Promise((resolve) => server.close(resolve)));

describe("signature", () => {
  it("accepts a valid signature and processes the event in the background", async () => {
    const res = await sendWebhook({ url, secret: SECRET, event: paid() });
    assert.equal(res.status, 200);
    assert.deepEqual(res.body, { received: true, duplicate: false });
    await queue.idle();
    assert.deepEqual(await orders.get("ord_1042"), { status: "paid", version: 1 });
  });

  it("rejects a signature made with another secret", async () => {
    const res = await sendWebhook({ url, secret: newSecret(), event: paid() });
    assert.equal(res.status, 400);
    assert.equal((await store.listUnprocessed()).length, 0);
  });

  it("rejects a body changed after signing", async () => {
    const id = "msg_tampered";
    const timestamp = Math.floor(Date.now() / 1000);
    const original = await sendWebhook({ url, secret: SECRET, id: "msg_original", event: paid() });
    assert.equal(original.status, 200);
    const signature = `v1,${sign(SECRET, id, timestamp, JSON.stringify(paid()))}`;
    const body = JSON.stringify({ ...paid(), data: { id: "ord_1042", status: "refunded", version: 9 } });
    const res = await sendWebhook({ url, secret: SECRET, id, timestamp, body, signature });
    assert.equal(res.status, 400);
  });

  it("rejects an expired timestamp even with a valid signature (replay)", async () => {
    const tenMinutesAgo = Math.floor(Date.now() / 1000) - 600;
    const res = await sendWebhook({ url, secret: SECRET, event: paid(), timestamp: tenMinutesAgo });
    assert.equal(res.status, 400);
  });

  it("accepts the old secret during rotation", async () => {
    await new Promise((resolve) => server.close(resolve));
    const oldSecret = newSecret();
    await start({ secrets: [newSecret(), oldSecret] });
    const res = await sendWebhook({ url, secret: oldSecret, event: paid() });
    assert.equal(res.status, 200);
  });

  it("checks the GitHub format with the test values from the docs", () => {
    const header = "sha256=757107ea0eb2509fc211221cce984b8a37570b6d7586c22c46f4379c8b043e17";
    assert.equal(verifyGithubSignature("It's a Secret to Everybody", "Hello, World!", header), true);
    assert.equal(verifyGithubSignature("wrong secret", "Hello, World!", header), false);
    assert.equal(verifyGithubSignature("It's a Secret to Everybody", "Hello, World!", "sha256=00"), false);
  });
});

describe("delivery", () => {
  it("does not process a repeated delivery twice", async () => {
    let calls = 0;
    const applyIfNewer = orders.applyIfNewer.bind(orders);
    orders.applyIfNewer = (...args) => { calls++; return applyIfNewer(...args); };

    const first = await sendWebhook({ url, secret: SECRET, id: "msg_dup", event: paid() });
    // The sender retries: same id, new timestamp and signature
    const retry = await sendWebhook({ url, secret: SECRET, id: "msg_dup", event: paid() });
    await queue.idle();
    assert.equal(first.body.duplicate, false);
    assert.equal(retry.status, 200);
    assert.equal(retry.body.duplicate, true);
    assert.equal(calls, 1);
  });

  it("answers before the slow processing finishes", async () => {
    await new Promise((resolve) => server.close(resolve));
    await start({ slowMs: 300 });
    const started = performance.now();
    const res = await sendWebhook({ url, secret: SECRET, event: paid() });
    const elapsed = performance.now() - started;
    assert.equal(res.status, 200);
    assert.ok(elapsed < 300, `answered in ${elapsed.toFixed(0)} ms`);
    assert.equal(await orders.get("ord_1042"), null); // not processed yet
    await queue.idle();
    assert.equal((await orders.get("ord_1042")).status, "paid");
  });

  it("ignores an older event that arrives after a newer one", async () => {
    const refunded = { type: "order.status_changed", data: { id: "ord_1042", status: "refunded", version: 2 } };
    await sendWebhook({ url, secret: SECRET, id: "msg_v2", event: refunded });
    await queue.idle();
    await sendWebhook({ url, secret: SECRET, id: "msg_v1", event: paid(1) });
    await queue.idle();
    assert.deepEqual(await orders.get("ord_1042"), { status: "refunded", version: 2 });
    assert.equal((await store.get("msg_v1")).result, "stale");
  });

  it("keeps a failed event and processes it on the next pass", async () => {
    const applyIfNewer = orders.applyIfNewer.bind(orders);
    orders.applyIfNewer = async () => { throw new Error("database is down"); };
    await sendWebhook({ url, secret: SECRET, id: "msg_fail", event: paid() });
    await queue.idle();
    assert.equal((await store.get("msg_fail")).status, "failed");

    orders.applyIfNewer = applyIfNewer; // the database is back
    await processor.reprocessPending();
    assert.equal((await store.get("msg_fail")).status, "processed");
    assert.equal((await orders.get("ord_1042")).status, "paid");
  });

  it("rejects a body without the JSON content type", async () => {
    const res = await fetch(url, { method: "POST", headers: { "content-type": "text/plain" }, body: "hi" });
    assert.equal(res.status, 415);
  });
});

> webhook-receiver@1.0.0 test
> node --test

▶ signature
  ✔ accepts a valid signature and processes the event in the background (39.7763ms)
  ✔ rejects a signature made with another secret (6.3364ms)
  ✔ rejects a body changed after signing (7.3984ms)
  ✔ rejects an expired timestamp even with a valid signature (replay) (3.4191ms)
  ✔ accepts the old secret during rotation (3.7364ms)
  ✔ checks the GitHub format with the test values from the docs (1.0054ms)
✔ signature (62.3897ms)
▶ delivery
  ✔ does not process a repeated delivery twice (4.4402ms)
  ✔ answers before the slow processing finishes (309.4118ms)
  ✔ ignores an older event that arrives after a newer one (9.2903ms)
  ✔ keeps a failed event and processes it on the next pass (2.8013ms)
  ✔ rejects a body without the JSON content type (3.1035ms)
✔ delivery (329.5465ms)
ℹ tests 11
ℹ suites 2
ℹ pass 11
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 550.4084