Verify webhook signatures

This page shows you how to confirm that a webhook push came from Wingspan and wasn't changed on the way, before your code acts on it.

Anyone who learns your endpoint URL can send it a request. The signature is how you tell a real Wingspan push from a forged or replayed one. Verify every request, and reject any that fail.

How the signature works

Every push carries a Wingspan-Signature header:

Wingspan-Signature: t=1790262602,v1=3b1f0c9a...e4d2

Part	Meaning
`t`	The Unix time in seconds when we signed this attempt. Each retry gets a new `t`.
`v1`	A hex-encoded HMAC-SHA256. There can be more than one `v1`.

Each v1 is computed as:

HMAC-SHA256(key = your subscription secret, message = t + "." + raw request body bytes)

Use the secret value exactly as it was returned when you created the subscription or rotated its secret.

For 72 hours after you rotate a secret without immediate: true, we send two v1 values: one for the new secret and one for the previous secret. The request is valid if any v1 matches a secret you hold.

Verification rules

Use the raw body. Compute the HMAC over the exact bytes you received, before any JSON parsing. Parsing and re-serializing changes whitespace and key order, and the signature will not match.
Check the timestamp. Reject the request if t is more than 300 seconds before or after your server's current time. This stops an attacker from replaying an old request. Keep your server clock in sync (for example, with NTP).
Compare in constant time. Use crypto.timingSafeEqual in Node.js or hmac.compare_digest in Python, never === or ==.
Accept any matching v1. Check every v1 in the header.
Verify before you parse. Only parse the JSON after the signature passes.

If verification fails, return 401 and don't process the body.

Node.js

This uses only the built-in crypto module.

const crypto = require('crypto');

const TOLERANCE_SECONDS = 300;

function parseSignatureHeader(header) {
  let timestamp = null;
  const signatures = [];
  for (const part of header.split(',')) {
    const index = part.indexOf('=');
    if (index === -1) continue;
    const key = part.slice(0, index).trim();
    const value = part.slice(index + 1).trim();
    if (key === 't') timestamp = value;
    if (key === 'v1') signatures.push(value);
  }
  return { timestamp, signatures };
}

function verifyWingspanSignature(rawBody, header, secret, nowSeconds = Math.floor(Date.now() / 1000)) {
  if (typeof header !== 'string' || header.length === 0) return false;
  const { timestamp, signatures } = parseSignatureHeader(header);
  if (!timestamp || !/^\d+$/.test(timestamp) || signatures.length === 0) return false;
  if (Math.abs(nowSeconds - Number(timestamp)) > TOLERANCE_SECONDS) return false;

  const expected = crypto
    .createHmac('sha256', secret)
    .update(`${timestamp}.`, 'utf8')
    .update(rawBody)
    .digest();

  // Check every candidate so the time taken doesn't depend on which one matched.
  let matched = false;
  for (const signature of signatures) {
    if (!/^[0-9a-fA-F]{64}$/.test(signature)) continue;
    const received = Buffer.from(signature, 'hex');
    if (crypto.timingSafeEqual(received, expected)) matched = true;
  }
  return matched;
}

module.exports = { verifyWingspanSignature };

Express: get the raw body

express.json() replaces the body with a parsed object, so the bytes are gone. Use express.raw() on the webhook route instead, and register the route before any global express.json().

const express = require('express');
const { verifyWingspanSignature } = require('./verifyWingspanSignature');

const app = express();
const secrets = [process.env.WINGSPAN_WEBHOOK_SECRET, process.env.WINGSPAN_WEBHOOK_SECRET_PREVIOUS].filter(Boolean);

app.post('/wingspan', express.raw({ type: '*/*' }), (req, res) => {
  const header = req.get('Wingspan-Signature');
  const valid = secrets.some((secret) => verifyWingspanSignature(req.body, header, secret));
  if (!valid) return res.sendStatus(401);

  const event = JSON.parse(req.body.toString('utf8'));
  enqueue(event); // your queue; do the real work outside this request
  res.sendStatus(200);
});

app.use(express.json()); // other routes can parse JSON as usual

Holding both the current and previous secret lets you switch secrets during a rotation without downtime. Remove the previous one after the 72-hour grace window.

Python

This uses only the standard library.

import hashlib
import hmac
import re
import time
from typing import Optional

TOLERANCE_SECONDS = 300
_HEX_SIGNATURE = re.compile(r"^[0-9a-fA-F]{64}$")


def verify_wingspan_signature(raw_body: bytes, header: Optional[str], secret: str, now: Optional[int] = None) -> bool:
    if not header:
        return False

    timestamp = None
    signatures = []
    for part in header.split(","):
        key, sep, value = part.strip().partition("=")
        if not sep:
            continue
        if key == "t":
            timestamp = value
        elif key == "v1":
            signatures.append(value)

    if timestamp is None or not timestamp.isdigit() or not signatures:
        return False

    now = int(time.time()) if now is None else now
    if abs(now - int(timestamp)) > TOLERANCE_SECONDS:
        return False

    expected = hmac.new(
        secret.encode("utf-8"),
        f"{timestamp}.".encode("ascii") + raw_body,
        hashlib.sha256,
    ).digest()

    # Check every candidate so the time taken doesn't depend on which one matched.
    matched = False
    for signature in signatures:
        if not _HEX_SIGNATURE.match(signature):
            continue
        if hmac.compare_digest(bytes.fromhex(signature), expected):
            matched = True
    return matched

Flask: get the raw body

Call request.get_data() before anything reads request.json, and verify against those bytes.

import json
import os

from flask import Flask, request

from verify import verify_wingspan_signature

app = Flask(__name__)
SECRETS = [s for s in (os.environ.get("WINGSPAN_WEBHOOK_SECRET"), os.environ.get("WINGSPAN_WEBHOOK_SECRET_PREVIOUS")) if s]


@app.post("/wingspan")
def wingspan_webhook():
    raw_body = request.get_data()
    header = request.headers.get("Wingspan-Signature")
    if not any(verify_wingspan_signature(raw_body, header, secret) for secret in SECRETS):
        return "", 401

    event = json.loads(raw_body)
    enqueue(event)  # your queue; do the real work outside this request
    return "", 200

In Django, use request.body. In FastAPI or Starlette, use await request.body().

Test your verifier

Before you go live, run your function against a request you sign yourself:

SECRET='whsec_test_only'
BODY='{"id":"abc","type":"Invoice.Paid"}'
T=$(date +%s)
SIG=$(printf '%s' "$T.$BODY" | openssl dgst -sha256 -hmac "$SECRET" -hex | sed 's/^.* //')
curl -X POST http://localhost:3000/wingspan \
  -H "Wingspan-Signature: t=$T,v1=$SIG" \
  --data-raw "$BODY"

Then check the failures: change one character of the body, set t to 10 minutes ago, and use the wrong secret. Each should return 401.

Common mistakes

Verifying a parsed-then-stringified body. The bytes differ from what we signed. Always use the raw bytes.
Checking only the first v1. During a rotation, the first value may belong to the secret you haven't deployed yet.
Skipping the timestamp check. Without it, a captured request can be replayed indefinitely.
Doing slow work before responding. Verification is fast. Everything after it belongs in a queue. See Delivery and retries.