Webhooks

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

export function verifyWebhook(
  rawBody: string, // exact bytes; read before any JSON parsing
  headers: { id: string; timestamp: string; signature: string },
  secret: string, // whsec_...
): boolean {
  if (!headers.id || !headers.timestamp || !headers.signature) return false;

  // Reject stale timestamps (replay protection)
  const skew = Math.abs(Date.now() / 1000 - Number(headers.timestamp));
  if (!Number.isFinite(skew) || skew > 300) return false;

  const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");
  const signedContent = `${headers.id}.${headers.timestamp}.${rawBody}`;
  const expected = Buffer.from(
    createHmac("sha256", key).update(signedContent).digest("base64"),
  );

  // Header may carry several space-separated "v1,<base64>" candidates
  return headers.signature.split(" ").some((part) => {
    const [version, value] = part.trim().split(",", 2);
    if (version !== "v1" || !value) return false;
    const candidate = Buffer.from(value);
    return candidate.length === expected.length && timingSafeEqual(candidate, expected);
  });
}

// Express example: note express.raw(), not express.json()
app.post("/webhooks/prescience", express.raw({ type: "application/json" }), (req, res) => {
  const ok = verifyWebhook(
    req.body.toString("utf8"),
    {
      id: req.header("webhook-id") ?? "",
      timestamp: req.header("webhook-timestamp") ?? "",
      signature: req.header("webhook-signature") ?? "",
    },
    process.env.PRESCIENCE_WEBHOOK_SECRET!,
  );
  if (!ok) return res.status(401).send("invalid signature");

  const event = JSON.parse(req.body.toString("utf8"));
  // ...route on event.type, dedupe on event.id...
  res.status(200).send("ok");
});

import base64
import hashlib
import hmac
import time


def verify_webhook(raw_body: bytes, headers: dict, secret: str) -> bool:
    """raw_body must be the exact request bytes; verify before parsing JSON."""
    event_id = headers.get("webhook-id", "")
    timestamp = headers.get("webhook-timestamp", "")
    signature = headers.get("webhook-signature", "")
    if not (event_id and timestamp and signature):
        return False

    # Reject stale timestamps (replay protection)
    try:
        if abs(time.time() - int(timestamp)) > 300:
            return False
    except ValueError:
        return False

    key = base64.b64decode(secret.removeprefix("whsec_"))
    signed_content = f"{event_id}.{timestamp}.".encode() + raw_body
    expected = base64.b64encode(hmac.new(key, signed_content, hashlib.sha256).digest()).decode()

    # Header may carry several space-separated "v1,<base64>" candidates
    for part in signature.split(" "):
        version, _, value = part.strip().partition(",")
        if version == "v1" and value and hmac.compare_digest(value, expected):
            return True
    return False


# Flask example
from flask import Flask, request

app = Flask(__name__)

@app.post("/webhooks/prescience")
def prescience_webhook():
    if not verify_webhook(request.get_data(), request.headers, WEBHOOK_SECRET):
        return "invalid signature", 401

    event = request.get_json()
    # ...route on event["type"], dedupe on event["id"]...
    return "ok", 200