Every event, signed and retried.

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

// The endpoint's signing secret is "whsec_" + base64. The HMAC key is the decoded part.
const key = Buffer.from(process.env.REFIREMAIL_WEBHOOK_SECRET!.replace('whsec_', ''), 'base64');

export function verifySignature(headers: Headers, body: string): boolean {
  const id = headers.get('webhook-id') ?? '';
  const ts = headers.get('webhook-timestamp') ?? '';
  if (!/^\d+$/.test(ts) || Math.abs(Date.now() / 1000 - Number(ts)) > 300) return false;
  const expected = createHmac('sha256', key).update(`${id}.${ts}.${body}`).digest();
  // "v1,<sig>", space-separated: two of them for 24 hours after a secret rotation.
  return (headers.get('webhook-signature') ?? '').split(' ').some((entry) => {
    const got = Buffer.from(entry.split(',')[1] ?? '', 'base64');
    return got.length === expected.length && timingSafeEqual(got, expected);
  });
}

export async function POST(req: Request) {
  const body = await req.text(); // the raw body, before any JSON parsing
  if (!verifySignature(req.headers, body)) return new Response('Invalid signature', { status: 400 });

  const event = JSON.parse(body);
  if (event.type === 'email.bounced') {
    // We have already suppressed the address. Update your own records too.
    await flagAddress(event.data.to);
  }
  return new Response(null, { status: 204 });
}

pip install flask

import base64, hashlib, hmac, json, os, time
from flask import Flask, request

app = Flask(__name__)
# The endpoint's signing secret is "whsec_" + base64. The HMAC key is the decoded part.
KEY = base64.b64decode(os.environ["REFIREMAIL_WEBHOOK_SECRET"].removeprefix("whsec_"))


def verified(headers, body: bytes) -> bool:
    msg_id, ts = headers.get("webhook-id", ""), headers.get("webhook-timestamp", "")
    if not ts.isdigit() or abs(time.time() - int(ts)) > 300:
        return False
    mac = hmac.new(KEY, f"{msg_id}.{ts}.".encode() + body, hashlib.sha256).digest()
    expected = base64.b64encode(mac).decode()
    # "v1,<sig>", space-separated: two of them for 24 hours after a secret rotation.
    sigs = [s.partition(",")[2] for s in headers.get("webhook-signature", "").split()]
    return any(hmac.compare_digest(s, expected) for s in sigs)


@app.post("/hooks/email")
def email_events():
    body = request.get_data()  # the raw bytes, before any JSON parsing
    if not verified(request.headers, body):
        return "Invalid signature", 400

    event = json.loads(body)
    if event["type"] == "email.bounced":
        # We have already suppressed the address. Update your own records too.
        flag_address(event["data"]["to"])
    return "", 204