tests/integration/webhooks.test.ts

Rama main · réplica de lectura del repositorio original.

import { afterAll, beforeAll, test } from "bun:test";
import assert from "node:assert/strict";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createSqliteDb, openDatabase } from "../../src/runtime/bun/sqlite";
import { serve } from "../../src/runtime/bun/server";
import { provisionWith, TEST_SECRET_CIPHER } from "../contract/harness";
import type { WebhookSecurity } from "../../src/core/webhook-security";

/**
 * Signed webhook delivery against a real HTTP receiver over a real socket.
 *
 * The contract suite proves a delivery row is queued. This proves the request
 * actually leaves the process, arrives with a signature the receiver can verify
 * independently, and that a rejecting receiver is recorded as a failure rather
 * than silently dropped.
 */
const directory = mkdtempSync(join(tmpdir(), "titen-hook-"));
const dbPath = join(directory, "titen.db");
const SECRET = "webhook-shared-secret-value";

interface Received {
  signature: string | null;
  signatureV2: string | null;
  timestamp: string | null;
  event: string | null;
  delivery: string | null;
  body: string;
}

let titen: Awaited<ReturnType<typeof serve>>;
let receiver: { stop: () => void; port: number };
let received: Received[] = [];
let rejectNext = false;
let key: string;
let transportMode: "deliver" | "unreachable" | "timeout" | "redirect" = "deliver";
let resolvedAddresses = ["93.184.216.34"];
const WEBHOOK_HOST = "hooks.example.com";

/** Independent verification: recompute the HMAC the way an integrator would. */
async function hmacHex(secret: string, payload: string): Promise<string> {
  const imported = await crypto.subtle.importKey(
    "raw",
    new TextEncoder().encode(secret),
    { name: "HMAC", hash: "SHA-256" },
    false,
    ["sign"],
  );
  const mac = await crypto.subtle.sign("HMAC", imported, new TextEncoder().encode(payload));
  return [...new Uint8Array(mac)].map((b) => b.toString(16).padStart(2, "0")).join("");
}

async function call(method: string, path: string, body?: unknown) {
  const res = await fetch(`${titen.url}${path}`, {
    method,
    headers: { "content-type": "application/json", authorization: `Bearer ${key}` },
    body: body === undefined ? undefined : JSON.stringify(body),
  });
  return { status: res.status, body: (await res.json()) as any };
}

beforeAll(async () => {
  const webhookSecurity: WebhookSecurity = {
    allowedHostnames: [WEBHOOK_HOST],
    timeoutMs: 30,
    resolve: async () => resolvedAddresses,
    async dispatch({ url, addresses, init }) {
      if (transportMode === "unreachable") throw new Error("unreachable");
      if (transportMode === "redirect")
        return {
          response: new Response(null, { status: 302, headers: { location: "http://169.254.169.254/" } }),
          connectedAddress: addresses[0]!,
        };
      if (transportMode === "timeout")
        await new Promise((_, reject) => {
          init.signal?.addEventListener("abort", () => reject(init.signal?.reason), { once: true });
        });
      const destination = new URL(url);
      return {
        response: await fetch(`http://127.0.0.1:${receiver.port}${destination.pathname}`, init),
        connectedAddress: addresses[0]!,
      };
    },
  };
  titen = await serve({
    dbPath,
    port: 0,
    hostname: "127.0.0.1",
    quiet: true,
    revision: "hook",
    maintenanceIntervalMs: 0,
    secretCipher: TEST_SECRET_CIPHER,
    webhookSecurity,
  });
  const provisioned = await provisionWith(createSqliteDb(openDatabase(dbPath)), { scopes: ["*"] });
  key = provisioned.key;

  // @ts-ignore - Bun global is provided by the runtime.
  const server = Bun.serve({
    port: 0,
    hostname: "127.0.0.1",
    async fetch(request: Request) {
      received.push({
        signature: request.headers.get("x-titen-signature"),
        signatureV2: request.headers.get("x-titen-signature-v2"),
        timestamp: request.headers.get("x-titen-timestamp"),
        event: request.headers.get("x-titen-event"),
        delivery: request.headers.get("x-titen-delivery"),
        body: await request.text(),
      });
      return new Response(rejectNext ? "no" : "ok", { status: rejectNext ? 500 : 200 });
    },
  });
  assert.ok(server.port, "the HTTP receiver must have a TCP port");
  receiver = { stop: () => server.stop(true), port: server.port };
});

afterAll(async () => {
  receiver.stop();
  await titen.stop();
  rmSync(directory, { recursive: true, force: true });
});

