How to Build a Webhook Receiver in Django

Well, I’m hooked.

A common way to receive data in a web application is with a webhook. The external system pushes data to yours with an HTTP request.

Correctly receiving and processing webhook data can be vital to your application working. In this post we’ll create a Django view to receive incoming webhook data.

Example use case

Imagine our site receives messages via webhook from a system at the infamous Acme Corporation. They follow the convention of sending POST requests with JSON bodies to a path on our site that we provide. They send a header with a secret token which we can use to authenticate their requests.

For the purposes of the example, we’ll ignore what we do with these messages and instead focus on the “scaffolding”.

One (1) Acme Anvil.

Message log model

Before we start building a view, we should consider storing all incoming messages. Logging all incoming messages allows us to debug failures, check their structure is as documented, and otherwise audit what’s happening.

We could use any data store for the messages, but the simplest solution is to use a database model. This provides all the benefits of Django’s ORM and our database server’s durability guarantees.

The messages are JSON, so we can store them directly in a JSONField. Since Django 3.1 this works for all database backends.

We should also store the time we received the message, and index it to improve query performance. This will allow us to see the messages in order. We can also use use it to clear old messages, avoiding indefinite table growth.

Combining these requirements we get this model:

from django.db import models


class AcmeWebhookMessage(models.Model):
    received_at = models.DateTimeField(help_text="When we received the event.")
    payload = models.JSONField(default=None, null=True)

    class Meta:
        indexes = [
            models.Index(fields=["received_at"]),
        ]

Note we’re using models.Index, the modern way to define indexes.

View

Our view should verify the request, receive the incoming message, store it, process it, and reply with a success response. We can do these steps like so:

import datetime as dt
import json
from secrets import compare_digest

from django.conf import settings
from django.db.transaction import atomic, non_atomic_requests
from django.http import HttpResponse, HttpResponseForbidden
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from django.utils import timezone

from example.core.models import AcmeWebhookMessage


@csrf_exempt
@require_POST
@non_atomic_requests
def acme_webhook(request):
    given_token = request.headers.get("Acme-Webhook-Token", "")
    if not compare_digest(given_token, settings.ACME_WEBHOOK_TOKEN):
        return HttpResponseForbidden(
            "Incorrect token in Acme-Webhook-Token header.",
            content_type="text/plain",
        )

    AcmeWebhookMessage.objects.filter(
        received_at__lte=timezone.now() - dt.timedelta(days=7)
    ).delete()
    payload = json.loads(request.body)
    AcmeWebhookMessage.objects.create(
        received_at=timezone.now(),
        payload=payload,
    )
    process_webhook_payload(payload)
    return HttpResponse("Message received okay.", content_type="text/plain")


@atomic
def process_webhook_payload(payload):
    # TODO: business logic
    ...

Note:

URL

We can add a URL mapping to our view with the standard path():

from django.urls import path

from example.core.views import acme_webhook

urlpatterns = [
    ...,
    path(
        "webhooks/acme/mPnBRC1qxapOAxQpWmjy4NofbgxCmXSj/",
        acme_webhook,
    ),
]

The path contains a random string, generated with a password manager. This adds a little extra security-by-obscurity, since we won’t provide this URL to anyone but Acme. This prevents at least URL enumeration attacks from discovering our receiver.

Random URLs in the strings don’t provide real protection. URLs often get copied to insecure places, such as logs, emails, or sticky notes. Unfortunately some webhook callers do not support any authentication mechanism, so this can be the best option.

Tests

To test our webhook view, we can make requests to it with Django’s test client:

import datetime as dt
from http import HTTPStatus

from django.test import Client, override_settings, TestCase
from django.utils import timezone

from example.core.models import AcmeWebhookMessage


@override_settings(ACME_WEBHOOK_TOKEN="abc123")
class AcmeWebhookTests(TestCase):
    def setUp(self):
        self.client = Client(enforce_csrf_checks=True)

    def test_bad_method(self):
        response = self.client.get("/webhooks/acme/mPnBRC1qxapOAxQpWmjy4NofbgxCmXSj/")

        assert response.status_code == HTTPStatus.METHOD_NOT_ALLOWED

    def test_missing_token(self):
        response = self.client.post(
            "/webhooks/acme/mPnBRC1qxapOAxQpWmjy4NofbgxCmXSj/",
        )

        assert response.status_code == HTTPStatus.FORBIDDEN
        assert (
            response.content.decode() == "Incorrect token in Acme-Webhook-Token header."
        )

    def test_bad_token(self):
        response = self.client.post(
            "/webhooks/acme/mPnBRC1qxapOAxQpWmjy4NofbgxCmXSj/",
            HTTP_ACME_WEBHOOK_TOKEN="def456",
        )

        assert response.status_code == HTTPStatus.FORBIDDEN
        assert (
            response.content.decode() == "Incorrect token in Acme-Webhook-Token header."
        )

    def test_success(self):
        start = timezone.now()
        old_message = AcmeWebhookMessage.objects.create(
            received_at=start - dt.timedelta(days=100),
        )

        response = self.client.post(
            "/webhooks/acme/mPnBRC1qxapOAxQpWmjy4NofbgxCmXSj/",
            HTTP_ACME_WEBHOOK_TOKEN="abc123",
            content_type="application/json",
            data={"this": "is a message"},
        )

        assert response.status_code == HTTPStatus.OK
        assert response.content.decode() == "Message received okay."
        assert not AcmeWebhookMessage.objects.filter(id=old_message.id).exists()
        awm = AcmeWebhookMessage.objects.get()
        assert awm.received_at >= start
        assert awm.payload == {"this": "is a message"}

Note:

Further changes

There are many ways we might need to improve our webhook receiver, beyond finishing its business logic. Here are some ideas:

Fin

I hope you’re web-hooked to my blog!

—Adam


Read my book Boost Your Git DX to Git better.


Subscribe via RSS, Twitter, Mastodon, or email:

One summary email a week, no spam, I pinky promise.

Tags: