Verifying signatures
Anyone who learns your endpoint URL can send it a request. Check the signature on every request before you act on it.
How the signature is made
- Take the
X-AgentlyLeads-Timestampheader (Unix seconds) and the raw request body, exactly as received. - Join them with a dot:
"<timestamp>.<body>". - Compute HMAC-SHA256 of that string, keyed with your endpoint's signing secret (the whole
whsec_...string, as UTF-8). - Hex-encode it and prefix
sha256=. That is theX-AgentlyLeads-Signatureheader.
To verify, compute the same value and compare it with the header using a constant-time comparison. Then reject the request if the timestamp is more than 5 minutes from your clock; the timestamp is part of the signed string, so an old request cannot be replayed with a new one.
Sign the raw bytes. If your framework parses the JSON and you re-serialise it, key order or spacing can change and the signature will not match. Read the body as text or bytes first.
Node.js
Uses Express, but the check itself is plain node:crypto.
import express from "express";
import { createHmac, timingSafeEqual } from "node:crypto";
const SECRET = process.env.AGENTLYLEADS_WEBHOOK_SECRET; // whsec_...
const TOLERANCE_SECONDS = 300;
function isGenuine(rawBody, timestamp, signature) {
if (!timestamp || !signature) return false;
const age = Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp));
if (!Number.isFinite(age) || age > TOLERANCE_SECONDS) return false;
const expected = "sha256=" + createHmac("sha256", SECRET).update(`${timestamp}.${rawBody}`).digest("hex");
const a = Buffer.from(expected);
const b = Buffer.from(signature);
return a.length === b.length && timingSafeEqual(a, b);
}
const app = express();
// express.text keeps the body as the exact string that was signed.
app.post("/agentlyleads/webhooks", express.text({ type: "application/json" }), (req, res) => {
if (!isGenuine(req.body, req.get("X-AgentlyLeads-Timestamp"), req.get("X-AgentlyLeads-Signature"))) {
return res.status(400).send("Invalid signature");
}
const event = JSON.parse(req.body);
// De-duplicate on event.id, queue the work, and answer quickly.
console.log(event.type, event.id);
res.sendStatus(200);
});
app.listen(3000);In a Next.js route handler, read the body with await req.text() before parsing it.
Python
Uses Flask; the check itself is standard library.
import hashlib
import hmac
import json
import os
import time
from flask import Flask, abort, request
SECRET = os.environ["AGENTLYLEADS_WEBHOOK_SECRET"].encode() # whsec_...
TOLERANCE_SECONDS = 300
app = Flask(__name__)
def is_genuine(raw_body: bytes, timestamp: str | None, signature: str | None) -> bool:
if not timestamp or not signature:
return False
try:
age = abs(int(time.time()) - int(timestamp))
except ValueError:
return False
if age > TOLERANCE_SECONDS:
return False
signed = timestamp.encode() + b"." + raw_body
expected = "sha256=" + hmac.new(SECRET, signed, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, signature)
@app.post("/agentlyleads/webhooks")
def agentlyleads_webhook():
raw = request.get_data() # the exact bytes that were signed
if not is_genuine(raw, request.headers.get("X-AgentlyLeads-Timestamp"), request.headers.get("X-AgentlyLeads-Signature")):
abort(400, "Invalid signature")
event = json.loads(raw)
# De-duplicate on event["id"], queue the work, and answer quickly.
print(event["type"], event["id"])
return "", 200Rotating the secret
Rotate from Settings, Webhooks (the circular arrow on the endpoint) or with POST /api/v1/webhooks/{id}/rotate-secret. The old secret stops working at once, including for retries of events already queued, so update your receiver as soon as you have the new one. Events that fail in the gap are retried and can be sent again.
Testing your receiver
Press Send test event on the endpoint, or call POST /api/v1/webhooks/{id}/test. It sends a signed webhook.test event straight away and shows you the status code your server answered with.