Every webhook delivery carries an X-DSS-Signature header. Verify it on
every request before doing anything with the body — a missing or invalid
signature means the request didn’t come from us (or it’s a replay).
The signature
X-DSS-Signature: t=1716714840,v1=99d56ccfe6de640971036fc31a8bb476415322e6b687301c96fe15ac81e3fcff
t — Unix timestamp (seconds) of when we signed.
v1 — HMAC-SHA256 of <t>.<raw-body-bytes>, encoded as lowercase hex.
The secret is the webhook signing secret we issued at onboarding —
separate from your client_secret. Rotate it independently via your DSS
account contact.
The verification rules
A verifier must enforce all four:
- Parse the header. If
t or v1 is missing or malformed, reject with
400.
- Reject stale events. If
|now - t| > 300 seconds, reject. This is
the replay protection window.
- Recompute HMAC-SHA256 of
f"{t}.".encode() + raw_body_bytes using
the signing secret. Compare in constant time against v1.
- Reject mismatches. Don’t trust any field of the body until the
signature checks.
Use the raw bytes of the request body, not a re-serialised JSON view of
it. JSON re-serialisation reorders keys or changes whitespace, both of which
break the signature. In FastAPI: await request.body(). In Express:
express.raw({ type: "application/json" }) and read req.body as a Buffer.
Python
Copy this verbatim:
import hashlib
import hmac
import time
from typing import Final
REPLAY_WINDOW_S: Final = 300
def verify_dss_signature(raw_body: bytes, header: str, secret: bytes) -> bool:
"""Returns True iff the signature matches AND the timestamp is fresh."""
parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)
try:
ts = int(parts["t"])
sig_hex = parts["v1"]
except (KeyError, ValueError):
return False
if abs(time.time() - ts) > REPLAY_WINDOW_S:
return False
signed = f"{ts}.".encode("ascii") + raw_body
expected = hmac.new(secret, signed, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, sig_hex)
FastAPI handler
import json
from fastapi import FastAPI, Header, HTTPException, Request, Response
app = FastAPI()
WEBHOOK_SECRET = b"..." # from your secret store
@app.post("/integrations/dss/webhook")
async def dss_webhook(
request: Request,
x_dss_signature: str | None = Header(default=None),
) -> Response:
if x_dss_signature is None:
raise HTTPException(status_code=400, detail="missing signature")
# IMPORTANT: raw bytes, not re-serialised JSON.
raw = await request.body()
if not verify_dss_signature(raw, x_dss_signature, WEBHOOK_SECRET):
raise HTTPException(status_code=400, detail="invalid signature")
event = json.loads(raw)
if seen_recently(event["id"]):
return Response(status_code=200) # idempotent dedupe
handle(event)
return Response(status_code=200)
Flask handler
from flask import Flask, request
app = Flask(__name__)
WEBHOOK_SECRET = b"..."
@app.post("/integrations/dss/webhook")
def dss_webhook():
sig = request.headers.get("X-DSS-Signature")
if sig is None:
return "missing signature", 400
raw = request.get_data() # raw bytes
if not verify_dss_signature(raw, sig, WEBHOOK_SECRET):
return "invalid signature", 400
event = request.get_json(force=True)
if not seen_recently(event["id"]):
handle(event)
return "", 200
Tests against the spec
Our canonical fixture:
import hmac
import hashlib
event_body = (
b'{"created_at":"2026-05-26T09:14:00Z",'
b'"data":{"user_id":"65a1f0e2c3b4d5e6f7a8b9c0"},'
b'"id":"evt_3f4a9c8e2b1d4f5a8c9e0d1f2a3b4c5d",'
b'"type":"user.sea_time.updated"}'
)
ts = 1716714840
secret = b"example-partner-webhook-secret-32"
expected = hmac.new(secret, f"{ts}.".encode() + event_body, hashlib.sha256).hexdigest()
assert expected == "99d56ccfe6de640971036fc31a8bb476415322e6b687301c96fe15ac81e3fcff"
Your verifier should accept this with header
t=1716714840,v1=99d56ccfe6de640971036fc31a8bb476415322e6b687301c96fe15ac81e3fcff
only within 5 minutes of the timestamp. If you’re testing offline, freeze
the clock at 1716714840.
Node
import { createHmac, timingSafeEqual } from "node:crypto";
import { Buffer } from "node:buffer";
const REPLAY_WINDOW_S = 300;
export function verifyDssSignature(rawBody, header, secret) {
if (typeof header !== "string") return false;
const parts = Object.fromEntries(
header.split(",").map((p) => {
const i = p.indexOf("=");
return i < 0 ? [p, ""] : [p.slice(0, i), p.slice(i + 1)];
})
);
const t = Number.parseInt(parts.t ?? "", 10);
const sigHex = parts.v1;
if (!Number.isFinite(t) || !sigHex) return false;
if (Math.abs(Date.now() / 1000 - t) > REPLAY_WINDOW_S) return false;
const signed = Buffer.concat([
Buffer.from(`${t}.`, "ascii"),
Buffer.isBuffer(rawBody) ? rawBody : Buffer.from(rawBody),
]);
const expectedHex = createHmac("sha256", secret).update(signed).digest("hex");
const expected = Buffer.from(expectedHex, "hex");
const provided = Buffer.from(sigHex, "hex");
if (expected.length !== provided.length) return false;
return timingSafeEqual(expected, provided);
}
Express handler
import express from "express";
import { verifyDssSignature } from "./verify.js";
const app = express();
const WEBHOOK_SECRET = process.env.DSS_WEBHOOK_SECRET;
app.post(
"/integrations/dss/webhook",
// IMPORTANT: raw, not express.json — we need the unmodified body bytes.
express.raw({ type: "application/json", limit: "32kb" }),
async (req, res) => {
const sig = req.get("X-DSS-Signature");
if (!verifyDssSignature(req.body, sig, WEBHOOK_SECRET)) {
return res.status(400).send("invalid signature");
}
const event = JSON.parse(req.body.toString("utf8"));
if (await seenRecently(event.id)) return res.status(200).end();
await handle(event);
res.status(200).end();
}
);
Next.js Route Handler (App Router)
import { verifyDssSignature } from "@/lib/dss";
const WEBHOOK_SECRET = process.env.DSS_WEBHOOK_SECRET!;
export async function POST(req: Request) {
const sig = req.headers.get("x-dss-signature");
const raw = Buffer.from(await req.arrayBuffer());
if (!verifyDssSignature(raw, sig, WEBHOOK_SECRET)) {
return new Response("invalid signature", { status: 400 });
}
const event = JSON.parse(raw.toString("utf8"));
if (await seenRecently(event.id)) return new Response(null, { status: 200 });
await handle(event);
return new Response(null, { status: 200 });
}
What you must do (checklist)
Verify the signature on every request
No verified signature, no body parsing. Return 400.
Enforce the 5-minute replay window
Even with a valid signature, reject events older than 300 seconds.
Compare in constant time
hmac.compare_digest (Python) or crypto.timingSafeEqual (Node).
String == is wrong.
Dedupe on event id
At-least-once delivery means you will see duplicates. The id is stable
across attempts — store the last N you’ve seen in Redis or your DB.
Respond 2xx within 5 seconds
Hand off heavy work to a background queue. Acknowledging fast keeps you
out of the retry schedule.
Always return 2xx on duplicates
A 4xx on a duplicate looks like a delivery failure to us and triggers
pointless retries.
What you must not do
- Do not trust the body before the signature checks.
- Do not verify against a re-serialised body. The signature is over the
bytes we sent, in the order we sent them.
- Do not log the signing secret. Treat it like any other long-lived
credential.
- Do not allow
http:// for your webhook URL outside local development.
We won’t deliver to plain HTTP in production.