Webhook signature verification

Every webhook delivery is signed with HMAC SHA256. Verify the signature before trusting the payload — anyone with the URL can POST to your handler. The SDK ships a Webhook.verify helper that handles signature parsing, timing-safe comparison, and the 5-minute timestamp tolerance window.

For registering and managing webhook endpoints (create, list, rotate secret, replay deliveries), see client.webhook_endpoints.

The verify helper

from thrustlab import Webhook
from thrustlab.exceptions import SignatureVerificationError

event = Webhook.verify(
    payload,    # raw request body (bytes)
    signature,  # value of the Thrustlab-Signature header
    secret,     # the whsec_... value from create / rotate_secret
)

Webhook.verify raises SignatureVerificationError on:

• signature mismatch (bad secret, body modification in transit)
• expired timestamp (default tolerance: 300 seconds = 5 minutes)
• malformed Thrustlab-Signature header

On success, it returns an Event (from thrustlab import Event) with .id, .type, and .data.

Critical: verify against the raw bytes of the request body. JSON re-encoding (e.g. request.json followed by json.dumps) changes the signature and verification will fail.

End-to-end example

"""Example FastAPI webhook handler verifying incoming events."""
import os
from fastapi import FastAPI, Request, HTTPException
from thrustlab import Webhook
from thrustlab.exceptions import SignatureVerificationError

app = FastAPI()
WEBHOOK_SECRET = os.environ["THRUSTLAB_WEBHOOK_SECRET"]


@app.post("/webhooks/thrustlab")
async def thrustlab_webhook(request: Request):
    payload = await request.body()
    sig = request.headers.get("Thrustlab-Signature", "")
    try:
        event = Webhook.verify(payload, sig, WEBHOOK_SECRET)
    except SignatureVerificationError as exc:
        raise HTTPException(status_code=400, detail=str(exc))

    if event.type == "simulation.succeeded":
        sim_id = event.data["id"]
        # ... do work ...

    return {"received": True}

Flask handler

from flask import Flask, request, abort
from thrustlab import Webhook
from thrustlab.exceptions import SignatureVerificationError

app = Flask(__name__)
WEBHOOK_SECRET = "whsec_..."  # from client.webhook_endpoints.create()["secret"]

@app.post("/webhooks/thrustlab")
def thrustlab_webhook():
    try:
        event = Webhook.verify(
            request.get_data(),  # raw bytes
            request.headers["Thrustlab-Signature"],
            WEBHOOK_SECRET,
        )
    except SignatureVerificationError:
        abort(400)

    if event.type == "simulation.completed":
        process_result(event.data)
    elif event.type == "simulation.failed":
        notify_failure(event.data)

    # Return any 2xx within 30 seconds, or the server treats it as a
    # delivery failure and retries.
    return "", 204

FastAPI handler

from fastapi import FastAPI, Request, HTTPException
from thrustlab import Webhook
from thrustlab.exceptions import SignatureVerificationError

app = FastAPI()
WEBHOOK_SECRET = "whsec_..."

@app.post("/webhooks/thrustlab")
async def thrustlab_webhook(request: Request):
    raw = await request.body()
    try:
        event = Webhook.verify(
            raw,
            request.headers.get("Thrustlab-Signature", ""),
            WEBHOOK_SECRET,
        )
    except SignatureVerificationError:
        raise HTTPException(400, "bad signature")

    if event.type == "simulation.completed":
        await process_result(event.data)
    return {"ok": True}

Django handler

from django.http import HttpResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from thrustlab import Webhook
from thrustlab.exceptions import SignatureVerificationError

WEBHOOK_SECRET = "whsec_..."

@csrf_exempt
@require_POST
def thrustlab_webhook(request):
    try:
        event = Webhook.verify(
            request.body,  # bytes — already raw in Django
            request.META.get("HTTP_THRUSTLAB_SIGNATURE", ""),
            WEBHOOK_SECRET,
        )
    except SignatureVerificationError:
        return HttpResponseBadRequest("bad signature")

    if event.type == "simulation.completed":
        process_result(event.data)
    return HttpResponse(status=204)

Replay protection (timestamp tolerance)

Each delivery's Thrustlab-Signature header contains a timestamp. The SDK rejects deliveries older than 5 minutes to mitigate replay attacks. If your handler is offline for longer, the server will retry the delivery — the retry has a fresh timestamp, so it verifies cleanly.

To loosen or tighten the window, pass tolerance= (in seconds):

event = Webhook.verify(payload, signature, secret, tolerance=600)  # 10 min
event = Webhook.verify(payload, signature, secret, tolerance=60)   # 1 min

Rotate the signing secret

Rotating the secret replaces it immediately — the old secret stops validating signatures the instant the call returns. There's no overlap window to fall back on, so a try-new-then-fall-back-to-old pattern doesn't apply here.

Rotate during a low-traffic window and redeploy your handler with the new secret as fast as possible:

rotated = client.webhook_endpoints.rotate_secret("wh_...")
WEBHOOK_SECRET = rotated["secret"]  # deploy this immediately

Deliveries that arrive in the gap between rotation and your handler picking up the new secret will fail verification — expect a brief failure window, not a seamless cutover. Nothing is lost: the server's normal retry/backoff schedule redelivers them once you're verifying against the new secret.

Handler contract

See also

• Webhook endpoints — endpoint management, deliveries, replay
• Webhooks guide — full HTTP-level contract
• Event types guide — every event the API emits