Node.js / Express:
import express from "express";
import crypto from "crypto";
const app = express();
app.use(
express.json({
verify: (req, res, buf) => {
req.rawBody = buf;
},
})
);
const HMAC_SECRET = process.env.SULUS_HMAC_SECRET;
function verifySulusSignature(req, res, next) {
const signature = req.headers["x-signature"];
const timestamp = req.headers["x-webhook-timestamp"];
if (!signature) {
return res.status(401).json({ error: "Missing signature header" });
}
if (timestamp) {
const requestTime = parseInt(timestamp, 10);
const now = Math.floor(Date.now() / 1000);
if (Math.abs(now - requestTime) > 300) {
return res.status(401).json({ error: "Request timestamp too old" });
}
}
const expectedSignature = crypto
.createHmac("sha256", HMAC_SECRET)
.update(req.rawBody)
.digest("hex");
const sigBuffer = Buffer.from(signature, "hex");
const expectedBuffer = Buffer.from(expectedSignature, "hex");
if (
sigBuffer.length !== expectedBuffer.length ||
!crypto.timingSafeEqual(sigBuffer, expectedBuffer)
) {
return res.status(401).json({ error: "Invalid signature" });
}
next();
}
app.post("/webhook", verifySulusSignature, async (req, res) => {
res.status(200).send();
});
Python / FastAPI:
import hmac
import hashlib
import time
import os
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
HMAC_SECRET = os.getenv("SULUS_HMAC_SECRET")
async def verify_sulus_signature(request: Request) -> bytes:
signature = request.headers.get("x-signature")
timestamp = request.headers.get("x-webhook-timestamp")
if not signature:
raise HTTPException(status_code=401, detail="Missing signature header")
if timestamp:
request_time = int(timestamp)
now = int(time.time())
if abs(now - request_time) > 300:
raise HTTPException(status_code=401, detail="Request timestamp too old")
raw_body = await request.body()
expected_signature = hmac.new(
HMAC_SECRET.encode("utf-8"),
raw_body,
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(signature, expected_signature):
raise HTTPException(status_code=401, detail="Invalid signature")
return raw_body
@app.post("/webhook")
async def handle_webhook(request: Request):
raw_body = await verify_sulus_signature(request)
return {"status": "ok"}
Always compute the signature from the raw request body — parsing JSON first can reorder bytes and break the match — and use a constant-time comparison (crypto.timingSafeEqual in Node.js, hmac.compare_digest in Python) to avoid timing attacks.