Signature Verification
HMAC-SHA256 verification of every delivery's ZG-Signature header, in a Stripe-compatible scheme, and how a rotation replaces the secret.
Every delivery carries a ZG-Signature: t=<unix>,v1=<hex> header. Verify it before acting on the payload, or anyone who guesses the endpoint's URL can forge events.
Scheme
ZG-Signature: t=<unix_ts>,v1=<hex(hmac_sha256(secret, "<t>.<raw_body>"))>There is exactly one signature per delivery, always v1=, and it is always made with the endpoint's current secret. Accept the request if that signature matches, within a five-minute timestamp tolerance.
The implementations below also accept a v0= value and a list of secrets.
Zuse Bifrost never sends v0=; that tolerance exists so you can hold two
secrets across a rotation, which is
necessary because rotation is instant and has no grace window.
Reference implementations
Node (Express)
secrets accepts a string or an array. Hold both the current and the next secret across a rotation so a single deploy rides through the instant cutover.
import crypto from "node:crypto";
export function verifySignature(
header: string,
rawBody: string,
secrets: string | string[],
toleranceSeconds = 300,
): boolean {
const parts = Object.fromEntries(
header.split(",").map((kv) => kv.split("=").map((s) => s.trim())),
);
if (!parts.t) return false;
const skew = Math.abs(Math.floor(Date.now() / 1000) - Number(parts.t));
if (skew > toleranceSeconds) return false;
const signed = `${parts.t}.${rawBody}`;
const secretList = Array.isArray(secrets) ? secrets : [secrets];
const candidates = [parts.v1, parts.v0].filter(Boolean) as string[];
for (const secret of secretList) {
const expected = crypto
.createHmac("sha256", secret)
.update(signed)
.digest("hex");
const expectedBuf = Buffer.from(expected, "hex");
for (const candidate of candidates) {
if (candidate.length !== expected.length) continue;
const candidateBuf = Buffer.from(candidate, "hex");
if (crypto.timingSafeEqual(candidateBuf, expectedBuf)) {
return true;
}
}
}
return false;
}PHP (Laravel)
Wire it as a route middleware so every webhook controller stays free of verification noise. Configure the secrets under services.zusegenius.webhook_secrets, an array that holds both the old and the new value during a rotation.
'zusegenius' => [
'webhook_secrets' => array_filter([
env('ZUSEGENIUS_WEBHOOK_SECRET'),
env('ZUSEGENIUS_WEBHOOK_SECRET_PREVIOUS'),
]),
],namespace App\Http\Middleware;
use Closure;
use Illuminate\Http\Request;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\Config;
use Illuminate\Support\Str;
use Symfony\Component\HttpFoundation\Response;
class VerifySignature
{
public function handle(Request $request, Closure $next, int $toleranceSeconds = 300): Response
{
$parts = Str::of($request->header('ZG-Signature', ''))
->explode(',')
->mapWithKeys(function (string $kv): array {
[$k, $v] = array_pad(explode('=', trim($kv), 2), 2, '');
return [trim($k) => trim($v)];
});
abort_if(blank($parts->get('t')), 400, 'missing timestamp');
abort_if(
abs(Carbon::now()->getTimestamp() - (int) $parts->get('t')) > $toleranceSeconds,
400,
'timestamp outside tolerance',
);
$signed = $parts->get('t').'.'.$request->getContent();
$candidates = collect([$parts->get('v1'), $parts->get('v0')])->filter();
$secrets = collect(Config::get('services.zusegenius.webhook_secrets', []))->filter();
$valid = $secrets->contains(
fn (string $secret): bool => $candidates->contains(
fn (string $candidate): bool => hash_equals(hash_hmac('sha256', $signed, $secret), $candidate),
),
);
abort_unless($valid, 400, 'bad signature');
return $next($request);
}
}Route::post('/webhooks/zusegenius', WebhookController::class)
->middleware(VerifySignature::class);Python (Flask)
import hashlib
import hmac
import time
from collections.abc import Iterable
def verify_signature(
header: str,
raw_body: bytes,
secrets: str | Iterable[str],
tolerance: int = 300,
) -> bool:
parts: dict[str, str] = {}
for kv in header.split(","):
k, _, v = kv.partition("=")
parts[k.strip()] = v.strip()
if "t" not in parts:
return False
if abs(int(time.time()) - int(parts["t"])) > tolerance:
return False
signed = f"{parts['t']}.{raw_body.decode()}".encode()
secret_list = [secrets] if isinstance(secrets, str) else list(secrets)
candidates = [c for c in (parts.get("v1"), parts.get("v0")) if c]
for secret in secret_list:
expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
for candidate in candidates:
if hmac.compare_digest(candidate, expected):
return True
return FalseCommon mistakes
- Parsing the body as JSON first. Canonicalisation differs between JSON libraries; compute the HMAC against the raw body bytes.
- Comparing with
==. Timing-safe comparison is required. Usecrypto.timingSafeEqual,hash_equalsorhmac.compare_digest. - Ignoring
t. Without the timestamp check, a captured payload can be replayed forever.
Rotating secrets
- Rotating from the endpoint's menu returns a fresh secret and replaces the old one immediately. There is no dual-signing and no grace window.
- Every delivery carries exactly one
v1=signature. The header never contains av0=value. - Because the cutover is instant, the safe order is: deploy a handler that accepts both the current and the next secret, then rotate, then drop the old value. The reference implementations above take an array of secrets for exactly this reason; the array is how you create the overlap, since the platform does not.
- Deliveries signed with the new secret while your handler still only knows the old one fail verification and are retried on the normal back-off schedule.
Do not store the secret beside the webhook URL in version control. Treat it with the same care as an API token. Once rotated, the old secret is unrecoverable.
How is this guide?
Last updated on