Webhook Payload Signing Best Practices (2026): HMAC, Ephemeral Keys and Replay Protection

// keys: loaded from your secret manager, refreshed on rotation
const keys = new Map([
  ["k_2026_09_26", { secret: process.env.WH_KEY_CURRENT, expiresAt: null }],
  ["k_2026_09_19", { secret: process.env.WH_KEY_PREVIOUS, expiresAt: 1790467200 }],
]);

function keyFor(kid: string): string | null {
  const key = keys.get(kid);
  if (!key) return null;                              // unknown kid: reject
  if (key.expiresAt && Date.now() / 1000 > key.expiresAt) return null; // retired
  return key.secret;
}

// const secret = keyFor(req.headers["x-webhook-key-id"]);
// if (!secret) return res.status(401).end();
// then run verify(rawBody, req.headers, secret, seenNonces) as below

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. Constant-time signature comparison
    signing_string = f"{timestamp}.{nonce}.".encode() + payload
    expected = hmac.new(secret, signing_string, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, received):
        return False

    # 3. Reject replays inside the window. Record the nonce only after the
    #    signature is valid, so forged requests cannot poison the cache.
    if nonce in seen_nonces:
        return False
    seen_nonces.add(nonce, ttl=MAX_AGE_SECONDS)
    return True

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;

  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 || !crypto.timingSafeEqual(a, b)) return false;

  // Record the nonce only after the signature is valid
  if (seenNonces.has(nonce)) return false;
  seenNonces.add(nonce, MAX_AGE_SECONDS);
  return true;
}

{
  "endpoint": "/webhooks/stripe",
  "ip_allowlist": {
    "enabled": true,
    "source": "https://stripe.com/files/ips/ips_webhooks.txt",
    "refresh_every": "24h",
    "on_fetch_failure": "keep_last_known_list"
  },
  "signature": {
    "scheme": "hmac-sha256",
    "header": "Stripe-Signature",
    "signed_content": "{timestamp}.{raw_body}",
    "tolerance_seconds": 300,
    "keys": [
      { "kid": "current",  "secret_ref": "vault:webhooks/stripe/current" },
      { "kid": "previous", "secret_ref": "vault:webhooks/stripe/previous", "expires_at": "2026-09-27T12:00:00Z" }
    ]
  },
  "replay": { "dedupe_on": "body.id", "ttl_seconds": 604800 },
  "payload": { "max_bytes": 1048576, "json_schema": "schemas/stripe-event.json" }
}

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"