Webhook Security Guide: HMAC Signatures, Replay Protection, and IP Allowlists

import hmac
import hashlib
import secrets
import time

MAX_AGE_SECONDS = 300  # reject anything older than 5 minutes

def sign(payload: bytes, secret: bytes) -> dict:
    timestamp = str(int(time.time()))
    nonce = secrets.token_hex(16)
    signing_string = f"{timestamp}.{nonce}.".encode() + payload
    mac = hmac.new(secret, signing_string, hashlib.sha256).hexdigest()
    return {
        "X-Webhook-Timestamp": timestamp,
        "X-Webhook-Nonce": nonce,
        "X-Webhook-Signature": f"sha256={mac}",
    }

def verify(payload: bytes, headers: dict, secret: bytes, seen_nonces) -> bool:
    try:
        timestamp = int(headers["X-Webhook-Timestamp"])
        nonce = headers["X-Webhook-Nonce"]
        received = headers["X-Webhook-Signature"].split("=", 1)[1]
    except (KeyError, ValueError):
        return False

    # 1. Reject stale requests (replay window)
    if abs(time.time() - timestamp) > MAX_AGE_SECONDS:
        return False

    # 2. Reject replays inside the window
    if nonce in seen_nonces:
        return False
    seen_nonces.add(nonce, ttl=MAX_AGE_SECONDS)

    # 3. Constant-time signature comparison
    signing_string = f"{timestamp}.{nonce}.".encode() + payload
    expected = hmac.new(secret, signing_string, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, received)

import crypto from "node:crypto";

const MAX_AGE_SECONDS = 300;

export function sign(payload: Buffer, secret: string) {
  const timestamp = Math.floor(Date.now() / 1000).toString();
  const nonce = crypto.randomBytes(16).toString("hex");
  const signingString = Buffer.concat([
    Buffer.from(`${timestamp}.${nonce}.`),
    payload,
  ]);
  const mac = crypto
    .createHmac("sha256", secret)
    .update(signingString)
    .digest("hex");
  return {
    "x-webhook-timestamp": timestamp,
    "x-webhook-nonce": nonce,
    "x-webhook-signature": `sha256=${mac}`,
  };
}

export function verify(
  payload: Buffer,
  headers: Record<string, string>,
  secret: string,
  seenNonces: { has(n: string): boolean; add(n: string, ttl: number): void },
): boolean {
  const timestamp = Number(headers["x-webhook-timestamp"]);
  const nonce = headers["x-webhook-nonce"];
  const received = (headers["x-webhook-signature"] ?? "").split("=", 2)[1];
  if (!timestamp || !nonce || !received) return false;

  // Replay window
  if (Math.abs(Date.now() / 1000 - timestamp) > MAX_AGE_SECONDS) return false;
  if (seenNonces.has(nonce)) return false;
  seenNonces.add(nonce, MAX_AGE_SECONDS);

  const signingString = Buffer.concat([
    Buffer.from(`${timestamp}.${nonce}.`),
    payload,
  ]);
  const expected = crypto
    .createHmac("sha256", secret)
    .update(signingString)
    .digest("hex");

  const a = Buffer.from(expected, "hex");
  const b = Buffer.from(received, "hex");
  if (a.length !== b.length) return false;
  return crypto.timingSafeEqual(a, b);
}

from ipaddress import ip_address, ip_network

ALLOWED_CIDRS = [ip_network(c) for c in load_provider_cidrs()]  # refreshed by cron

def webhook_middleware(request, secret, seen_nonces):
    # 1. L4 filter: reject if source IP is outside the provider allow-list
    src = ip_address(request.client_ip)
    if not any(src in net for net in ALLOWED_CIDRS):
        return 403, "ip_not_allowed"

    # 2. L7 auth: HMAC signature is still the source of truth
    if not verify(request.body, request.headers, secret, seen_nonces):
        return 401, "invalid_signature"

    return 200, "ok"