Webhook Listener with FastAPI

Last year I wrote a webhook listener using Flask-RESTPlus. Now I wrote the same using FastAPI:

import datetime
import hmac
import logging
from uuid import UUID
from fastapi import FastAPI, Header, Request, Response
from pydantic import BaseModel

class WebhookResponse(BaseModel):
    result: str

class WebhookData(BaseModel):
    username: str
    data: dict
    event: str
    timestamp: datetime.datetime
    model: str
    request_id: UUID

app = FastAPI(
    title="NetBox Webhook Listener",
    description="Tested with NetBox 2.9.3",
    version="1.0",
)

APP_NAME = "webhook-listener"
WEBHOOK_SECRET = "My precious"

logger = logging.getLogger(APP_NAME)
logger.setLevel(logging.INFO)
formatter = logging.Formatter("%(asctime)s %(name)s %(levelname)s: %(message)s")
file_logging = logging.FileHandler(f"{APP_NAME}.log")
file_logging.setFormatter(formatter)
logger.addHandler(file_logging)

def do_something_with_the_event(data):
    logger.info("WebhookData received:")
    logger.info(f"Raw data: {data}")
    logger.info(f"Request ID: {data.request_id}")
    logger.info(f"Username: {data.username}")
    logger.info(f"Event: {data.event}")
    logger.info(f"Timestamp: {data.timestamp}")
    logger.info(f"Model: {data.model}")
    logger.info(f"Data: {data.data}")
    logger.info(f"URL in data: {data.data['url']}")

@app.post("/webhook/", response_model=WebhookResponse, status_code=200)
async def webhook(
    webhook_input: WebhookData,
    request: Request,
    response: Response,
    content_length: int = Header(...),
    x_hook_signature: str = Header(None)
):
    if content_length > 1_000_000:
        # To prevent memory allocation attacks
        logger.error(f"Content too long ({content_length})")
        response.status_code = 400
        return {"result": "Content too long"}
    if x_hook_signature:
        raw_input = await request.body()
        input_hmac = hmac.new(
            key=WEBHOOK_SECRET.encode(),
            msg=raw_input,
            digestmod="sha512"
        )
        if not hmac.compare_digest(
                input_hmac.hexdigest(),
                x_hook_signature
        ):
            logger.error("Invalid message signature")
            response.status_code = 400
            return {"result": "Invalid message signature"}
        logger.info("Message signature checked ok")
    else:
        logger.info("No message signature to check")
    do_something_with_the_event(webhook_input)
    return {"result": "ok"}

There is so much to unpack there! Some of the FastAPI features I used (there are some links to FastAPI documentation):

• Request validation: I defined my own WebhookData class that defines the expected data in the webhook request. For example, I’m expecting that a JSON field request_id is an UUID string, and that data field is a dictionary.
• Response validation: I defined the class WebhookResponse to describe that my responses to NetBox will just contain a result string.
• The request object contains the raw request. I needed that data to be able to calculate the signature for the request. Without that I wouldn’t have needed the raw request at all, I could just read the FastAPI-validated fields in webhook_input (as demonstrated in the do_something_with_the_event(webhook_input) call).
• The interesting headers were defined in the function arguments.

Let’s see what is the output like:

2020-10-26 20:14:18,812 webhook-listener INFO: Message signature checked ok
2020-10-26 20:14:18,812 webhook-listener INFO: WebhookData received:
2020-10-26 20:14:18,812 webhook-listener INFO: Raw data: username='admin' data={'id': 14, 'url': '/api/dcim/sites/14/', 'name': 'Test_site', 'slug': 'test_site', 'status': {'value': 'active', 'label': 'Active'}, 'region': None, 'tenant': None, 'facility': '', 'asn': None, 'time_zone': None, 'description': '', 'physical_address': '', 'shipping_address': '', 'latitude': None, 'longitude': None, 'contact_name': '', 'contact_phone': '', 'contact_email': '', 'comments': '', 'tags': [], 'custom_fields': {}, 'created': '2020-10-26', 'last_updated': '2020-10-26T20:14:18.744032+02:00'} event='created' timestamp=datetime.datetime(2020, 10, 26, 18, 14, 18, 764443, tzinfo=datetime.timezone.utc) model='site' request_id=UUID('b3dc3247-883f-4810-ad01-0f50ee6b722d')
2020-10-26 20:14:18,812 webhook-listener INFO: Request ID: b3dc3247-883f-4810-ad01-0f50ee6b722d
2020-10-26 20:14:18,812 webhook-listener INFO: Username: admin
2020-10-26 20:14:18,812 webhook-listener INFO: Event: created
2020-10-26 20:14:18,812 webhook-listener INFO: Timestamp: 2020-10-26 18:14:18.764443+00:00
2020-10-26 20:14:18,812 webhook-listener INFO: Model: site
2020-10-26 20:14:18,812 webhook-listener INFO: Data: {'id': 14, 'url': '/api/dcim/sites/14/', 'name': 'Test_site', 'slug': 'test_site', 'status': {'value': 'active', 'label': 'Active'}, 'region': None, 'tenant': None, 'facility': '', 'asn': None, 'time_zone': None, 'description': '', 'physical_address': '', 'shipping_address': '', 'latitude': None, 'longitude': None, 'contact_name': '', 'contact_phone': '', 'contact_email': '', 'comments': '', 'tags': [], 'custom_fields': {}, 'created': '2020-10-26', 'last_updated': '2020-10-26T20:14:18.744032+02:00'}
2020-10-26 20:14:18,812 webhook-listener INFO: URL in data: /api/dcim/sites/14/

In the output you can first see the “raw” contents of the validated input data. See how it shows the timestamp and request_id fields with the Python data types! Then it printed each field one by one, and finally demonstrated how to access one field (the URL) from the data field of the webhook request.

I must say I’m impressed: This FastAPI thingy looks nice and useful!

To run the code you will need for example uvicorn (that requires also uvloop and httptools to be installed) and gunicorn. I’ve described the running setup in the next post.