Verify webhook signatures

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

// The whsec_… secret you were shown when you registered the endpoint, prefix included.
const SECRET = process.env.SURFAIS_WEBHOOK_SECRET;
if (!SECRET) throw new Error("Set SURFAIS_WEBHOOK_SECRET");

const TOLERANCE_SECONDS = 300;

function verify({ secret, signatureHeader, timestampHeader, rawBody, nowSeconds }) {
  if (!signatureHeader || !timestampHeader) return false;
  if (!/^\d{1,12}$/.test(timestampHeader)) return false;
  const ts = Number(timestampHeader);
  if (Math.abs(nowSeconds - ts) > TOLERANCE_SECONDS) return false;
  const prefix = "v1=";
  if (!signatureHeader.startsWith(prefix)) return false;
  const presented = signatureHeader.slice(prefix.length);
  if (!/^[0-9a-f]{64}$/.test(presented)) return false;
  const expected = createHmac("sha256", Buffer.from(secret, "utf8"))
    .update(`${ts}.`, "utf8")
    .update(rawBody)
    .digest("hex");
  return timingSafeEqual(Buffer.from(presented, "hex"), Buffer.from(expected, "hex"));
}

function handleEvent(rawBody) {
  // Parse only after the signature has been verified.
  const event = JSON.parse(rawBody.toString("utf8"));
  // Your processing goes here. In production, put the event on a durable queue.
  console.log(`received ${event.type} ${event.id}`);
}

const app = express();

// Do not put express.json() in front of this route: it replaces the raw bytes with a parsed object.
app.post("/webhooks/surfais", express.raw({ type: "application/json" }), (req, res) => {
  const ok =
    Buffer.isBuffer(req.body) &&
    verify({
      secret: SECRET,
      signatureHeader: req.get("X-Surfais-Signature"),
      timestampHeader: req.get("X-Surfais-Timestamp"),
      rawBody: req.body,
      nowSeconds: Math.floor(Date.now() / 1000),
    });
  if (!ok) {
    res.status(401).end();
    return;
  }

  // Acknowledge first, process afterwards.
  res.status(204).end();
  setImmediate(() => {
    try {
      handleEvent(req.body);
    } catch (err) {
      console.error("webhook processing failed", err);
    }
  });
});

app.listen(3000);

import hashlib
import hmac
import json
import os
import re
import threading
import time

from flask import Flask, request

# The whsec_… secret you were shown when you registered the endpoint, prefix included.
SECRET = os.environ["SURFAIS_WEBHOOK_SECRET"]

TOLERANCE_SECONDS = 300

app = Flask(__name__)


def verify(secret, signature_header, timestamp_header, raw_body, now_seconds):
    if not signature_header or not timestamp_header:
        return False
    if not re.fullmatch(r"[0-9]{1,12}", timestamp_header):
        return False
    ts = int(timestamp_header)
    if abs(now_seconds - ts) > TOLERANCE_SECONDS:
        return False
    prefix = "v1="
    if not signature_header.startswith(prefix):
        return False
    presented = signature_header[len(prefix):]
    if not re.fullmatch(r"[0-9a-f]{64}", presented):
        return False
    expected = hmac.new(
        secret.encode("utf-8"),
        f"{ts}.".encode("utf-8") + raw_body,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(bytes.fromhex(presented), bytes.fromhex(expected))


def handle_event(raw_body):
    # Parse only after the signature has been verified.
    event = json.loads(raw_body)
    # Your processing goes here. In production, put the event on a durable queue.
    print(f"received {event['type']} {event['id']}")


@app.route("/webhooks/surfais", methods=["POST"])
def surfais_webhook():
    raw_body = request.get_data()  # bytes, exactly as received
    ok = verify(
        SECRET,
        request.headers.get("X-Surfais-Signature"),
        request.headers.get("X-Surfais-Timestamp"),
        raw_body,
        int(time.time()),
    )
    if not ok:
        return "", 401

    # Acknowledge first, process afterwards.
    threading.Thread(target=handle_event, args=(raw_body,), daemon=True).start()
    return "", 204


if __name__ == "__main__":
    app.run(port=3000)