Verifying signatures

const crypto = require("crypto");
const express = require("express");

const SECRET = process.env.CONTOUR_WEBHOOK_SECRET; // whsec_...
const app = express();

// Keep the raw body: express.json() would parse and discard it.
app.post(
  "/webhooks/contour",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const header = req.get("X-Contour-Signature") || "";
    const parts = Object.fromEntries(header.split(",").map((p) => p.trim().split("=")));
    const t = parts.t;
    const v1 = parts.v1;

    const expected = crypto
      .createHmac("sha256", SECRET)
      .update(`${t}.`)
      .update(req.body) // Buffer: the raw bytes
      .digest("hex");

    const fresh = Math.abs(Date.now() / 1000 - Number(t)) < 300;
    const valid =
      v1 && v1.length === expected.length &&
      crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(v1));

    if (!fresh || !valid) return res.status(401).send("bad signature");

    const event = JSON.parse(req.body.toString("utf8"));
    // Respond first, then process (or hand off to a queue).
    res.sendStatus(200);
    handle(event); // event.event, event.call.call_id, event.call.metadata, ...
  }
);

import hmac, hashlib, os, time
from fastapi import FastAPI, Request, Response

SECRET = os.environ["CONTOUR_WEBHOOK_SECRET"]  # whsec_...
app = FastAPI()

@app.post("/webhooks/contour")
async def contour_webhook(request: Request):
    raw = await request.body()                      # bytes, not request.json()
    header = request.headers.get("X-Contour-Signature", "")
    parts = dict(p.strip().split("=", 1) for p in header.split(",") if "=" in p)
    t, v1 = parts.get("t", ""), parts.get("v1", "")

    expected = hmac.new(SECRET.encode(), f"{t}.".encode() + raw, hashlib.sha256).hexdigest()
    fresh = t.isdigit() and abs(time.time() - int(t)) < 300
    if not fresh or not hmac.compare_digest(expected, v1):
        return Response(status_code=401)

    event = await request.json()
    # Respond quickly; do the real work in a background task or queue.
    handle(event)
    return Response(status_code=200)