Webhooks

import base64
import hashlib
import hmac
import time


def verify_webhook(secret: str, headers: dict, body: bytes) -> bool:
    msg_id = headers["webhook-id"]
    timestamp = headers["webhook-timestamp"]

    # Replay guard: 5 minute tolerance.
    if abs(time.time() - int(timestamp)) > 300:
        return False

    key = base64.b64decode(secret.removeprefix("whsec_"))
    signed_content = f"{msg_id}.{timestamp}.".encode() + body
    expected = base64.b64encode(
        hmac.new(key, signed_content, hashlib.sha256).digest()
    ).decode()

    # The header may list several space-delimited signatures during a
    # secret rotation; accept if any v1 signature matches.
    for versioned in headers["webhook-signature"].split():
        version, _, signature = versioned.partition(",")
        if version == "v1" and hmac.compare_digest(signature, expected):
            return True
    return False

const crypto = require("crypto");

function verifyWebhook(secret, headers, rawBody) {
  const msgId = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];

  // Replay guard: 5 minute tolerance.
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) {
    return false;
  }

  const key = Buffer.from(secret.slice("whsec_".length), "base64");
  const signedContent = `${msgId}.${timestamp}.${rawBody}`;
  const expected = crypto
    .createHmac("sha256", key)
    .update(signedContent)
    .digest("base64");

  return (headers["webhook-signature"] || "").split(" ").some((versioned) => {
    const comma = versioned.indexOf(",");
    if (comma === -1 || versioned.slice(0, comma) !== "v1") return false;
    const signature = Buffer.from(versioned.slice(comma + 1));
    const expectedBuf = Buffer.from(expected);
    return (
      signature.length === expectedBuf.length &&
      crypto.timingSafeEqual(signature, expectedBuf)
    );
  });
}