Verifying a delivery
`verify_webhook_signature`: constant-time, with a replay window, and the parsed event back.
In a request handler
A webhook URL is public. Anything on the internet can POST the right-shaped JSON at it, so a handler that reads payload['type'] without checking the signature is an open write API.
import os from fastapi import FastAPI, Request, Responsefrom openemail import WebhookVerificationError, verify_webhook_signature app = FastAPI() @app.post('/webhooks/openemail')async def webhook(request: Request) -> Response: try: event = verify_webhook_signature( payload=await request.body(), headers=request.headers, secret=os.environ['OPENEMAIL_WEBHOOK_SECRET'], tolerance_seconds=300, ) except WebhookVerificationError: return Response('bad signature', status_code=400) print(event['type'], event['data']) return Response(status_code=204)Pass the RAW body. Parsing and re-serialising changes key order and whitespace, and the signature will not match. It may be bytes or str: await request.body() in FastAPI, request.body in Django, request.get_data() in Flask. headers takes any mapping, such as the request.headers of each of those frameworks or a plain dict, and X-OpenEmail-Signature is found whatever its case.
Two things this handles that a hand-rolled check usually does not: it compares the MAC in constant time, so the correct prefix cannot be recovered by timing it, and it rejects a delivery more than tolerance_seconds old in either direction, five minutes unless you say otherwise, so a captured request is not replayable for ever. tolerance_seconds=0 turns the replay check off. Both bugs are silent. A handler with either one passes every test you would think to write.
It raises WebhookVerificationError on every failure: an empty secret, a missing X-OpenEmail-Signature header, one not in the form t=<seconds>,v1=<hex>, a timestamp outside the window, a signature that does not match, or a body that is not JSON. On success it returns the body parsed as a WebhookPayload, so there is no second json.loads to get wrong. Annotate the result, such as WebhookPayload[EmailOpenedData], to type data.
import os from openemail import verify_webhook_signaturefrom openemail.types import EmailOpenedData, WebhookPayload def on_open(body: bytes, headers: dict[str, str]) -> None: event: WebhookPayload[EmailOpenedData] = verify_webhook_signature( payload=body, headers=headers, secret=os.environ['OPENEMAIL_WEBHOOK_SECRET'], ) print(event['data']['recipient'], event['data']['first'], event['data']['country'])WebhookVerificationError subclasses both OpenEmailError and ValueError. The function is a plain one rather than a coroutine, so call it without await from sync and async code alike. It needs nothing beyond hmac and hashlib from the standard library.