Webhooks

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

const TOLERANCE_SECONDS = 5 * 60;

/**
 * Verify one andteam webhook delivery.
 *
 * secret   the endpoint's signing secret, "whsec_" followed by base64
 * headers  the request headers, names in lower case
 * body     the RAW request body, exactly as received (do not re-serialise it)
 *
 * Returns the parsed event, or throws.
 */
export function verifyAndteamWebhook(secret, headers, body, nowMs = Date.now()) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatures = headers["webhook-signature"];
  if (!id || !timestamp || !signatures) throw new Error("Missing webhook headers");

  // Reject anything more than 5 minutes from now, either way.
  if (!/^\d+$/.test(timestamp) || Math.abs(nowMs / 1000 - Number(timestamp)) > TOLERANCE_SECONDS) {
    throw new Error("Timestamp out of range");
  }

  // The HMAC key is the base64-DECODED part after "whsec_", not the string itself.
  const key = Buffer.from(secret.slice("whsec_".length), "base64");
  const expected = createHmac("sha256", key).update(`${id}.${timestamp}.${body}`).digest();

  // While a secret is being rolled the header holds two signatures, "v1,<a> v1,<b>": any one may match.
  for (const part of signatures.split(" ")) {
    const [version, value] = part.split(",");
    if (version !== "v1" || !value) continue;
    const given = Buffer.from(value, "base64");
    if (given.length === expected.length && timingSafeEqual(given, expected)) return JSON.parse(body);
  }
  throw new Error("No matching signature");
}

import base64, hashlib, hmac, json, time

TOLERANCE_SECONDS = 5 * 60


def verify_andteam_webhook(secret: str, headers: dict, body: bytes, now: float | None = None):
    """Return the parsed event, or raise ValueError. `body` is the raw request body."""
    id_ = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not id_ or not timestamp or not signatures:
        raise ValueError("Missing webhook headers")
    if not timestamp.isdigit() or abs((now or time.time()) - int(timestamp)) > TOLERANCE_SECONDS:
        raise ValueError("Timestamp out of range")

    key = base64.b64decode(secret[len("whsec_"):])
    signed = f"{id_}.{timestamp}.".encode() + body
    expected = hmac.new(key, signed, hashlib.sha256).digest()

    for part in signatures.split(" "):
        version, _, value = part.partition(",")
        if version != "v1" or not value:
            continue
        if hmac.compare_digest(base64.b64decode(value), expected):
            return json.loads(body)
    raise ValueError("No matching signature")