Starlette: await request.body() to verify a Sume webhook signature

In a Starlette route, call await request.body() before anything touches the payload, compute HMAC-SHA256 over <timestamp>.<raw_body> with your signing secret, and compare it with each sume-v1= entry in x-sume-webhook-signature using hmac.compare_digest. Only after that call json.loads. The signature covers the exact bytes Sume sent, so a parsed and re-serialized body will not match.

The route below also refuses to start without SUME_COM_WEBHOOK_SIGNING_SECRET. An empty key still produces a valid-looking HMAC, so a missing variable in staging would otherwise accept forged bodies. Sume's webhook docs give the header names, the five-minute replay window and the rotation format.

The route

The module has no top-level await. Run it with uvicorn module:app. It returns 401 for a bad signature, 204 for a verified event, and ignores webhook.test because it carries no job_id.

import hashlib, hmac, json, os, time
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route

SECRET = os.environ.get("SUME_COM_WEBHOOK_SIGNING_SECRET", "")
if not SECRET:
    raise RuntimeError("SUME_COM_WEBHOOK_SIGNING_SECRET is empty")

def valid(raw: bytes, ts: str, header: str) -> bool:
    if not (ts.isascii() and ts.isdigit()) or abs(time.time() - int(ts)) > 300:
        return False
    mac = hmac.new(SECRET.encode(), ts.encode() + b"." + raw, hashlib.sha256).hexdigest()
    want = f"sume-v1={mac}".encode()
    return any(hmac.compare_digest(e.strip().encode(), want) for e in header.split(","))

async def sume(request: Request) -> Response:
    raw = await request.body()
    ts = request.headers.get("x-sume-webhook-timestamp", "")
    sig = request.headers.get("x-sume-webhook-signature", "")
    if not valid(raw, ts, sig):
        return Response(status_code=401)
    event = json.loads(raw)
    if event.get("job_id"):
        pass  # store event["job_id"] durably here, then return
    return Response(status_code=204)

app = Starlette(routes=[Route("/sume", sume, methods=["POST"])])

What each line of the check is for

Each check closes a specific hole. The header can hold several entries during a secret rotation, and the any(...) loop accepts the delivery when one matches.

Caveats

• Do not read the body through a parsed model, a form helper or request.json() before the check. Reading the raw bytes first also keeps the signature test independent of JSON key order and whitespace.
• A verified event is not yet a processed event. Store job_id first and use it as the idempotency key, because delivery makes up to 10 attempts and Redeliver can repeat a terminal event.
• Keep a poll of GET /v1/jobs/{job_id}/status available for jobs whose delivery never arrived.