Verifying signatures

Verifying signatures

Check that a delivery really came from Heizen before you trust it.

Anyone can post to a public URL. Verify each delivery's signature so only Heizen's requests get through. Heizen signs in the Standard Webhooks format, so any Standard Webhooks library works too.

How the signature works

  1. Take the webhook-id, webhook-timestamp and the raw request body, exactly as received.
  2. Join them with dots: {webhook-id}.{webhook-timestamp}.{body}.
  3. Strip whsec_ from your secret and base64-decode the rest. That is the HMAC key.
  4. Compute HMAC-SHA256 of the joined string and base64-encode it.
  5. webhook-signature holds one or more space-separated v1,<signature> values. Accept the request if any of them matches yours, using a constant-time comparison.
  6. Reject the request if webhook-timestamp is more than five minutes from your clock. This stops replayed requests.

Always verify against the raw body. Parsing the JSON and re-serialising it changes the bytes and the signature will not match.

Node.js with the SDK

import express from "express";
import { verifyWebhook, WebhookVerificationError } from "@heizen/interviewer";

const app = express();

app.post("/webhooks/heizen", express.raw({ type: "application/json" }), async (req, res) => {
	let event;
	try {
		event = await verifyWebhook(req.body, req.headers, process.env.HEIZEN_WEBHOOK_SECRET!);
	} catch (err) {
		if (err instanceof WebhookVerificationError) return res.sendStatus(400);
		throw err;
	}
	res.sendStatus(204);
	if (event.type === "result.ready") await queueResultSync(event.data.object.id);
});

Node.js without dependencies

import { createHmac, timingSafeEqual } from "node:crypto";

export function verify(rawBody, headers, secret) {
	const id = headers["webhook-id"];
	const timestamp = headers["webhook-timestamp"];
	const signatures = headers["webhook-signature"];
	if (!id || !timestamp || !signatures) throw new Error("missing headers");
	if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) throw new Error("stale");

	const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");
	const expected = createHmac("sha256", key)
		.update(`${id}.${timestamp}.${rawBody}`)
		.digest("base64");
	const ok = signatures.split(" ").some((sig) => {
		const value = sig.split(",")[1] ?? "";
		return value.length === expected.length &&
			timingSafeEqual(Buffer.from(value), Buffer.from(expected));
	});
	if (!ok) throw new Error("bad signature");
	return JSON.parse(rawBody);
}

Python

import base64, hashlib, hmac, json, os, time

def verify(raw_body: bytes, headers, secret: str) -> dict:
    msg_id = headers["webhook-id"]
    timestamp = headers["webhook-timestamp"]
    signatures = headers["webhook-signature"]
    if abs(time.time() - int(timestamp)) > 300:
        raise ValueError("stale")

    key = base64.b64decode(secret.removeprefix("whsec_"))
    signed = f"{msg_id}.{timestamp}.".encode() + raw_body
    expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode()

    if not any(
        hmac.compare_digest(sig.split(",", 1)[-1], expected)
        for sig in signatures.split(" ")
    ):
        raise ValueError("bad signature")
    return json.loads(raw_body)

# Flask: verify(request.get_data(), request.headers, os.environ["HEIZEN_WEBHOOK_SECRET"])

Check your implementation

With the secret whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw, webhook-id msg_p5jXN8AQM9LWM0D4loKWxJek, timestamp 1614265330 and body {"test": 2432232314}, the signature is:

v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE=

This is a published test vector, not a real secret.

Rotating the secret

Rotate from the dashboard or with rotate_secret. The old secret stops working immediately, and deliveries signed with it that are still retrying will fail your check. Deploy the new secret right away; failed deliveries retry and will then pass.