Webhooks

const { createHmac, timingSafeEqual } = require("node:crypto");

// rawBody: the request body exactly as received (a Buffer, or the raw string).
// now: the current time in Unix seconds.
function verifyWebhook(secret, headers, rawBody, now) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatures = headers["webhook-signature"];
  if (!id || !timestamp || !signatures) return false;
  const seconds = Number(timestamp);
  if (!Number.isInteger(seconds) || Math.abs(now - seconds) > 300) {
    return false;
  }
  const key = Buffer.from(secret.slice("whsec_".length), "base64");
  const expected = Buffer.from(
    "v1," +
      createHmac("sha256", key)
        .update(`${id}.${timestamp}.`)
        .update(rawBody)
        .digest("base64"),
  );
  return signatures.split(" ").some((candidate) => {
    const given = Buffer.from(candidate);
    return given.length === expected.length && timingSafeEqual(given, expected);
  });
}

import base64, hashlib, hmac, time


# raw_body: the request body bytes exactly as received.
def verify_webhook(secret, headers, raw_body, now=None):
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not (msg_id and timestamp and signatures and timestamp.isdigit()):
        return False
    if now is None:
        now = int(time.time())
    if abs(now - int(timestamp)) > 300:
        return False
    key = base64.b64decode(secret[len("whsec_"):])
    signed = f"{msg_id}.{timestamp}.".encode() + raw_body
    expected = "v1," + base64.b64encode(
        hmac.new(key, signed, hashlib.sha256).digest()
    ).decode()
    return any(
        hmac.compare_digest(expected, candidate)
        for candidate in signatures.split(" ")
    )