Verify webhook signatures (Node.js, Python, PHP)

Every webhook request is signed. Verify it before trusting the body — otherwise anyone who learns your URL could send you fake events.

How the signature works

• Header: X-Vatli-Signature: t=<unix seconds>,v1=<hex>
• v1 = HMAC-SHA256 of "{t}.{raw request body}", keyed with your endpoint’s signing secret (whsec_…).
• Compute it over the raw body. A parsed and re-serialised JSON object won’t match.
• Compare in constant time, and reject a t more than 5 minutes old (replay protection).

Node.js (Express)

const crypto = require('crypto');

function verifyVatliSignature(rawBody, header, secret, toleranceSeconds = 300) {
  const parts = Object.fromEntries(
    header.split(',').map((p) => {
      const i = p.indexOf('=');
      return [p.slice(0, i).trim(), p.slice(i + 1).trim()];
    })
  );
  const t = Number(parts.t);
  if (!Number.isFinite(t) || !parts.v1) return false;
  if (Math.abs(Date.now() / 1000 - t) > toleranceSeconds) return false;
  const expected = crypto.createHmac('sha256', secret).update(`${t}.${rawBody}`).digest('hex');
  const given = String(parts.v1).toLowerCase();
  return expected.length === given.length &&
    crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(given));
}

// express.raw keeps the body as bytes so the signature matches
app.post('/vatli', express.raw({ type: 'application/json' }), (req, res) => {
  const ok = verifyVatliSignature(req.body.toString('utf8'), req.get('X-Vatli-Signature') || '', process.env.VATLI_WEBHOOK_SECRET);
  if (!ok) return res.status(401).end();
  const event = JSON.parse(req.body);
  // Deduplicate on event.id — a retry or resend carries the same id.
  res.status(200).end();
});

Python (Flask)

import hashlib, hmac, os, time
from flask import Flask, request, abort

def verify_vatli_signature(raw_body: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
    parts = dict(p.strip().split("=", 1) for p in header.split(",") if "=" in p)
    try:
        t = int(parts["t"])
    except (KeyError, ValueError):
        return False
    if abs(time.time() - t) > tolerance:
        return False
    signed = f"{t}.".encode() + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, parts.get("v1", "").strip().lower())

app = Flask(__name__)

@app.post("/vatli")
def vatli():
    if not verify_vatli_signature(request.get_data(), request.headers.get("X-Vatli-Signature", ""), os.environ["VATLI_WEBHOOK_SECRET"]):
        abort(401)
    event = request.get_json()
    # Deduplicate on event["id"].
    return "", 200

PHP

<?php
function verify_vatli_signature(string $rawBody, string $header, string $secret, int $tolerance = 300): bool {
    $parts = [];
    foreach (explode(',', $header) as $pair) {
        $kv = explode('=', $pair, 2);
        if (count($kv) === 2) $parts[trim($kv[0])] = trim($kv[1]);
    }
    if (!isset($parts['t'], $parts['v1']) || !ctype_digit($parts['t'])) return false;
    $t = (int) $parts['t'];
    if (abs(time() - $t) > $tolerance) return false;
    $expected = hash_hmac('sha256', $t . '.' . $rawBody, $secret);
    return hash_equals($expected, strtolower($parts['v1']));
}

$raw = file_get_contents('php://input');   // never $_POST
if (!verify_vatli_signature($raw, $_SERVER['HTTP_X_VATLI_SIGNATURE'] ?? '', getenv('VATLI_WEBHOOK_SECRET'))) {
    http_response_code(401);
    exit;
}
$event = json_decode($raw, true);
// Deduplicate on $event['id'].
http_response_code(200);

Endpoint restrictions

Your URL must be https:// and resolve to a public address. localhost, private network ranges and cloud metadata addresses are refused.