Webhooks and signatures

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

export function verifyWebhook(secret, headers, rawBody, now = Date.now() / 1000) {
  const id = headers['webhook-id'];
  const timestamp = headers['webhook-timestamp'];
  const signatures = headers['webhook-signature'];
  if (!Buffer.isBuffer(rawBody) || typeof id !== 'string' || !id ||
      typeof timestamp !== 'string' || !/^\d+$/.test(timestamp) ||
      typeof signatures !== 'string' || !Number.isFinite(now)) {
    throw new Error('Invalid webhook input');
  }
  const seconds = Number(timestamp);
  if (!Number.isSafeInteger(seconds) || Math.abs(now - seconds) > 300) {
    throw new Error('Webhook timestamp outside tolerance');
  }
  if (typeof secret !== 'string' || !secret.startsWith('whsec_')) {
    throw new Error('Invalid signing secret');
  }
  const encoded = secret.slice(6);
  const key = Buffer.from(encoded, 'base64');
  if (!key.length || key.toString('base64') !== encoded) {
    throw new Error('Invalid signing secret');
  }
  const expected = createHmac('sha256', key)
    .update(`${id}.${timestamp}.`, 'utf8')
    .update(rawBody)
    .digest();
  const valid = signatures.split(/\s+/).some(part => {
    if (!/^v1,[A-Za-z0-9+/]+={0,2}$/.test(part)) return false;
    const supplied = Buffer.from(part.slice(3), 'base64');
    return supplied.length === expected.length && timingSafeEqual(supplied, expected);
  });
  if (!valid) throw new Error('Invalid webhook signature');
  const event = JSON.parse(rawBody.toString('utf8'));
  if (!event || Array.isArray(event) || event.id !== id) {
    throw new Error('Webhook id mismatch');
  }
  return event;
}

import base64
import hashlib
import hmac
import json
import math
import re
import time

def verify_webhook(secret, headers, raw_body, now=None):
    headers = {name.lower(): value for name, value in headers.items()}
    event_id = headers.get("webhook-id", "")
    timestamp = headers.get("webhook-timestamp", "")
    signatures = headers.get("webhook-signature", "")
    now = time.time() if now is None else now
    if (not isinstance(raw_body, bytes) or not isinstance(event_id, str)
            or not event_id or not isinstance(timestamp, str)
            or not re.fullmatch(r"[0-9]+", timestamp)
            or not isinstance(signatures, str) or not math.isfinite(now)):
        raise ValueError("Invalid webhook input")
    if abs(now - int(timestamp)) > 300:
        raise ValueError("Webhook timestamp outside tolerance")
    if not isinstance(secret, str) or not secret.startswith("whsec_"):
        raise ValueError("Invalid signing secret")
    key = base64.b64decode(secret[6:], validate=True)
    if not key:
        raise ValueError("Invalid signing secret")
    signed = f"{event_id}.{timestamp}.".encode("utf-8") + raw_body
    expected = hmac.new(key, signed, hashlib.sha256).digest()
    valid = False
    for part in signatures.split():
        if not part.startswith("v1,"):
            continue
        try:
            supplied = base64.b64decode(part[3:], validate=True)
        except ValueError:
            continue
        if hmac.compare_digest(supplied, expected):
            valid = True
    if not valid:
        raise ValueError("Invalid webhook signature")
    event = json.loads(raw_body)
    if not isinstance(event, dict) or event.get("id") != event_id:
        raise ValueError("Webhook id mismatch")
    return event