Verifying signatures
Anyone who knows your address can send it a request. A delivery is from Quatlas only when its signature checks out with your subscription’s signing secret.
The scheme
Section titled “The scheme”- Read
Quatlas-Timestamp(Unix seconds) andQuatlas-Signature. - Refuse the delivery if the timestamp is more than 5 minutes from your clock. This stops an old delivery being replayed.
- Compute the HMAC-SHA256 of
<timestamp>.<raw body>: the timestamp header’s value, a dot, then the body’s bytes exactly as you received them. The key is your signing secret (whsec_...) as UTF-8 bytes. Write the result as lowercase hex. Quatlas-Signatureholds one or more comma-separated values likev1=<hex>. Accept the delivery when anyv1=value equals yours, compared in constant time. There are two values while a secret rotation is in its grace period, one per valid secret.- Dedupe on
Quatlas-Event-Id: the same event can arrive more than once.
Each function below takes the raw body, the two headers and your secret, and returns whether the delivery is genuine. They are tested against deliveries signed exactly the way the platform signs them, including a rotation’s two signatures and the 5-minute window.
import crypto from "node:crypto";
const TOLERANCE_SECONDS = 300; // refuse deliveries signed more than 5 minutes from your clock
/** * Verifies a Quatlas webhook delivery. * rawBody: the request body as a Buffer, exactly as received (not re-serialised JSON). * headers: the request headers, lower-cased names (as Node gives them). * secret: the subscription's signing secret (whsec_...). */export function verifyQuatlasSignature(rawBody, headers, secret) { const timestamp = headers["quatlas-timestamp"]; const signature = headers["quatlas-signature"]; if (typeof timestamp !== "string" || typeof signature !== "string" || !/^\d+$/.test(timestamp)) return false; if (Math.abs(Date.now() / 1000 - Number(timestamp)) > TOLERANCE_SECONDS) return false;
const expected = Buffer.from( crypto.createHmac("sha256", secret).update(`${timestamp}.`).update(rawBody).digest("hex"), );
// One v1= value per valid secret (two during a secret rotation): accept when any one matches. return signature.split(",").some((part) => { const value = part.trim(); if (!value.startsWith("v1=")) return false; const candidate = Buffer.from(value.slice(3)); return candidate.length === expected.length && crypto.timingSafeEqual(candidate, expected); });}With Express, take the raw body for this route only:
import express from "express";import { verifyQuatlasSignature } from "./verify.mjs";
const app = express();app.post("/quatlas/webhooks", express.raw({ type: "application/json" }), (req, res) => { if (!verifyQuatlasSignature(req.body, req.headers, process.env.QUATLAS_WEBHOOK_SECRET)) { return res.sendStatus(400); } const event = JSON.parse(req.body.toString("utf8")); // Skip event.id if you have processed it already; queue the work, then answer. res.sendStatus(204);});import hashlibimport hmacimport time
TOLERANCE_SECONDS = 300 # refuse deliveries signed more than 5 minutes from your clock
def verify_quatlas_signature(raw_body: bytes, timestamp: str | None, signature: str | None, secret: str) -> bool: """Verifies a Quatlas webhook delivery.
raw_body: the request body as bytes, exactly as received (not re-serialised JSON). timestamp, signature: the Quatlas-Timestamp and Quatlas-Signature headers. secret: the subscription's signing secret (whsec_...). """ if not timestamp or not signature or not (timestamp.isascii() and timestamp.isdigit()): return False if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS: return False
expected = hmac.new(secret.encode("utf-8"), timestamp.encode("ascii") + b"." + raw_body, hashlib.sha256).hexdigest()
# One v1= value per valid secret (two during a secret rotation): accept when any one matches. for part in signature.split(","): part = part.strip() if part.startswith("v1=") and hmac.compare_digest(part[3:], expected): return True return FalseWith Flask:
import json, osfrom flask import Flask, requestfrom verify import verify_quatlas_signature
app = Flask(__name__)
@app.post("/quatlas/webhooks")def quatlas_webhook(): raw = request.get_data() # the raw bytes, before any parsing if not verify_quatlas_signature(raw, request.headers.get("Quatlas-Timestamp"), request.headers.get("Quatlas-Signature"), os.environ["QUATLAS_WEBHOOK_SECRET"]): return "", 400 event = json.loads(raw) # Skip event["id"] if you have processed it already; queue the work, then answer. return "", 204using System;using System.Globalization;using System.Security.Cryptography;using System.Text;
public static class QuatlasWebhook{ // Refuse deliveries signed more than 5 minutes from your clock. private const long ToleranceSeconds = 300;
/// <summary>Verifies a Quatlas webhook delivery.</summary> /// <param name="rawBody">The request body, exactly as received (not re-serialised JSON).</param> /// <param name="timestamp">The Quatlas-Timestamp header.</param> /// <param name="signature">The Quatlas-Signature header.</param> /// <param name="secret">The subscription's signing secret (whsec_...).</param> public static bool Verify(byte[] rawBody, string? timestamp, string? signature, string secret) { if (string.IsNullOrEmpty(signature) || !long.TryParse(timestamp, NumberStyles.None, CultureInfo.InvariantCulture, out var seconds)) return false; if (Math.Abs(DateTimeOffset.UtcNow.ToUnixTimeSeconds() - seconds) > ToleranceSeconds) return false;
var prefix = Encoding.UTF8.GetBytes(timestamp + "."); var signed = new byte[prefix.Length + rawBody.Length]; prefix.CopyTo(signed, 0); rawBody.CopyTo(signed, prefix.Length); var hash = HMACSHA256.HashData(Encoding.UTF8.GetBytes(secret), signed); var expected = Encoding.ASCII.GetBytes(Convert.ToHexString(hash).ToLowerInvariant());
// One v1= value per valid secret (two during a secret rotation): accept when any one matches. foreach (var part in signature.Split(',', StringSplitOptions.TrimEntries | StringSplitOptions.RemoveEmptyEntries)) { if (!part.StartsWith("v1=", StringComparison.Ordinal)) continue; if (CryptographicOperations.FixedTimeEquals(Encoding.ASCII.GetBytes(part[3..]), expected)) return true; } return false; }}With ASP.NET Core minimal APIs:
app.MapPost("/quatlas/webhooks", async (HttpRequest request, IConfiguration config) =>{ using var buffer = new MemoryStream(); await request.Body.CopyToAsync(buffer); var body = buffer.ToArray();
if (!QuatlasWebhook.Verify(body, request.Headers["Quatlas-Timestamp"], request.Headers["Quatlas-Signature"], config["Quatlas:WebhookSecret"]!)) return Results.BadRequest();
var evt = JsonDocument.Parse(body).RootElement; // Skip evt.GetProperty("id") if you have processed it already; queue the work, then answer. return Results.NoContent();});import java.nio.charset.StandardCharsets;import java.security.GeneralSecurityException;import java.security.MessageDigest;import java.time.Instant;import java.util.HexFormat;import javax.crypto.Mac;import javax.crypto.spec.SecretKeySpec;
public final class QuatlasWebhook { // Refuse deliveries signed more than 5 minutes from your clock. private static final long TOLERANCE_SECONDS = 300;
private QuatlasWebhook() {}
/** * Verifies a Quatlas webhook delivery (Java 17 or later). * * @param rawBody the request body, exactly as received (not re-serialised JSON) * @param timestamp the Quatlas-Timestamp header * @param signature the Quatlas-Signature header * @param secret the subscription's signing secret (whsec_...) */ public static boolean verify(byte[] rawBody, String timestamp, String signature, String secret) { if (timestamp == null || signature == null || !timestamp.matches("[0-9]{1,18}")) return false; if (Math.abs(Instant.now().getEpochSecond() - Long.parseLong(timestamp)) > TOLERANCE_SECONDS) return false;
byte[] expected; try { Mac mac = Mac.getInstance("HmacSHA256"); mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256")); mac.update((timestamp + ".").getBytes(StandardCharsets.UTF_8)); expected = HexFormat.of().formatHex(mac.doFinal(rawBody)).getBytes(StandardCharsets.US_ASCII); } catch (GeneralSecurityException e) { throw new IllegalStateException(e); }
// One v1= value per valid secret (two during a secret rotation): accept when any one matches. for (String part : signature.split(",")) { String value = part.trim(); if (value.startsWith("v1=") && MessageDigest.isEqual(value.substring(3).getBytes(StandardCharsets.US_ASCII), expected)) { return true; } } return false; }}With Spring Web, take the body as bytes:
@PostMapping("/quatlas/webhooks")public ResponseEntity<Void> quatlasWebhook( @RequestBody byte[] body, @RequestHeader(value = "Quatlas-Timestamp", required = false) String timestamp, @RequestHeader(value = "Quatlas-Signature", required = false) String signature) { if (!QuatlasWebhook.verify(body, timestamp, signature, secret)) { return ResponseEntity.badRequest().build(); } // Parse the JSON, skip the event id if you have processed it already, queue the work, then answer. return ResponseEntity.noContent().build();}A worked example
Section titled “A worked example”With the secret whsec_test, the timestamp 1791963721 and the body {"id":"evt"}, the string to sign is:
1791963721.{"id":"evt"}and the header is:
Quatlas-Signature: v1=878fa619bb541a51b45e860b0d129b198853c2768f6a15d2d020a9f1e5b2b459Reproduce it with OpenSSL to check your own code. (A full check would still refuse this fixed example by its timestamp, unless your clock is within 5 minutes of it.)
printf '%s' '1791963721.{"id":"evt"}' | openssl dgst -sha256 -hmac 'whsec_test'# SHA2-256(stdin)= 878fa619bb541a51b45e860b0d129b198853c2768f6a15d2d020a9f1e5b2b459When the check fails
Section titled “When the check fails”Answer with a 4xx and do not process the delivery. Quatlas treats any answer other than 2xx as a failure and retries it on its schedule, so a receiver that keeps refusing genuine deliveries (for example, after a secret was rotated and not deployed) will see its subscription switched off after 20 failed attempts in a row. See Webhooks.
Check, in order: you sign the raw bytes; you use the current secret (or the previous one, during a rotation); your clock is right (use NTP); you compare lowercase hex.
