Webhooks with FastAPI
Meet the maestro behind the code, the architect of software solutions, and a burgeoning woodworking aficionado: your humble narrator. By day, I'm a Staff Software Engineer at noteable.io, where since September 2020, I've been constructing digital symphonies with Python, a love affair dating back to 2012. But when the screen glow dims and the keyboard clatter fades, you'll find me amidst the whirr of the saw and the scent of fresh-cut lumber, immersed in my latest woodworking project. It's my respite from the pixelated pandemonium, a testament to all good software engineers finding solace in tangible creations after the inevitable burnout. My haven is nestled near Ann Arbor, MI, where I share my life's joys and journeys with my wife, our discerning cat, Miru, and our spirited dog, Penny. Here's to the code we write, the wood we shape, and the stories we continue to carve.
Webhooks are a common approach for third parties to notify their customers about events happening in real-time. FastAPI is one of the best API frameworks for Python, adding support for webhooks is fast and simple.
A common signing algorithm for webhooks is a SHA256 HMAC signature, but it's easy to write a common function to support md5, sha1, and sha256.
import hmac
from hashlib import md5, sha1, sha256
from functools import partial
def compare_signature(digestmod, key: bytes, message: bytes, expected_signature: str) -> bool:
mac = hmac.new(key, message, digestmod)
return hmac.compare_digest(mac.hexdigest(), expected_signature)
compare_md5_signature = partial(compare_signature, md5)
compare_sha1_signature = partial(compare_signature, sha1)
compare_sha256_signature = partial(compare_signature, sha256)
The most important part is to ensure the use of the hmac.compare_digest function instead of simple string comparison (==), this prevents timing attacks.
from fastapi import Request, Header, FastAPI, Depends, HTTPException
app = FastAPI()
# 32 bytes or more for the best security
WEBHOOK_SECRET = b"443c9cec-9dd6-11e9-b610-3c15c2eb7774"
async def _validate_webhook_signature(request: Request):
webhook_signature = request.headers.get("webhook-signature")
if not webhook_signature:
raise HTTPException(400, "Missing webhook signature header")
request_body = await request.body()
# CHANGE: compare_sha256_signature for the signature your integration uses
if not compare_sha256_signature(WEBHOOK_SECRET, request_body, webhook_signature):
raise HTTPException(401, "Invalid webhook signature")
@app.post("/my-webhook", dependencies=[Depends(_validate_webhook_signature)])
def my_webhook():
# ... process data
return {}
After creating the _validate_webhook_signature function, it can be used on any webhook routes that are defined in the future. The general idea with a webhook signature is to ensure that whatever third-party service sending data to your webhook is who they say they are and not a malicious actor.
The most important concept to remember is that a webhook signature is a replacement for your normal app authentication but is not a replacement for authorization. This difference can be read about more in Appendix I on the Authentication page.
