Webhook Signing & Verification

import hashlib
import hmac
import time
from flask import request, abort

SIGNING_SECRET = "your_endpoint_signing_secret"
MAX_AGE_S = 300


def verify(raw_body: bytes, signature_header: str) -> bool:
    parts = dict(p.split("=", 1) for p in signature_header.split(","))
    ts, sig = parts.get("t"), parts.get("v1")
    if not ts or not sig:
        return False
    if abs(time.time() - int(ts)) > MAX_AGE_S:
        return False
    expected = hmac.new(
        SIGNING_SECRET.encode(), f"{ts}.{raw_body.decode()}".encode(), hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, sig)


@app.post("/trackly/webhooks")
def receive():
    if not verify(request.get_data(), request.headers.get("X-Trackly-Signature", "")):
        abort(401)
    event = request.get_json()
    # dedupe on event["event_id"], then process, then return 2xx quickly
    return "", 200

const crypto = require("crypto");

const SIGNING_SECRET = "your_endpoint_signing_secret";
const MAX_AGE_S = 300;

function verify(rawBody, signatureHeader) {
  const parts = Object.fromEntries(
    signatureHeader.split(",").map((p) => p.split("="))
  );
  const { t, v1 } = parts;
  if (!t || !v1) return false;
  if (Math.abs(Date.now() / 1000 - Number(t)) > MAX_AGE_S) return false;
  const expected = crypto
    .createHmac("sha256", SIGNING_SECRET)
    .update(`${t}.${rawBody}`)
    .digest("hex");
  const a = Buffer.from(expected);
  const b = Buffer.from(v1);
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

// Mount with the raw body available, e.g. express.raw({ type: "application/json" })
app.post("/trackly/webhooks", (req, res) => {
  if (!verify(req.body.toString(), req.get("X-Trackly-Signature") || "")) {
    return res.sendStatus(401);
  }
  const event = JSON.parse(req.body.toString());
  // dedupe on event.event_id, then process, then return 2xx quickly
  res.sendStatus(200);
});