test("a delivered event arrives signed, and the signature verifies independently", async () => {
  received = [];
  rejectNext = false;
  transportMode = "deliver";
  resolvedAddresses = ["93.184.216.34"];

  const hook = await call("POST", "/v1/webhooks", {
    url: `https://${WEBHOOK_HOST}/sink`,
    secret: SECRET,
    events: ["*"],
  });
  assert.equal(hook.status, 201);
  assert.deepEqual(hook.body.data.signing.v2, {
    header: "X-Titen-Signature-V2",
    timestamp_header: "X-Titen-Timestamp",
    delivery_header: "X-Titen-Delivery",
    signed_content: "<timestamp>.<delivery>.<raw body>",
    algorithm: "HMAC-SHA256",
    format: "sha256=<hex>",
    tolerance_seconds: 300,
  });

  // Any canonical write emits an event, which is what gets delivered.
  const observation = await call("POST", "/v1/observations", {
    subject_id: "user_hook",
    kind: "tool_result",
    content: "Webhook delivery integration evidence.",
    source: { type: "tool", ref: "hook#1" },
    trust: "verified",
  });
  assert.equal(observation.status, 201);

  const drained = await call("POST", "/v1/webhooks/deliver", {});
  assert.equal(drained.status, 200);
  assert.ok(drained.body.data.events_drained >= 1);

  assert.ok(received.length >= 1, "the receiver must have been contacted over HTTP");
  const first = received[0]!;
  assert.equal(first.event, "observation.appended");
  assert.ok(first.delivery?.startsWith("dlv_"), "a delivery id must identify the attempt");
  assert.match(first.signature ?? "", /^sha256=[0-9a-f]{64}$/);

  // The receiver can authenticate the payload without trusting the sender.
  const expected = await hmacHex(SECRET, first.body);
  assert.equal(
    first.signature,
    `sha256=${expected}`,
    "the signature must verify against the shared secret and the exact body",
  );

  // A tampered body must not verify, which is the property the header exists for.
  const tampered = await hmacHex(SECRET, `${first.body} `);
  assert.notEqual(first.signature, `sha256=${tampered}`);

  // The timestamped signature binds the attempt time and the delivery id, so a
  // receiver can reject an old capture and drop a duplicate delivery.
  assert.match(first.timestamp ?? "", /^\d+$/);
  assert.ok(Math.abs(Number(first.timestamp) - Date.now() / 1000) < 300, "the timestamp must be current");
  assert.equal(
    first.signatureV2,
    `sha256=${await hmacHex(SECRET, `${first.timestamp}.${first.delivery}.${first.body}`)}`,
    "the V2 signature must verify against the timestamp, the delivery id, and the exact body",
  );
  assert.notEqual(
    first.signatureV2,
    `sha256=${await hmacHex(SECRET, `${Number(first.timestamp) + 1}.${first.delivery}.${first.body}`)}`,
  );

  const payload = JSON.parse(first.body);
  assert.ok(payload, "the body must be the JSON event payload");
  assert.ok(
    !first.body.includes(SECRET) && !first.body.includes("titen_sk_"),
    "a delivery must carry no credential",
  );
});

test("a rejecting receiver is recorded as a failure and scheduled for retry", async () => {
  received = [];
  rejectNext = true;
  transportMode = "deliver";

  const hook = await call("POST", "/v1/webhooks", {
    url: `https://${WEBHOOK_HOST}/failing`,
    secret: SECRET,
    events: ["claim.materialized"],
  });
  assert.equal(hook.status, 201);
  const hookId = hook.body.data.webhook_id as string;

  const observation = await call("POST", "/v1/observations", {
    subject_id: "user_hook_fail",
    kind: "tool_result",
    content: "Evidence for a delivery that will be refused.",
    source: { type: "tool", ref: "hook#2" },
    trust: "verified",
  });
  await call("POST", "/v1/consolidations", {
    subject_id: "user_hook_fail",
    claims: [
      {
        kind: "procedural",
        statement: "A refused delivery must be retried, not lost.",
        sources: [
          { observation_id: observation.body.data.observation_id, relation: "supports" },
        ],
      },
    ],
  });

  const drain = await call("POST", "/v1/webhooks/deliver", {});
  assert.ok(received.length >= 1, "the failing receiver must still have been contacted");

  const deliveries = await call("GET", `/v1/webhooks/${hookId}/deliveries`, undefined);
  assert.equal(deliveries.status, 200);
  const attempted = deliveries.body.data.deliveries;
  assert.ok(attempted.length >= 1, "the attempt must be recorded");
  const record = attempted[0];
  assert.match(
    record.event_id,
    /^evt_/,
    "a delivery must identify the event it attempted, not its kind",
  );
  assert.equal(record.response_status, 500, "the receiver's status must be stored");
  assert.equal(record.attempts, 1);
  assert.ok(record.next_retry_at, "a failed delivery must be scheduled for retry");
  // oldest_retry_at is the minimum over every pending delivery of this principal.
  // Retry times now come from each attempt's own finish time, so another hook's
  // delivery in the same pass can be a few milliseconds earlier, never later.
  const oldestRetry = Date.parse(drain.body.data.oldest_retry_at);
  assert.ok(oldestRetry <= Date.parse(record.next_retry_at), "the oldest retry is not after this one");
  assert.ok(Date.parse(record.next_retry_at) - oldestRetry < 10_000, "both retries come from the same pass");
  assert.equal(record.status, "pending", "it stays pending until attempts are exhausted");

  rejectNext = false;
});

