feat(reliability): idempotency store, custom webhook replay protection, correlation ids

Add LICENSE (MIT), an IdempotencyStore abstraction with a KV implementation and provider-scoped delivery keys, optional replay protection for custom webhooks (X-WebHooker-Timestamp + X-WebHooker-Nonce), and per-request correlation ids in webhook responses and logs.

server/lib/lib/correlation.ts

@@ -0,0 +1,7 @@
+/** A short, unique correlation id for tracing one webhook request. */
+export function newCorrelationId(): string {
+  if (typeof crypto !== "undefined" && typeof crypto.randomUUID === "function") {
+    return crypto.randomUUID();
+  }
+  return `${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 10)}`;
+}

server/lib/lib/idempotency.ts

@@ -0,0 +1,39 @@
+/**
+ * Delivery idempotency: a small, reusable abstraction over "has this key been
+ * seen / can I claim it once" so webhook dedup, nonce replay protection and
+ * future delivery retry all share one semantics. `claim` is best-effort atomic
+ * on KV (get-then-put): under a concurrent double-send both callers may see
+ * "unclaimed", but that matches the existing dedup behavior and is acceptable
+ * because dispatch itself is idempotent per (provider, group, delivery).
+ */
+export interface IdempotencyStore {
+  has(key: string): Promise<boolean>;
+  claim(key: string, ttlSeconds: number): Promise<boolean>;
+}
+
+export function kvIdempotencyStore(kv: KVNamespace): IdempotencyStore {
+  return {
+    async has(key): Promise<boolean> {
+      const value = await kv.get(key);
+      return value !== null && value !== undefined;
+    },
+    async claim(key, ttlSeconds): Promise<boolean> {
+      const value = await kv.get(key);
+      if (value !== null && value !== undefined) return false;
+      await kv.put(key, "1", { expirationTtl: ttlSeconds });
+      return true;
+    },
+  };
+}
+
+/**
+ * Canonical dedup key for a webhook delivery: provider + tenant + delivery id.
+ * The legacy global endpoint has no tenant, so `groupId` is "global".
+ */
+export function deliveryKey(
+  provider: string,
+  groupId: string | undefined,
+  deliveryId: string,
+): string {
+  return `delivery:${provider}:${groupId ?? "global"}:${deliveryId}`;
+}

server/lib/providers/custom/index.ts

@@ -1,12 +1,55 @@
 import type { Env, WebhookEvent } from "../../types";
 import type { Provider } from "../types";
+import { hmacSha256Hex, timingSafeEqual } from "../hmac";
 import { verifySignature } from "../github/verify";
 
