Webhook signature verification

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

/**
 * Whether a delivery really came from Backed. `header` is the Backed-Signature header,
 * `rawBody` the request body exactly as received (a Buffer or string, before any JSON parsing),
 * `secret` the endpoint's signing secret, whsec_ included.
 */
export function verifyBackedSignature(header, rawBody, secret, toleranceSeconds = 300) {
  let t = null;
  const signatures = [];
  for (const part of header.split(',')) {
    const [key, value] = part.trim().split('=', 2);
    if (key === 't' && /^[0-9]{1,12}$/.test(value)) t = Number(value);
    else if (key === 'v1' && /^[0-9a-f]{64}$/.test(value)) signatures.push(value);
  }
  if (t === null || signatures.length === 0) return false;
  if (Math.floor(Date.now() / 1000) - t > toleranceSeconds) return false;
  const expected = createHmac('sha256', secret).update(`${t}.`).update(rawBody).digest();
  return signatures.some((signature) => {
    const given = Buffer.from(signature, 'hex');
    return given.length === expected.length && timingSafeEqual(given, expected);
  });
}

/**
 * The route's handler. `enqueue(event)` is yours: store the event durably (a queue or a table,
 * de-duplicated on event.id, the Backed-Event-Id header) and resolve once it is stored.
 * Do the slow work from there, after answering, so the answer comes within 10 seconds.
 */
export function backedWebhookHandler(secret, enqueue) {
  return async (req, res) => {
    const header = req.get('Backed-Signature') ?? '';
    if (!verifyBackedSignature(header, req.body, secret)) return res.sendStatus(400);
    try {
      await enqueue(JSON.parse(req.body.toString('utf8')));
    } catch {
      return res.sendStatus(500); // not stored: a non-2xx answer makes Backed retry it
    }
    return res.sendStatus(204); // stored: a 2xx is final, Backed will not send it again
  };
}

// Express: read the raw body for this route, not parsed JSON.
const app = express();
app.post(
  '/backed/webhooks',
  express.raw({ type: 'application/json' }),
  backedWebhookHandler(process.env.BACKED_WEBHOOK_SECRET, enqueue),
);

import hashlib
import hmac
import os
import re
import time

from flask import Flask, request


def verify_backed_signature(header: str, raw_body: bytes, secret: str, tolerance: int = 300) -> bool:
    """Whether a delivery really came from Backed.

    header is the Backed-Signature header, raw_body the request body exactly as received,
    secret the endpoint's signing secret, whsec_ included.
    """
    t = None
    signatures = []
    for part in header.split(","):
        key, _, value = part.strip().partition("=")
        # ASCII only: str.isdigit() accepts "²", and a non-ASCII signature makes compare_digest raise.
        if key == "t" and re.fullmatch(r"[0-9]{1,12}", value):
            t = int(value)
        elif key == "v1" and re.fullmatch(r"[0-9a-f]{64}", value):
            signatures.append(value)
    if t is None or not signatures:
        return False
    if int(time.time()) - t > tolerance:
        return False
    signed = f"{t}.".encode() + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, signature) for signature in signatures)


# Flask: request.get_data() is the raw body.
app = Flask(__name__)


@app.post("/backed/webhooks")
def backed_webhook():
    header = request.headers.get("Backed-Signature", "")
    if not verify_backed_signature(header, request.get_data(), os.environ["BACKED_WEBHOOK_SECRET"]):
        return "", 400
    event = request.get_json()
    try:
        # Yours: store the event durably (a queue or a table, de-duplicated on event["id"]),
        # returning once it is stored. Do the slow work from there, after answering.
        enqueue(event)
    except Exception:
        return "", 500  # not stored: a non-2xx answer makes Backed retry it
    return "", 204  # stored: a 2xx is final, Backed will not send it again