test("an unreachable destination fails without taking down the request", async () => {
  received = [];
  transportMode = "unreachable";
  const hook = await call("POST", "/v1/webhooks", {
    url: `https://${WEBHOOK_HOST}/unreachable`,
    secret: SECRET,
    events: ["*"],
  });
  assert.equal(hook.status, 201);
  const hookId = hook.body.data.webhook_id as string;

  await call("POST", "/v1/observations", {
    subject_id: "user_hook_unreachable",
    kind: "tool_result",
    content: "Evidence whose delivery target does not answer.",
    source: { type: "tool", ref: "hook#3" },
    trust: "verified",
  });

  const drained = await call("POST", "/v1/webhooks/deliver", {});
  assert.equal(drained.status, 200, "a network failure must not fail the drain request");

  const deliveries = await call("GET", `/v1/webhooks/${hookId}/deliveries`, undefined);
  assert.ok(deliveries.body.data.deliveries.length >= 1, "the attempt is still recorded");
  assert.equal(
    deliveries.body.data.deliveries[0].response_status,
    null,
    "no status exists when the connection never completed",
  );
});

test("a second event of the same kind is also delivered", async () => {
  // The regression this file exists to prevent. Delivery dedup once matched on
  // event kind, so the first observation.appended was sent and every later one was
  // silently skipped forever.
  received = [];
  rejectNext = false;
  transportMode = "deliver";

  const hook = await call("POST", "/v1/webhooks", {
    url: `https://${WEBHOOK_HOST}/repeat`,
    secret: SECRET,
    events: ["observation.appended"],
  });
  assert.equal(hook.status, 201);
  const hookId = hook.body.data.webhook_id as string;

  const write = async (marker: string) => {
    const res = await call("POST", "/v1/observations", {
      subject_id: "user_hook_repeat",
      kind: "tool_result",
      content: `Repeat delivery evidence ${marker}.`,
      source: { type: "tool", ref: `repeat#${marker}` },
      trust: "verified",
    });
    assert.equal(res.status, 201);
    return res.body.data.observation_id as string;
  };

  await write("first");
  await call("POST", "/v1/webhooks/deliver", {});
  const afterFirst = (
    await call("GET", `/v1/webhooks/${hookId}/deliveries`, undefined)
  ).body.data.deliveries.length;
  assert.ok(afterFirst >= 1, "the first event of this kind must be delivered");

  await write("second");
  await call("POST", "/v1/webhooks/deliver", {});
  const afterSecond = (
    await call("GET", `/v1/webhooks/${hookId}/deliveries`, undefined)
  ).body.data.deliveries;

  assert.ok(
    afterSecond.length > afterFirst,
    `a later event of the same kind must also be delivered; had ${afterFirst}, now ${afterSecond.length}`,
  );
  // Each attempt names a distinct event, which is what makes dedup correct.
  const ids = afterSecond.map((d: any) => d.event_id);
  assert.equal(new Set(ids).size, ids.length, "each delivery must target one event");
});

test("draining twice does not deliver the same event twice", async () => {
  received = [];
  rejectNext = false;
  transportMode = "deliver";

  const hook = await call("POST", "/v1/webhooks", {
    url: `https://${WEBHOOK_HOST}/once`,
    secret: SECRET,
    events: ["*"],
  });
  const hookId = hook.body.data.webhook_id as string;

  await call("POST", "/v1/observations", {
    subject_id: "user_hook_once",
    kind: "tool_result",
    content: "Evidence that must be delivered exactly once.",
    source: { type: "tool", ref: "once#1" },
    trust: "verified",
  });

  await call("POST", "/v1/webhooks/deliver", {});
  const first = (await call("GET", `/v1/webhooks/${hookId}/deliveries`, undefined)).body.data
    .deliveries.length;
  await call("POST", "/v1/webhooks/deliver", {});
  const second = (await call("GET", `/v1/webhooks/${hookId}/deliveries`, undefined)).body.data
    .deliveries.length;

  assert.equal(second, first, "a repeated drain must not re-deliver an event");
});