Tornado: verify a Sume webhook in a RequestHandler

A Tornado RequestHandler already holds the raw body as self.request.body. Hash <timestamp>.<body> with HMAC-SHA256, accept any matching sume-v1= entry in the signature header, and raise tornado.web.HTTPError(401) when nothing matches. The program ran on Tornado 6.5 with Python 3.14.

Read the raw bytes in Tornado

Tornado never parses JSON for you, which makes it the least surprising framework for a signed webhook. self.request.body is the bytes of the request, and json.loads takes bytes directly.

Header lookup on self.request.headers is case-insensitive. Missing headers return the default you give, so the verifier receives empty strings and returns False instead of raising.

• self.request.body for the check and for json.loads.
• raise tornado.web.HTTPError(401) to answer 401.
• asyncio.run(main()) with an asyncio.Event().wait() keeps the server up without a top-level await.

The receiver

The file builds the app inside async def main() and runs it with asyncio.run(main()). The secret check is one line that calls sys.exit with a message. Install with pip install tornado and run python server.py with PORT=8000.

import asyncio, hashlib, hmac, json, os, sys, time
import tornado.web

SECRET = os.environ.get("SUME_COM_WEBHOOK_SIGNING_SECRET") or sys.exit("SUME_COM_WEBHOOK_SIGNING_SECRET is empty")
SEEN: set[str] = set()

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

class Hook(tornado.web.RequestHandler):
    def post(self):
        h = self.request.headers
        if not verify(self.request.body, h.get("x-sume-webhook-timestamp", ""),
                      h.get("x-sume-webhook-signature", "")):
            raise tornado.web.HTTPError(401)
        event = json.loads(self.request.body)
        if event.get("job_id") and event["job_id"] not in SEEN:
            SEEN.add(event["job_id"])
            print("new job", event["job_id"], flush=True)
        self.finish("ok")

async def main():
    tornado.web.Application([(r"/hooks/sume", Hook)]).listen(int(os.environ.get("PORT", "8000")))
    await asyncio.Event().wait()

asyncio.run(main())

What the check has to do

The rules come from the webhooks guide. Sume signs the raw JSON body, so the receiver hashes the bytes it received and never a re-serialised object. During a secret rotation the signature header can hold several comma-separated entries, newest first, and a delivery is good when any one matches.

The secret comes from SUME_COM_WEBHOOK_SIGNING_SECRET. The program above stops at start-up when the variable is empty, so a missing secret cannot turn into an endpoint that accepts everything.

The webhook.test event has no job_id

Send test posts a webhook.test event that carries no job_id. The handler reads event.get("job_id") and, when it is missing, goes straight to self.finish("ok"). A 200 tells Sume the endpoint is up and the signature check works.

Sume also lets you re-send a real terminal event with POST /v1/jobs/{job_id}/webhook/redeliver. A redelivery carries a fresh timestamp and signature, so it passes the same replay window as any new delivery.

Prove it with a signed request

Do the slow work after you stored the event. Sume waits 10 seconds for each attempt, and a handler that renders or downloads inside the request will time out and be retried.

The next block is a Python script that signs one body three ways and prints the status of each answer. Start the receiver with SUME_COM_WEBHOOK_SIGNING_SECRET=whsec_test PORT=8000 python server.py first. The output should be 200, then 401, then 401.

import hashlib, hmac, json, os, sys, time
import urllib.error, urllib.request

secret = os.environ["SUME_COM_WEBHOOK_SIGNING_SECRET"]
url = sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:8000/hooks/sume"
body = json.dumps({"event": "job.completed", "request_id": "job_demo", "job_id": "job_demo",
                   "status": "OK", "payload": {"artifacts": []}}, separators=(",", ":")).encode()

def post(ts: str, key: str) -> int:
    sig = hmac.new(key.encode(), ts.encode() + b"." + body, hashlib.sha256).hexdigest()
    req = urllib.request.Request(url, data=body, method="POST", headers={
        "Content-Type": "application/json",
        "x-sume-webhook-timestamp": ts,
        "x-sume-webhook-signature": "sume-v1=" + sig,
    })
    try:
        return urllib.request.urlopen(req, timeout=10).status
    except urllib.error.HTTPError as err:
        return err.code

now = str(int(time.time()))
print("signed with the right secret:", post(now, secret))
print("signed with a wrong secret:  ", post(now, secret + "x"))
print("right secret, stale timestamp:", post(str(int(now) - 900), secret))