+const REPLAY_WINDOW_SECONDS = 300;
+const NONCE_TTL_SECONDS = 600;
+
+function parseTimestamp(value: string | undefined): number | null {
+  if (!value) return null;
+  const n = Number(value);
+  return Number.isFinite(n) ? n : null;
+}
+
+/**
+ * Replay protection for `custom` webhooks: the sender adds `X-WebHooker-Timestamp`
+ * (Unix seconds) and `X-WebHooker-Nonce` and signs
+ * `timestamp + "." + nonce + "." + body` instead of the bare body. We reject
+ * timestamps outside a ±5min window and reject any nonce already seen (nonces
+ * live in KV for 10min). Senders without those headers keep the legacy
+ * body-only signature so existing integrations continue to work.
+ */
+async function verifyReplayProtected(
+  body: string,
+  headers: Record<string, string>,
+  secret: string,
+  kv: KVNamespace,
+): Promise<boolean> {
+  const timestamp = parseTimestamp(headers["x-webhooker-timestamp"]);
+  const nonce = headers["x-webhooker-nonce"];
+  const signature = headers["x-webhooker-signature"];
+  if (timestamp == null || !nonce || !secret) return false;
+
+  const now = Math.floor(Date.now() / 1000);
+  if (Math.abs(now - timestamp) > REPLAY_WINDOW_SECONDS) return false;
+
+  const expected = `sha256=${await hmacSha256Hex(secret, `${timestamp}.${nonce}.${body}`)}`;
+  if (!timingSafeEqual(signature ?? "", expected)) return false;
+
+  const nonceKey = `nonce:${nonce}`;
+  const seen = await kv.get(nonceKey);
+  if (seen !== null && seen !== undefined) return false;
+  await kv.put(nonceKey, "1", { expirationTtl: NONCE_TTL_SECONDS });
+  return true;
+}
+
 /**
  * Custom webhook provider: accepts arbitrary JSON posts (monitoring, CI,
- * scripts, ...) that are not signed by a forge. The sender signs the raw body
+ * scripts, ...) that are not signed by a forge. The sender signs the payload
  * with the tenant's secret using the GitHub-style `sha256=<hex>` HMAC header
- * `X-WebHooker-Signature`. Payloads become `custom` events that flow through
+ * `X-WebHooker-Signature`, optionally adding replay-protection headers (see
+ * `verifyReplayProtected`). Payloads become `custom` events that flow through
  * the normal route matching pipeline (a route with `event: custom`).
  */
 export const customProvider: Provider = {
@@ -24,7 +67,14 @@ export const customProvider: Provider = {
     // The tenant webhook handler overrides GITHUB_WEBHOOK_SECRET with the
     // group's secret; on the legacy global endpoint this falls back to the
     // operator's global secret.
-    return verifySignature(body, headers["x-webhooker-signature"], env.GITHUB_WEBHOOK_SECRET);
+    const secret = env.GITHUB_WEBHOOK_SECRET;
+    if (
+      headers["x-webhooker-timestamp"] !== undefined &&
+      headers["x-webhooker-nonce"] !== undefined
+    ) {
+      return verifyReplayProtected(body, headers, secret, env.KV);
+    }
+    return verifySignature(body, headers["x-webhooker-signature"], secret);
   },
 
   parse(body, _headers): WebhookEvent | null {

server/lib/webhook.ts

@@ -9,6 +9,8 @@ import { getTenantSecret } from "./web/tenants";
 import { recordAudit } from "./lib/audit";
 import { cfEnv, cfWaitUntil, headersFrom } from "./cf";
 import { log } from "./lib/log";
+import { deliveryKey, kvIdempotencyStore } from "./lib/idempotency";
+import { newCorrelationId } from "./lib/correlation";
 
 const MAX_BODY_SIZE = 1024 * 1024;
 
@@ -32,6 +34,7 @@ export async function processWebhook(
   waitUntil: (promise: Promise<unknown>) => void,
   tenantId?: string,
 ): Promise<WebhookResult> {
+  const requestId = newCorrelationId();
   let effectiveEnv = env;
   const groups = await loadGroups(env.KV);
   if (tenantId) {
@@ -56,7 +59,10 @@ export async function processWebhook(
   } catch (err) {
     // A malformed secret or an unavailable crypto implementation must fail as
     // a clean 401, never an uncaught 500.
-    log.warn({ provider: provider.id, err: String(err) }, "Webhook signature verification failed");
+    log.warn(
+      { provider: provider.id, requestId, err: String(err) },
+      "Webhook signature verification failed",
+    );
     return { status: 401, body: { error: "Invalid signature" } };
   }
   if (!verified) {
@@ -67,9 +73,12 @@ export async function processWebhook(
         ? effectiveEnv.GITEA_WEBHOOK_SECRET
         : effectiveEnv.GITHUB_WEBHOOK_SECRET;
     if (!secret) {
-      log.warn({ provider: provider.id }, "Webhook rejected: provider secret is not configured");
+      log.warn(
+        { provider: provider.id, requestId },
+        "Webhook rejected: provider secret is not configured",
+      );
     } else {
-      log.warn({ provider: provider.id }, "Webhook rejected: invalid signature");
+      log.warn({ provider: provider.id, requestId }, "Webhook rejected: invalid signature");
     }
     return { status: 401, body: { error: "Invalid signature" } };
   }
@@ -111,16 +120,14 @@ export async function processWebhook(
   }
 
   if (event.deliveryId) {
-    // Tenant-scoped dedup keys: different accounts can reuse the same
-    // delivery id, so the global key would wrongly dedupe across tenants.
-    const key = tenantId
-      ? `delivery:${tenantId}:${event.deliveryId}`
-      : `delivery:${event.deliveryId}`;
-    const seen = await env.KV.get(key);
-    if (seen) {
-      return { status: 200, body: { ok: true, duplicate: true } };
+    // Dedup via the idempotency store: a provider/tenant-scoped key means
+    // different accounts may reuse a delivery id without colliding, while
+    // retries of the same delivery never dispatch twice.
+    const store = kvIdempotencyStore(env.KV);
+    const key = deliveryKey(provider.id, tenantId, event.deliveryId);
+    if (!(await store.claim(key, 300))) {
+      return { status: 200, body: { ok: true, duplicate: true, requestId } };
     }
-    await env.KV.put(key, "1", { expirationTtl: 300 });
   }
 
   const config = await loadConfig(env);
@@ -129,11 +136,11 @@ export async function processWebhook(
   }
 
   const dispatch = dispatchEvent(config, event, env, groups).catch((err) =>
-    log.error(err, "Dispatch failed"),
+    log.error({ requestId, err }, "Dispatch failed"),
   );
   waitUntil(dispatch);
 
-  return { status: 200, body: { ok: true } };
+  return { status: 200, body: { ok: true, requestId } };
 }
 
 /** h3 wrapper for `POST /webhook` / `POST /webhook/:groupId`. */