# Verifying webhook signatures Every webhook request carries `X-Webhook-Signature: t=,v1=`. Check it before you trust a request: anyone can send a `POST` to your endpoint. ## How the signature works - `t` is when we sent this attempt, as Unix time in seconds. Each retry is signed again with a new `t`. - `v1` is the HMAC-SHA256 of the string `.`, keyed with the endpoint's signing secret (the whole `whsec_...` string, as UTF-8), written in lowercase hex. To verify a request: 1. Read the raw request body exactly as received, **before** parsing the JSON. JSON that you parse and serialise again won't match. 2. Split the header on `,` and `=` to get `t` and `v1`. 3. Reject the request if `t` is more than 5 minutes (300 seconds) away from your clock, so old requests can't be replayed. Keep your server clock in sync. 4. Compute the HMAC-SHA256 of `t`, a dot and the raw body with your secret, and compare it with `v1` in constant time. 5. If it doesn't match, answer `400` and ignore the request. ## Node.js ```js import { createHmac, timingSafeEqual } from "node:crypto"; // rawBody: the request body exactly as received (a string, before JSON parsing) // now: Unix time in seconds, optional; pass 1762653600 to check the sample request in the docs function isValidWebhook(secret, signatureHeader, rawBody, now = Date.now() / 1000) { const parts = new Map( signatureHeader.split(",").map((part) => { const index = part.indexOf("="); return index < 0 ? [part.trim(), ""] : [part.slice(0, index).trim(), part.slice(index + 1)]; }), ); const rawTimestamp = parts.get("t") ?? ""; const given = parts.get("v1") ?? ""; const timestamp = Number(rawTimestamp); if (!/^[0-9]+$/.test(rawTimestamp) || !Number.isSafeInteger(timestamp) || Math.abs(now - timestamp) > 300) return false; if (!/^[0-9a-f]{64}$/.test(given)) return false; const expected = createHmac("sha256", secret).update(`${timestamp}.${rawBody}`).digest(); return timingSafeEqual(Buffer.from(given, "hex"), expected); } ``` `rawBody` is the body as a string, exactly as received. ## Python ```python import hashlib import hmac import re import time # raw_body: the request body exactly as received (bytes, before JSON parsing) # now: Unix time in seconds, optional; pass 1762653600 to check the sample request in the docs def is_valid_webhook(secret, signature_header, raw_body, now=None): if now is None: now = time.time() parts = {} for part in signature_header.split(","): key, _, value = part.partition("=") parts[key.strip()] = value timestamp = parts.get("t", "") given = parts.get("v1", "") if not re.fullmatch(r"[0-9]{1,16}", timestamp) or abs(now - int(timestamp)) > 300: return False if not re.fullmatch(r"[0-9a-f]{64}", given): return False signed = str(int(timestamp)).encode("utf-8") + b"." + raw_body expected = hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, given) ``` `raw_body` is the body as bytes, exactly as received. ## Complete receivers These minimal servers verify the signature, answer at once and skip duplicates. They read the secret from `FULFILLBOT_WEBHOOK_SECRET`. With a framework, make sure you get the raw body: for example `express.raw({ type: "application/json" })` in Express, or `request.get_data()` in Flask. ### Node.js receiver ```js import { createServer } from "node:http"; import { createHmac, timingSafeEqual } from "node:crypto"; const SECRET = process.env.FULFILLBOT_WEBHOOK_SECRET; // Keep processed event ids in your database in production; a Set is lost on restart. const seenEventIds = new Set(); function isValidWebhook(secret, signatureHeader, rawBody, now = Date.now() / 1000) { const parts = new Map( signatureHeader.split(",").map((part) => { const index = part.indexOf("="); return index < 0 ? [part.trim(), ""] : [part.slice(0, index).trim(), part.slice(index + 1)]; }), ); const rawTimestamp = parts.get("t") ?? ""; const given = parts.get("v1") ?? ""; const timestamp = Number(rawTimestamp); if (!/^[0-9]+$/.test(rawTimestamp) || !Number.isSafeInteger(timestamp) || Math.abs(now - timestamp) > 300) return false; if (!/^[0-9a-f]{64}$/.test(given)) return false; const expected = createHmac("sha256", secret).update(`${timestamp}.${rawBody}`).digest(); return timingSafeEqual(Buffer.from(given, "hex"), expected); } // Set a body limit: reject oversized requests before they use your memory. const MAX_BODY_BYTES = 1_000_000; createServer((request, response) => { const chunks = []; let size = 0; request.on("data", (chunk) => { size += chunk.length; if (size <= MAX_BODY_BYTES) chunks.push(chunk); }); request.on("end", () => { if (size > MAX_BODY_BYTES) { response.writeHead(413).end(); return; } const rawBody = Buffer.concat(chunks).toString("utf8"); if (!isValidWebhook(SECRET, request.headers["x-webhook-signature"] ?? "", rawBody)) { response.writeHead(400).end(); return; } let event; try { event = JSON.parse(rawBody); } catch { response.writeHead(400).end(); return; } // Answer within the timeout first; do slow work (such as GET calls) after answering or in a queue. response.writeHead(200).end(); if (seenEventIds.has(event.id)) return; seenEventIds.add(event.id); console.log("Received", event.type, event.data.id, event.data.status); }); }).listen(8000); ``` ### Python receiver ```python import hashlib import hmac import re import time import json import os from http.server import BaseHTTPRequestHandler, HTTPServer SECRET = os.environ["FULFILLBOT_WEBHOOK_SECRET"] # Keep processed event ids in your database in production; a set is lost on restart. seen_event_ids = set() # raw_body: the request body exactly as received (bytes, before JSON parsing) # now: Unix time in seconds, optional; pass 1762653600 to check the sample request in the docs def is_valid_webhook(secret, signature_header, raw_body, now=None): if now is None: now = time.time() parts = {} for part in signature_header.split(","): key, _, value = part.partition("=") parts[key.strip()] = value timestamp = parts.get("t", "") given = parts.get("v1", "") if not re.fullmatch(r"[0-9]{1,16}", timestamp) or abs(now - int(timestamp)) > 300: return False if not re.fullmatch(r"[0-9a-f]{64}", given): return False signed = str(int(timestamp)).encode("utf-8") + b"." + raw_body expected = hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, given) # Set a body limit: reject oversized requests before reading them. MAX_BODY_BYTES = 1_000_000 class WebhookHandler(BaseHTTPRequestHandler): def reply(self, status): self.send_response(status) self.send_header("Content-Length", "0") self.end_headers() def do_POST(self): length_header = self.headers.get("Content-Length", "") if not re.fullmatch(r"[0-9]{1,12}", length_header): self.reply(400) return length = int(length_header) if length > MAX_BODY_BYTES: self.reply(413) return raw_body = self.rfile.read(length) if not is_valid_webhook(SECRET, self.headers.get("X-Webhook-Signature", ""), raw_body): self.reply(400) return try: event = json.loads(raw_body) except ValueError: self.reply(400) return # Answer within the timeout first; do slow work after answering or in a queue. self.reply(200) if event["id"] in seen_event_ids: return seen_event_ids.add(event["id"]) print("Received", event["type"], event["data"]["id"], event["data"]["status"]) HTTPServer(("", 8000), WebhookHandler).serve_forever() ``` ## Rotating the secret **Create a new secret** on the endpoint's page replaces the secret at once; requests signed with the old one stop at that moment. Update your receiver with the new secret straight after. Attempts that fail in between are retried ([Webhook delivery and retries](https://docs.rephut.com/developers/webhooks/delivery)) and are signed with the new secret.