Webhooks

{
  "id": "01H7X6DEMOA5RVQ8D3QF6M0Q9E",
  "type": "key.created",
  "created_at": "2026-07-11T10:00:00Z",
  "tenant_id": "acme",
  "data": { "...": "event-specific" }
}

import crypto from "node:crypto";
 
/**
 * Verify a webhook. `secret` is the whsec_-prefixed string you saved
 * from the portal. `headers` is a plain object of the incoming
 * request headers (all lower-case). `body` is the raw request body
 * bytes as a Buffer or string — DO NOT re-serialise a parsed JSON
 * object or the signature will not match.
 */
export function verifyWebhook(secret, headers, body) {
  const id = headers["webhook-id"];
  const ts = headers["webhook-timestamp"];
  const sigHeader = headers["webhook-signature"];
  if (!id || !ts || !sigHeader) throw new Error("missing webhook headers");
 
  const rawSecret = secret.startsWith("whsec_") ? secret.slice(6) : secret;
  const key = Buffer.from(rawSecret, "base64");
  const signed = `${id}.${ts}.${typeof body === "string" ? body : body.toString("utf8")}`;
  const expected = crypto.createHmac("sha256", key).update(signed).digest("base64");
 
  // sigHeader is space-separated "v1,<b64> [v1,<b64> …]"
  for (const entry of sigHeader.split(" ")) {
    const [scheme, sig] = entry.split(",");
    if (scheme === "v1" && crypto.timingSafeEqual(Buffer.from(sig), Buffer.from(expected))) {
      return true;
    }
  }
  throw new Error("no matching v1 signature");
}

import base64, hmac, hashlib
 
def verify_webhook(secret: str, headers: dict, body: bytes) -> bool:
    msg_id = headers["webhook-id"]
    ts     = headers["webhook-timestamp"]
    sigs   = headers["webhook-signature"]
 
    raw = base64.b64decode(secret[6:] if secret.startswith("whsec_") else secret)
    signed = f"{msg_id}.{ts}.".encode() + body
    expected = base64.b64encode(hmac.new(raw, signed, hashlib.sha256).digest()).decode()
 
    for entry in sigs.split(" "):
        scheme, _, sig = entry.partition(",")
        if scheme == "v1" and hmac.compare_digest(sig, expected):
            return True
    return False

package main
 
import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/base64"
    "errors"
    "strings"
)
 
func VerifyWebhook(secret string, headers map[string]string, body []byte) error {
    id  := headers["webhook-id"]
    ts  := headers["webhook-timestamp"]
    sig := headers["webhook-signature"]
 
    raw, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_"))
    if err != nil {
        return err
    }
    mac := hmac.New(sha256.New, raw)
    mac.Write([]byte(id + "." + ts + "."))
    mac.Write(body)
    expected := base64.StdEncoding.EncodeToString(mac.Sum(nil))
 
    for _, entry := range strings.Split(sig, " ") {
        scheme, sigB64, ok := strings.Cut(entry, ",")
        if !ok || scheme != "v1" {
            continue
        }
        if hmac.Equal([]byte(sigB64), []byte(expected)) {
            return nil
        }
    }
    return errors.New("no matching v1 signature")
}