A webhook refers to an HTTP callback initiated when an event occurs. They help notify different web apps on the internet or any network about an occurrence.
An example of a Webhook
GitHub makes it simple to set up Webhooks for your git repositories. You can choose which events you want to be notified about, such as pushes and pull requests, and you’ll only be told when they happen. It may connect external apps to GitHub, execute Continuous Integration operations, and automate deployments.
Consider the following continuous integration situation. The requirement is that the testing and deployment process must begin as soon as a developer sends his code modifications to the repository. A webhook is ideal to accomplish this.
Webhooks in Django
Let’s use Django to establish an event receiver endpoint.
from django.views.decorators.http import require_http_methods
@require_http_methods(["GET", "POST"]) # Listens only for GET and POST requests
def hook_receiver_view(request):
# returns django.http.HttpResponseNotAllowed for other requests
# Handle the event appropriately
return HttpResponse('success')
It is a detailed view that supports GET and POST requests. The view must adequately handle the event. In the application that generates the event, use the URL of this View to configure the webhook.
That’s all it takes to get notified of an upcoming event.
Security
In general, every such call will require passing some data, which will be done as part of the GET or POST request. We must be cautious of the information sent as part of the request because it can be accessible by anybody on the internet, and no authentication is provided.
Never trust the values given by the client, as the saying goes. Webhooks are the same way. They could be a malicious attack and could come from anyone -anyone could set off the incident.
Always check to see if the request is genuine.
Consider the case when a payment gateway sends a signal to our application after a user completes a transaction. The gateway must provide some form of user identity, such as ‘id.’ Assume https://thirdeyemedia.wpmudev.host/codeunderscored/payment-confirm/ is the callback URL.
The callback URL would be https://thirdeyemedia.wpmudev.host/codeunderscored/payment-confirm/?id=abc if the request is a GET request and the user’s id is ‘abc’. Thus, we could do something similar in our view.
@require_http_methods(["GET", "POST"])
def hook_receiver_view(request):
user_id = request.GET.get('id', None)
# Save the payment status
pay_info = Payment.objects.get(user_id=user_id)
pay_info.payment_successful = True
pay_info.save()
return HttpResponse('successful payment')
But hold on a second. We can’t presume that the user’s payment with id ‘abc’ is successful because the user’s id is supplied as a URL parameter. With their id, anyone can make a GET request. As a result, the view must confirm that the user ‘abc’ has completed the payment via the payment gateway service’s API. That is something we should be doing.
@require_http_methods(["GET", "POST"])
def hook_receiver_view(request):
user_id = request.GET.get('id', None)
# This is where we are verifying the payment
if payment_service.hasUserPaid(user_id):
# Save the payment status
pay_info = Payment.objects.get(user_id=user_id)
pay_info.payment_successful = True
pay_info.save()
return HttpResponse('success')
Because the webhook service expects it, we should always return the success HttpResponse to it. Otherwise, the service may assume that our callback failed to handle the event correctly and attempt to trigger the event again.
You can also think about restricting the number of requests you accept. The latter concept, referred to as throttling, is built-in if you’re utilizing the Django Rest Framework. For Django, you can use the Django Ratelimit.
Using dj-webhooks to create webhooks
dj-webhooks enables us to create webhooks with event creation and management features, callbacks, and logs. Let us look at the library, even though it is ancient. First, let us install the package by running the following command,
pip install dj-webhooks
Then, create the events triggered in the Django settings model as follows.
WEBHOOK_EVENTS = ( "pay_info.paid", "pay_info.cancelled", "pay_info.refunded", "pay_info.fulfilled" )
The webhooks model must be saved when a user registers a callback URL for an event. It is what our point of view should be like.
from djwebhooks.models import WebhookTarget
def save_payment_success_webhook(request):
WebhookTarget.objects.create(
owner=request.user,
event='pay_info.succeeded',
target_url= request.POST.get('callback_url'),
header_content_type=WebhookTarget.CONTENT_TYPE_JSON,
)
# Some other operations
We need to use the URL listed for an event now that the webhook has been added to the database by a user. To do so, we’ll need to create a function that produces a JSON – serializable object, such as a dictionary. To have the function trigger the webhook, apply the decorator djwebhooks.decorators.hook to it.
This is how the process appears.
from djwebhooks.decorators import hook
# The argument to the decorator specifies the event
def send_purchase_confirmation(pay_info, owner):
return {
"order_num": pay_info.order_num,
"date": pay_info.confirm_date,
"email": pay_info.email
}
Once this method is complete, invoking it will send a request to the callback URL supplied by the user as the parameter owner, with the delivered data in JSON format as the request payload. Django rest hooks, which are still in development, are another option.
Sample Webhook Receiver in Django
This section will create a Django view to receive incoming webhook data. We will assume our site receives notifications from the Codeunderscored system via webhook. So, they submit POST requests with JSON contents to a specific path on our website that we specify. These include a secret token in their header, which we may use to authenticate their requests.
We’ll disregard what we do with these messages for the sake of this example. Instead, focus on the “scaffolding.”
Model: a message log
We might consider saving all incoming messages before we start creating a view. Then, we can diagnose problems, confirm their structure is documented, and audit what’s going on by logging all incoming messages.
We could use any data storage for the messages, but the simplest way is to utilize a database model. It combines the advantages of Django’s ORM with the reliability of our database server.
Because the messages are JSON, we can directly save them in a JSONField. It has worked for all database backends since Django 3.1.
To increase query performance, we should also save and index the time we received the message. It will enable us to see the messages in a chronological sequence. We may also use it to delete old messages, preventing the table from growing indefinitely.
We get the following model by combining these requirements:
from django.db import models
class CodeunderscoredWebhookMessage(models.Model):
received_at = models.DateTimeField(help_text=" time event is received .")
payload = models.JSONField(default=None, null=True)
class Meta:
indexes = [
models.Index(fields=["received_at"]),
]
It’s worth noting that we’re working with the modern approach of defining indexes, models.Index.
View
Our view should validate the request, accept the incoming message, store it, process it, and respond successfully. These steps can be carried out as follows:
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
import datetime as date_time
import json
from secrets import compare_digest
from example.core.models import CodeunderscoredWebhookMessage
@csrf_exempt
@require_POST
@non_atomic_requests
def codeunderscored_webhook(request):
given_token = request.headers.get("Codeunderscored-Webhook-Token", "")
if not compare_digest(given_token, settings.CODEUNDERSCORED_WEBHOOK_TOKEN):
return HttpResponseForbidden(
"Incorrect token in Codeunderscored-Webhook-Token header.",
content_type="text/plain",
)
CodeunderscoredWebhookMessage.objects.filter(
received_at__lte=timezone.now() - date_time.timedelta(days=7)
).delete()
payload = json.loads(request.body)
CodeunderscoredWebhookMessage.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
…
Discussion
@csrf_exempt
Django’s default cross-site request forgery (CSRF) protection is disabled via @csrf_exempt. We wouldn’t usually allow a POST request without a CSRF token since it could signal that a user has been duped into submitting a malicious form to our site from another site. However, we use various authentication systems to check requests for webhooks, allowing us to disable CSRF.
@non_atomic_requests
@non atomic requests disable this view’s ATOMIC REQUESTS (transaction-per-request). Adding transactions to your Django application with ATOMIC REQUESTS is usually a good idea and a simple process. We’re utilizing direct transaction control here (the @atomic on process webhook payload) to ensure that the CodeunderscoredWebhookMessage is saved for debugging if our business logic fails. As a result, we don’t want a transaction centered on the entire screen.
Codeunderscorede’s system uses a token in the Codeunderscored-Webhook-Token header to establish authentication. This header is compared to the token they should be using, saved in an environment variable, and read in our settings. We can reject the incoming message if the two do not match.
The comparison is done with secrets.compare_digest(). Unlike regular string comparisons, this will always take the same amount of time regardless of the supplied string. It makes it impossible for timed attacks to obtain our secret token.
Because webhook receivers are on the public web, anyone could find them, authentication is essential. Because there is no universal standard for webhooks, callers use a variety of authentication mechanisms. Check your caller’s documentation if you’re adapting this code.
We clear up saved messages older than a week before storing the new message. It is a primary method for deleting old data.
If our webhook is used frequently, running this remove query each time could get costly. In this situation, similar to Django’s clearsessions, we may transfer the deletion to a background process.
To load the request body, we use json.loads(). It is done without validating the Content-Type header or resolving any errors if the body isn’t valid JSON. If an error occurs, the view will crash, and our error reporting program (such as Sentry) will notify us.
For our purposes, this is a suitable failure mode. If the body is not JSON, then that means that something has gone wrong. As a result, we’d like to know about it now that we’ve confirmed the message is from Codeunderscored.
Before attempting to process the data, we save it in the CodeunderscoredWebhookMessage model. It assures that even if we crash later, it will be logged.
Our business logic handler is what we name it. It has a stub implementation, which has been left empty in this example. We’d put some code here in a real-world application. However, delivering the first version with an empty handler is helpful to ensure that messages are appropriately received.
Our view responds with a plain-text OK response. Because most webhook callers only examine the status code, we can keep the body short.
URL
We use the standard path() to add a URL mapping to our view as follows:
from django.urls import path from example.core.views import codeunderscored_webhook urlpatterns = [ path( "webhooks/code/x6si2ioub15a5xh4/", codeunderscored_webhook, ), ]
A random string was produced with a password manager and placed in the route. We won’t give this URL to anyone but Codeunderscored. Thus this adds a layer of protection by obscurity. At the very least, this prevents URL enumeration attempts from detecting our receiver.
The use of random URLs in the strings does not give genuine security. URLs are frequently copied to insecure locations like logs, emails, and sticky notes. However, it may be the best alternative because specific webhook callers do not allow authentication.
from django.test import TestCase, Client, override_settings
from django.utils import timezone
import datetime as date_time
from http import HTTPStatus
from example.core.models import CodeunderscoredWebhookMessage
@override_settings(CODEUNDERSCORED_WEBHOOK_TOKEN="123456")
class CodeunderscoredWebhookTests(TestCase):
def setUp(self):
self.client = Client(enforce_csrf_checks=True)
def test_bad_method(self):
response = self.client.get("/webhooks/code/x6si2ioub15a5xh4/")
assert response.status_code == HTTPStatus.METHOD_NOT_ALLOWED
def test_missing_token(self):
response = self.client.post(
"/webhooks/code/x6si2ioub15a5xh4/",
)
assert response.status_code == HTTPStatus.FORBIDDEN
assert (
response.content.decode() == "Incorrect token in Codeunderscored-Webhook-Token header."
)
def test_bad_token(self):
response = self.client.post(
"/webhooks/code/x6si2ioub15a5xh4/",
HTTP_CODEUNDERSCORED_WEBHOOK_TOKEN="def456",
)
assert response.status_code == HTTPStatus.FORBIDDEN
assert (
response.content.decode() == "Incorrect token in Codeunderscored-Webhook-Token header."
)
def test_success_message(self):
start = timezone.now()
old_message = CodeunderscoredWebhookMessage.objects.create(
received_at=start - date_time.timedelta(days=24),
)
response = self.client.post(
"/webhooks/code/x6si2ioub15a5xh4/",
HTTP_CODEUNDERSCORED_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 CodeunderscoredWebhookMessage.objects.filter(id=old_message.id).exists()
_code = CodeunderscoredWebhookMessage.objects.get()
assert _code.received_at >= start
assert _code.payload == {"this": "is a message"}
To replace the token setting for each test in the test case, we use @override_settings. It implies we don’t need to use the real sensitive token, which we shouldn’t save in our code base or set a value in our test settings.
We use the enforce_csrf_checks flag on our test client to verify the @csrf_exempt decorator. If we unintentionally removed the decorator from the view, the test client would raise a CSRF issue.
Before we test the view’s success case, we test its different failure modes. Testing the missing and wrong token scenarios for coverage is unnecessary, but it is done for completeness if the code changes.
We match the response status codes to the HTTPStatus enum from the Python standard library when making assertions. As a result, we must use the rather uncomfortable HTTP_* syntax to transmit the Codeunderscored-Webhook-Token header.
Tests
We can use Django’s test client to make queries to our webhook view to test it:
Conclusion
Webhooks are a simple way to alert external services when a specific event occurs. It is a widespread technique for a web application to receive data. With an HTTP request, the external system sends data to yours.
Receiving and processing webhook data correctly can be critical to your application’s success.
