Webhook security best practices signatures, replay protection, rotation

// Express receiver with raw-body signature verification (HMAC + timestamp)
// npm i express
import crypto from "crypto";
import express from "express";

const app = express();

// Capture raw body bytes for signature verification.
app.post(
  "/webhook",
  express.raw({ type: "*/*", limit: "2mb" }),
  async (req, res) => {
    const secret = process.env.WEBHOOK_SECRET ?? "whsec_replace_me";
    const timestamp = req.header("x-webhook-timestamp") ?? "";
    const signature = req.header("x-webhook-signature") ?? "";

    const rawBody = Buffer.isBuffer(req.body) ? req.body : Buffer.from(req.body ?? "");
    const maxAgeSeconds = 300;
    const nowSeconds = Math.floor(Date.now() / 1000);

    if (!timestamp || !signature) return res.status(401).send("missing signature headers");
    if (Math.abs(nowSeconds - Number(timestamp)) > maxAgeSeconds) return res.status(401).send("stale timestamp");

    // Example scheme: HMAC_SHA256(secret, timestamp + "." + rawBody)
    const signed = Buffer.concat([Buffer.from(timestamp + ".", "utf8"), rawBody]);
    const expected = crypto.createHmac("sha256", secret).update(signed).digest("hex");

    const expectedBuf = Buffer.from(expected, "utf8");
    const providedBuf = Buffer.from(signature, "utf8");
    if (expectedBuf.length !== providedBuf.length) return res.status(401).send("bad signature");
    if (!crypto.timingSafeEqual(expectedBuf, providedBuf)) return res.status(401).send("bad signature");

    // At this point the request is authentic. Ack fast.
    res.sendStatus(202);

    // TODO: enqueue rawBody + selected headers for async processing
    // IMPORTANT: store a provider event ID / delivery ID to dedupe replays.
  }
);

app.listen(3000, () => console.log("listening on :3000"));

# FastAPI receiver with raw-body signature verification (HMAC + timestamp)
# pip install fastapi uvicorn
import hashlib
import hmac
import os
import time
from fastapi import FastAPI, Request, Response

app = FastAPI()

@app.post("/webhook")
async def webhook(request: Request) -> Response:
    secret = os.getenv("WEBHOOK_SECRET", "whsec_replace_me").encode("utf-8")
    timestamp = request.headers.get("x-webhook-timestamp", "")
    signature = request.headers.get("x-webhook-signature", "")
    raw_body = await request.body()

    if not timestamp or not signature:
        return Response("missing signature headers", status_code=401)

    max_age_seconds = 300
    now = int(time.time())
    if abs(now - int(timestamp)) > max_age_seconds:
        return Response("stale timestamp", status_code=401)

    signed = (timestamp + ".").encode("utf-8") + raw_body
    expected = hmac.new(secret, signed, hashlib.sha256).hexdigest()

    # Constant-time compare
    if not hmac.compare_digest(expected, signature):
        return Response("bad signature", status_code=401)

    # Authentic. Ack fast.
    # TODO: enqueue raw_body + selected headers for async processing (dedupe on event id/delivery id).
    return Response(status_code=202)