Webhooks

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

const TOLERANCE_SECONDS = 5 * 60;

export function verifySignature(params: {
  rawBody: string | Buffer;       // exactly as received
  signatureHeader: string | undefined;
  secret: string;                 // shown once at endpoint creation
}): boolean {
  if (!params.signatureHeader) return false;

  const parts = new Map<string, string>();
  for (const piece of params.signatureHeader.split(",")) {
    const eq = piece.indexOf("=");
    if (eq > 0) parts.set(piece.slice(0, eq).trim(), piece.slice(eq + 1).trim());
  }
  const timestamp = parts.get("t");
  // Every v1, not just the first: a rotation window carries one per secret.
  const signatures = params.signatureHeader
    .split(",")
    .map((piece) => piece.split("="))
    .filter(([key]) => key?.trim() === "v1")
    .map(([, value]) => value?.trim() ?? "");
  if (!timestamp || signatures.length === 0) return false;

  const timestampSeconds = Number(timestamp);
  if (!Number.isFinite(timestampSeconds)) return false;
  if (Math.abs(Date.now() / 1000 - timestampSeconds) > TOLERANCE_SECONDS) return false;

  const expected = createHmac("sha256", params.secret)
    .update(`${timestamp}.`)
    .update(params.rawBody)
    .digest();
  let matched = false;
  for (const signature of signatures) {
    let received: Buffer;
    try {
      received = Buffer.from(signature, "hex");
    } catch {
      continue;
    }
    // Compare every candidate even after a match, so the work does not
    // depend on WHICH signature matched.
    if (received.length === expected.length && timingSafeEqual(received, expected)) {
      matched = true;
    }
  }
  return matched;
}

import hashlib
import hmac
import time

TOLERANCE_SECONDS = 5 * 60


def verify_signature(raw_body: bytes, signature_header: str | None, secret: str) -> bool:
    if not signature_header:
        return False

    timestamp: str | None = None
    signatures: list[str] = []
    for piece in signature_header.split(","):
        key, eq, value = piece.partition("=")
        if not eq:
            continue
        if key.strip() == "t":
            timestamp = value.strip()
        elif key.strip() == "v1":
            # Every v1: a rotation window carries one per valid secret.
            signatures.append(value.strip())
    if not timestamp or not signatures:
        return False

    try:
        timestamp_seconds = float(timestamp)
    except ValueError:
        return False
    if abs(time.time() - timestamp_seconds) > TOLERANCE_SECONDS:
        return False

    expected = hmac.new(
        secret.encode(), f"{timestamp}.".encode() + raw_body, hashlib.sha256
    ).hexdigest()
    # Compare every candidate even after a match, so the work does not
    # depend on WHICH signature matched.
    matched = False
    for signature in signatures:
        if hmac.compare_digest(expected, signature):
            matched = True
    return matched