Webhook signature verifier

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

// rawBody: the request body exactly as received (a string), before any JSON parsing.
export function verifyNotiflyRequest(rawBody, header, secretKey, toleranceMs = 5 * 60 * 1000) {
  const fields = {};
  for (const part of header.split(",")) {
    const at = part.indexOf("=");
    const key = part.slice(0, at).trim();
    if (at > 0 && !(key in fields)) fields[key] = part.slice(at + 1);
  }
  const t = Number(fields.t);
  if (!Number.isFinite(t) || Math.abs(Date.now() - t) > toleranceMs) return false;
  const expected = createHmac("sha256", secretKey).update(`${t}.${rawBody === "" ? "{}" : rawBody}`).digest("hex");
  const received = Buffer.from(fields.v1 ?? "");
  return received.length === expected.length && timingSafeEqual(received, Buffer.from(expected));
}

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

// headers: the request headers, lower-cased; rawBody: the body exactly as received (a string).
export function verifyStandardWebhook(rawBody, headers, secret, toleranceSeconds = 5 * 60) {
  const id = headers["webhook-id"];
  const timestamp = Number.parseInt(headers["webhook-timestamp"], 10);
  if (!id || !Number.isFinite(timestamp) || Math.abs(Date.now() / 1000 - timestamp) > toleranceSeconds) return false;
  const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");
  const expected = Buffer.from(createHmac("sha256", key).update(`${id}.${timestamp}.${rawBody}`).digest("base64"));
  return (headers["webhook-signature"] ?? "").split(" ").some((entry) => {
    const [version, signature] = entry.split(",");
    const received = Buffer.from(signature ?? "");
    return version === "v1" && received.length === expected.length && timingSafeEqual(received, expected);
  });
}

import base64
import hashlib
import hmac
import time


def verify_standard_webhook(raw_body: bytes, headers: dict, secret: str, tolerance_seconds: int = 300) -> bool:
    msg_id = headers.get("webhook-id", "")
    timestamp = headers.get("webhook-timestamp", "")
    if not msg_id or not timestamp.isdigit() or abs(time.time() - int(timestamp)) > tolerance_seconds:
        return False
    key = base64.b64decode(secret.removeprefix("whsec_"))
    signed = f"{msg_id}.{int(timestamp)}.".encode() + raw_body
    expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode()
    return any(
        version == "v1" and hmac.compare_digest(signature, expected)
        for version, _, signature in (entry.partition(",") for entry in headers.get("webhook-signature", "").split(" "))
    )