Webhook

const crypto = require("crypto");

const TOLERANCE_SEC = 5 * 60;

// rawBody: the request body as a string, before JSON.parse
// (e.g. express.raw({ type: "application/json" }) → req.body.toString("utf8"))
function verifyWebhook(rawBody, headers, secret) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatureHeader = headers["webhook-signature"];
  if (!id || !timestamp || !signatureHeader) throw new Error("Missing webhook headers");

  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - Number(timestamp)) > TOLERANCE_SEC) throw new Error("Timestamp outside tolerance");

  const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");
  const expected = crypto
    .createHmac("sha256", key)
    .update(`${id}.${timestamp}.${rawBody}`)
    .digest("base64");

  const valid = signatureHeader.split(" ").some((part) => {
    const [version, signature] = part.split(",");
    return version === "v1" && signature.length === expected.length &&
      crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected));
  });
  if (!valid) throw new Error("Invalid signature");
  return JSON.parse(rawBody);
}

import base64
import hashlib
import hmac
import json
import time

TOLERANCE_SEC = 5 * 60


# raw_body: the request body as bytes, before json.loads
# (e.g. Flask: request.get_data(), Django: request.body)
def verify_webhook(raw_body: bytes, headers: dict, secret: str) -> dict:
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signature_header = headers.get("webhook-signature")
    if not (msg_id and timestamp and signature_header):
        raise ValueError("Missing webhook headers")

    if abs(time.time() - int(timestamp)) > TOLERANCE_SEC:
        raise ValueError("Timestamp outside tolerance")

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

    for part in signature_header.split(" "):
        version, _, signature = part.partition(",")
        if version == "v1" and hmac.compare_digest(signature, expected):
            return json.loads(raw_body)
    raise ValueError("Invalid signature")

<?php
const TOLERANCE_SEC = 300;

// $rawBody: file_get_contents('php://input'), before json_decode
// $headers: header names in lower case
function verifyWebhook(string $rawBody, array $headers, string $secret): array
{
    $id = $headers['webhook-id'] ?? null;
    $timestamp = $headers['webhook-timestamp'] ?? null;
    $signatureHeader = $headers['webhook-signature'] ?? null;
    if (!$id || !$timestamp || !$signatureHeader) {
        throw new RuntimeException('Missing webhook headers');
    }

    if (abs(time() - (int) $timestamp) > TOLERANCE_SEC) {
        throw new RuntimeException('Timestamp outside tolerance');
    }

    $key = base64_decode(preg_replace('/^whsec_/', '', $secret));
    $expected = base64_encode(hash_hmac('sha256', "{$id}.{$timestamp}.{$rawBody}", $key, true));

    foreach (explode(' ', $signatureHeader) as $part) {
        [$version, $signature] = array_pad(explode(',', $part, 2), 2, '');
        if ($version === 'v1' && hash_equals($expected, $signature)) {
            return json_decode($rawBody, true);
        }
    }
    throw new RuntimeException('Invalid signature');
}