Falcon ASGI: verify a Sume webhook with req.bounded_stream.read()

In Falcon's ASGI app, read the raw body of a Sume webhook with await req.bounded_stream.read(), compute HMAC SHA-256 over <timestamp>.<raw_body>, and compare it with each sume-v1= entry in the signature header. Set resp.status = falcon.HTTP_204 when the check passes and raise falcon.HTTPUnauthorized when it does not.

Falcon parses media only when you ask, so there is no body-consumed-by-middleware trap here. The pitfall is the opposite one: calling await req.get_media() first and signing a re-serialized object.

Why bounded_stream?

Falcon's ASGI reference documents Request.bounded_stream as an alias to stream, and its stream object has an awaitable read(). Calling read() with no arguments returns the whole body. The same page warns that an app may not use both read() and the asynchronous iterator on the same request, so pick one reader and stick with it.

Sume webhook bodies are small, and run receipts above 1 MiB are never sent inline. For a hard cap, check req.content_length first and answer 413 yourself above 1,048,576 bytes.

import hashlib, hmac, os, time

def verify(secret, ts, header, body: bytes) -> bool:
    if not secret or not ts or not header:
        return False
    try:
        t = int(ts)
    except ValueError:
        return False
    if abs(time.time() - t) > 300:
        return False
    msg = f"{t}.".encode() + body
    want = "sume-v1=" + hmac.new(secret.encode(), msg, hashlib.sha256).hexdigest()
    return any([hmac.compare_digest(e.strip().encode(), want.encode())
                for e in header.split(",")])

What does the resource look like?

Falcon routes by responder method. Add on_post to a resource class, mount it on the path you registered with Sume, and serve the ASGI app with uvicorn app:app.

import falcon
import falcon.asgi

class SumeHook:
    async def on_post(self, req, resp):
        raw = await req.bounded_stream.read()
        ok = verify(
            os.environ.get("SUME_COM_WEBHOOK_SIGNING_SECRET", ""),
            req.get_header("X-Sume-Webhook-Timestamp") or "",
            req.get_header("X-Sume-Webhook-Signature") or "",
            raw,
        )
        if not ok:
            raise falcon.HTTPUnauthorized(title="bad signature")
        # insert the event keyed by job_id or request_id, then acknowledge
        resp.status = falcon.HTTP_204

app = falcon.asgi.App()
app.add_route("/hooks/sume", SumeHook())

Which header and id do I log?

Each Sume delivery also carries x-sume-webhook-secret-fingerprint. When a signature fails to verify, log that header next to the fingerprint shown beside the secret in the dashboard. A mismatch means you hold the wrong secret, and neither side has to send the secret itself.

How do I test it before pointing Sume at it?

Use the dashboard's Send test control, or POST /v1/webhooks/test-deliveries with an API key that has account:write. It posts a signed webhook.test payload to a URL you type, so you can confirm the route returns 204 without spending a generation.

Then submit one real job with mode: "webhook" and a public HTTPS webhook_url. Sume rejects localhost, private-network and non-HTTPS URLs, so use a tunnel while developing. When the job finishes you should see exactly one job.completed, job.failed or job.canceled delivery.

• Sign await req.bounded_stream.read(), never a re-dumped get_media() result.
• Return 204 only after the event is stored.
• Keep polling the job as the backup path.

What should happen after the response?

Write the event down, answer 204, then do the slow work. Sume gives each attempt 10 seconds. Dedupe on job_id for job events and request_id for run events, and treat polling the status_url as your recovery path for anything that never arrives.