Verifying webhook signatures
Quorum signs deliveries exactly as the Standard Webhooks specification describes, so any compliant library works (standardwebhooks on npm, PyPI, and others; pass it the whsec_... secret). To do it by hand:
- Read the raw request body, byte for byte. Do not parse and re-serialise it first.
- Take
webhook-id,webhook-timestampandwebhook-signaturefrom the headers. Reject the request if the timestamp is more than 5 minutes from your clock; that stops replays. - The signed content is
{webhook-id}.{webhook-timestamp}.{body}. - The key is the secret without its
whsec_prefix, base64-decoded. - The signature is the base64 of the HMAC-SHA256 of the signed content.
- The header holds one or more
v1,{signature}entries separated by spaces (two while a secret is being rotated). Compare each with a constant-time comparison; accept if any matches.
The specification’s own example, which every sample below passes:
secret whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSwid msg_p5jXN8AQM9LWM0D4loKWxJektimestamp 1614265330body {"test": 2432232314}signature v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE=import { createHmac, timingSafeEqual } from "node:crypto";
export function verify( secret, headers, rawBody, { toleranceSeconds = 300, now = Date.now() } = {},) { const id = headers["webhook-id"]; const timestamp = headers["webhook-timestamp"]; const signatures = headers["webhook-signature"]; if (!id || !timestamp || !signatures) return false; if (!/^\d+$/.test(timestamp)) return false; if (Math.abs(Math.floor(now / 1000) - Number(timestamp)) > toleranceSeconds) return false;
const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64"); const expected = createHmac("sha256", key).update(`${id}.${timestamp}.${rawBody}`).digest();
return signatures.split(" ").some((entry) => { const [version, signature] = entry.split(","); if (version !== "v1" || !signature) return false; const given = Buffer.from(signature, "base64"); return given.length === expected.length && timingSafeEqual(given, expected); });}With Express, keep the raw body for the route (express.raw({ type: "application/json" })) and call verify(process.env.QUORUM_WEBHOOK_SECRET, req.headers, req.body.toString("utf8")). Return 401 when it is false and 204 quickly when it is true.
import base64, hashlib, hmac, time
def verify(secret, headers, raw_body, tolerance_seconds=300, now=None): msg_id = headers.get("webhook-id") timestamp = headers.get("webhook-timestamp") signatures = headers.get("webhook-signature") if not (msg_id and timestamp and signatures) or not timestamp.isdigit(): return False now = time.time() if now is None else now if abs(int(now) - int(timestamp)) > tolerance_seconds: return False
key = base64.b64decode(secret.removeprefix("whsec_")) signed = f"{msg_id}.{timestamp}.".encode() + raw_body # raw_body is bytes expected = hmac.new(key, signed, hashlib.sha256).digest()
for entry in signatures.split(" "): version, _, signature = entry.partition(",") if version == "v1" and hmac.compare_digest(base64.b64decode(signature), expected): return True return FalseIn Flask use request.get_data() for raw_body; in FastAPI await request.body().
package quorum
import ( "crypto/hmac" "crypto/sha256" "encoding/base64" "strconv" "strings" "time")
// Verify reports whether a delivery is genuine. rawBody is the request body, unparsed.func Verify(secret string, header func(string) string, rawBody []byte, now time.Time) bool { id, ts, sigs := header("webhook-id"), header("webhook-timestamp"), header("webhook-signature") if id == "" || ts == "" || sigs == "" { return false } sent, err := strconv.ParseInt(ts, 10, 64) if err != nil || now.Sub(time.Unix(sent, 0)).Abs() > 5*time.Minute { return false } key, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_")) if err != nil { return false } mac := hmac.New(sha256.New, key) mac.Write([]byte(id + "." + ts + ".")) mac.Write(rawBody) expected := mac.Sum(nil)
for _, entry := range strings.Split(sigs, " ") { version, sig, ok := strings.Cut(entry, ",") if !ok || version != "v1" { continue } if got, err := base64.StdEncoding.DecodeString(sig); err == nil && hmac.Equal(got, expected) { return true } } return false}Call it with r.Header.Get and io.ReadAll(r.Body).
# Prints the signature for a payload: useful to check your receiver by hand.secret="whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw"id="msg_p5jXN8AQM9LWM0D4loKWxJek"; ts=1614265330; body='{"test": 2432232314}'key=$(printf '%s' "${secret#whsec_}" | base64 -d | xxd -p -c 256)printf 'v1,%s\n' "$(printf '%s.%s.%s' "$id" "$ts" "$body" \ | openssl dgst -sha256 -mac HMAC -macopt hexkey:"$key" -binary | base64)"Then replay it at your receiver: curl -X POST "$URL" -H "webhook-id: $id" -H "webhook-timestamp: $ts" -H "webhook-signature: $sig" -d "$body". Or use the dashboard’s Send test event.
Common mistakes
Section titled “Common mistakes”- Parsing first. Re-serialising JSON changes whitespace and key order; sign over the raw bytes.
- Using the secret as text. The key is the base64-decoded secret without
whsec_. - Comparing with
==. Use a constant-time comparison. - Ignoring the timestamp. Without the 5-minute check a captured request can be replayed.
- Only knowing one secret during a rotation. Either secret may sign; the header carries both, so one match is enough.
- Treating delivery as exactly once. De-duplicate on
webhook-id.