Verify webhook signatures in Node.js, Python and PHP

// server.mjs — npm install express; run: node server.mjs
import crypto from "node:crypto";
import express from "express";

const app = express();
const secret = process.env.ANYTIME_WEBHOOK_SECRET; // whsec_…

// Read the raw body: the signature covers the exact bytes we sent.
app.post("/webhooks/anytime", express.raw({ type: "application/json" }), (req, res) => {
  const header = req.get("Anytime-Signature") ?? "";
  // t=<timestamp>, then one v1= per signing secret (two during a rotation).
  const pairs = header.split(",").map((part) => part.trim().split("="));
  const t = pairs.find(([key]) => key === "t")?.[1] ?? "";
  const signatures = pairs.filter(([key]) => key === "v1").map(([, value]) => value);
  const body = req.body.toString("utf8");

  const expected = Buffer.from(
    crypto.createHmac("sha256", secret).update(`${t}.${body}`).digest("hex"),
  );
  const signed = signatures.some(
    (signature) => signature.length === expected.length && crypto.timingSafeEqual(Buffer.from(signature), expected),
  );
  const fresh = Math.abs(Date.now() / 1000 - Number(t)) <= 300;
  if (!signed || !fresh) return res.status(400).send("Invalid signature");

  const event = JSON.parse(body);
  // The same event can arrive more than once: skip ids you've already handled.
  switch (event.type) {
    case "payment.succeeded":
      // Fulfil the order for event.data.id
      break;
    case "payment.refunded":
      // event.data.amountRefundedCents is the total refunded so far
      break;
  }
  res.sendStatus(200);
});

app.listen(3000);

// app/api/webhooks/anytime/route.ts (Next.js App Router)
import crypto from "node:crypto";

export const runtime = "nodejs";

export async function POST(request: Request) {
  const secret = process.env.ANYTIME_WEBHOOK_SECRET!; // whsec_…
  const body = await request.text(); // the raw body, exactly as sent
  const header = request.headers.get("Anytime-Signature") ?? "";
  // t=<timestamp>, then one v1= per signing secret (two while a secret is rotated).
  const pairs = header.split(",").map((part) => part.trim().split("="));
  const t = pairs.find(([key]) => key === "t")?.[1] ?? "";
  const signatures = pairs.filter(([key]) => key === "v1").map(([, value]) => value ?? "");

  const expected = Buffer.from(crypto.createHmac("sha256", secret).update(`${t}.${body}`).digest("hex"));
  const signed = signatures.some(
    (signature) => signature.length === expected.length && crypto.timingSafeEqual(Buffer.from(signature), expected),
  );
  const fresh = Math.abs(Date.now() / 1000 - Number(t)) <= 300;
  if (!signed || !fresh) return new Response("Invalid signature", { status: 400 });

  const event = JSON.parse(body);
  // Skip event.id if you've handled it before, then act on event.type.
  if (event.type === "payment.succeeded") {
    // Mark the order in event.data.metadata (or event.data.clientReference) paid.
  }
  return new Response(null, { status: 200 });
}

# app.py — pip install flask; run: flask --app app run
import hashlib
import hmac
import json
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["ANYTIME_WEBHOOK_SECRET"]  # whsec_…


@app.post("/webhooks/anytime")
def anytime_webhook():
    body = request.get_data()  # the raw bytes, exactly as sent
    header = request.headers.get("Anytime-Signature", "")
    # t=<timestamp>, then one v1= per signing secret (two while a secret is rotated).
    pairs = [part.strip().split("=", 1) for part in header.split(",") if "=" in part]
    timestamp = next((value for key, value in pairs if key == "t"), "")
    signatures = [value for key, value in pairs if key == "v1"]
    if not timestamp.isdigit() or abs(time.time() - int(timestamp)) > 300:
        abort(400)
    expected = hmac.new(SECRET.encode(), timestamp.encode() + b"." + body, hashlib.sha256).hexdigest()
    if not any(hmac.compare_digest(expected.encode(), signature.encode()) for signature in signatures):
        abort(400)

    event = json.loads(body)
    # Skip event["id"] if you've handled it before, then act on event["type"].
    if event["type"] == "payment.succeeded":
        pass  # Mark the order in event["data"]["metadata"] paid.
    return "", 200

# main.py — pip install fastapi uvicorn; run: uvicorn main:app
import hashlib
import hmac
import json
import os
import time

from fastapi import FastAPI, HTTPException, Request

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


@app.post("/webhooks/anytime")
async def anytime_webhook(request: Request):
    body = await request.body()  # the raw bytes: don't take a parsed model here
    header = request.headers.get("Anytime-Signature", "")
    # t=<timestamp>, then one v1= per signing secret (two while a secret is rotated).
    pairs = [part.strip().split("=", 1) for part in header.split(",") if "=" in part]
    timestamp = next((value for key, value in pairs if key == "t"), "")
    signatures = [value for key, value in pairs if key == "v1"]
    if not timestamp.isdigit() or abs(time.time() - int(timestamp)) > 300:
        raise HTTPException(status_code=400, detail="Invalid signature")
    expected = hmac.new(SECRET.encode(), timestamp.encode() + b"." + body, hashlib.sha256).hexdigest()
    if not any(hmac.compare_digest(expected.encode(), signature.encode()) for signature in signatures):
        raise HTTPException(status_code=400, detail="Invalid signature")

    event = json.loads(body)
    # Skip event["id"] if you've handled it before, then act on event["type"].
    return {"received": True}

<?php
// webhook.php — plain PHP 7.4+, no framework.
$secret = getenv('ANYTIME_WEBHOOK_SECRET'); // whsec_…
$payload = file_get_contents('php://input'); // the raw body, exactly as sent
$header = $_SERVER['HTTP_ANYTIME_SIGNATURE'] ?? '';

// t=<timestamp>, then one v1= per signing secret (two while a secret is rotated).
$timestamp = '';
$signatures = [];
foreach (explode(',', $header) as $part) {
    [$key, $value] = array_pad(explode('=', trim($part), 2), 2, '');
    if ($key === 't') {
        $timestamp = $value;
    } elseif ($key === 'v1') {
        $signatures[] = $value;
    }
}
$expected = hash_hmac('sha256', $timestamp . '.' . $payload, $secret);
$signed = false;
foreach ($signatures as $signature) {
    $signed = $signed || hash_equals($expected, $signature);
}

if (!ctype_digit($timestamp) || abs(time() - (int) $timestamp) > 300 || !$signed) {
    http_response_code(400);
    exit('Invalid signature');
}

$event = json_decode($payload, true);
// Skip $event['id'] if you've handled it before, then act on $event['type'].
if ($event['type'] === 'payment.succeeded') {
    // Mark the order in $event['data']['metadata'] paid.
}
http_response_code(200);

SECRET="whsec_…"   # your endpoint's signing secret
BODY='{"id":"evt_local_1","type":"webhook.test","created":"2026-10-24T12:00:00.000Z","data":{}}'
T=$(date +%s)
SIG=$(printf '%s.%s' "$T" "$BODY" | openssl dgst -sha256 -hmac "$SECRET" -hex | sed 's/^.* //')

curl -X POST http://localhost:3000/webhooks/anytime \
  -H "Content-Type: application/json" \
  -H "Anytime-Signature: t=$T,v1=$SIG" \
  -d "$BODY"