Webhook subscriptions

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

// IMPORTANT: capture the raw body BEFORE any JSON middleware parses it.
// app.use("/webhooks/chert", express.raw({ type: "application/json" }))

export function verifyChertWebhook(rawBody, headers, subscriptionSecret) {
  // Prefer the modern header; fall back to the legacy one.
  const header =
    headers["x-webhook-signature"] || headers["x-chert-signature"] || ""

  const modern = /^t=(\d+),v1=([a-f0-9]+)$/i.exec(header)
  const legacy = /^v1,(\d+),([a-f0-9]+)$/i.exec(header)
  const m = modern || legacy
  if (!m) return false

  const ts = Number(m[1])
  const provided = m[2].toLowerCase()

  // 5-minute replay window.
  if (Math.abs(Date.now() / 1000 - ts) > 300) return false

  const expected = createHmac("sha256", subscriptionSecret)
    .update(`${ts}.${rawBody}`)        // <-- period, not colon
    .digest("hex")

  const a = Buffer.from(provided, "hex")
  const b = Buffer.from(expected, "hex")
  return a.length === b.length && timingSafeEqual(a, b)
}

import hmac, hashlib, re, time

# IMPORTANT: read request.get_data() — NOT request.get_json() —
# so you verify against the exact bytes Chert sent.

def verify_chert_webhook(raw_body: bytes, headers, subscription_secret: str) -> bool:
    header = headers.get("X-Webhook-Signature") or headers.get("x-chert-signature") or ""
    m = re.match(r"^t=(\d+),v1=([a-f0-9]+)$", header) \
        or re.match(r"^v1,(\d+),([a-f0-9]+)$", header)
    if not m:
        return False
    ts, provided = int(m.group(1)), m.group(2).lower()
    if abs(time.time() - ts) > 300:
        return False
    expected = hmac.new(
        subscription_secret.encode(),
        f"{ts}.".encode() + raw_body,    # <-- period, then raw bytes
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(provided, expected)

# Sanity-check a single delivery from your logs.
TS=1715000000                                # from X-Webhook-Timestamp
BODY='{"event":"message.received",...}'      # the exact bytes received
HEX_FROM_HEADER=abcdef...                    # the v1=... portion

EXPECTED=$(printf '%s.%s' "$TS" "$BODY" \
  | openssl dgst -sha256 -hmac "$SUBSCRIPTION_SECRET" -hex \
  | awk '{print $2}')

[ "$EXPECTED" = "$HEX_FROM_HEADER" ] && echo "ok" || echo "mismatch"