Date:

Share:

Webhooks in Django | Code Underscored

Related Articles

Webhook refers to HTTP reconnection that is activated when an event occurs. They help inform various web applications on the web or on any network about an occurrence.

Webhook example

GitHub makes it easy to set up Webhooks for your Git repositories. You can choose which events you want to be notified about, such as urgency and withdrawal requests, and you will be notified only when they occur. It may connect external applications to GitHub, perform continuous integration operations and automate deployments.

Consider the following ongoing integration mode. The requirement is that the testing and deployment process must begin as soon as a developer submits his code changes to the repository. Webhook is ideal to achieve this.

Webhooks in Django

Let’s use Django to set up an endpoint for an event shelter.

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')

This is a detailed view that supports GET and POST requests. The landscape must adequately deal with the event. In the application that creates the event, use the URL of this view to configure the webhook.

This is all it takes to get notified of an upcoming event.

security

In general, any such call will require the transmission of certain data, which will be made as part of a GET or POST request. We must be careful about the information sent as part of the request as it can be accessible to anyone online, and no authentication is provided.

Never rely on the values ​​given by the customer, as stated. Webhooks are the same thing. They can be a malicious attack and can come from anyone – anyone can start the event.

Always check if the request is genuine.

Consider the case when a payment gateway sends a signal to our app after a user completes a transaction. The gateway must provide some form of user identity, such as ‘ID’. Suppose https://www.codeunderscored.com/payment-confirm/ is the return call URL.

The redial URL will be https://www.codeunderscored.com/payment-confirm/?id=abc if the request is a GET request and the user ID is ‘abc’. Thus, we can do something similar in our opinion.

@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 wait a second. We can not assume that the payment of the user with the ‘abc’ ID is successful because the user ID is provided as a URL parameter. With his ID, anyone can apply for a GET. As a result, the display must confirm that the user ‘abc’ completed the payment through the API of the payment gateway service. This is something we need to do.

@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 this, we must always return the HttpResponse success to it. Otherwise, the service may assume that our callback failed to handle the event correctly and try to run the event again.

You can also think about limiting the number of requests you receive. The latter concept, known as the throttle, is built-in if you use the Django Rest Framework. For Django, you can use Django Ratelimit.

Using dj-webhooks to create webhooks

dj-webhooks allows us to create webhooks with event creation and management features, engagements and calendars. Let’s take a look at the library, even though it is ancient. First, let’s install the package by running the following command,

Next, create the triggered events in Django’s configuration model as follows.

WEBHOOK_EVENTS = (
"pay_info.paid",
"pay_info.cancelled",
"pay_info.refunded",
"pay_info.fulfilled"
)

The webhooks model should be saved when a user registers a redial URL for an event. That should be our point of view.

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 registered for an event now that the webhook has been added to the database by a user. To do this, we’ll need to create a function that produces JSON – a serial object, such as a dictionary. For the function to start the webhook, apply the djwebhooks.decorators.hook decorator to it.

This is what the process looks like.

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
    

Upon completion of this method, its activation will send a request to the callback address provided by the user as the parameter owner, with the data provided in JSON format as the load of the request. Django rest hooks, which are still in development, are another option.

Example Webhook receiver in Django

This section will create a Django view to get incoming webhook data. We assume that our site receives messages from the Codeunderscored system via webhook. So, they submit POST requests with JSON content for a specific path on our site that we specify. These include a secret token in their title, which we may use to verify their requests.

We will ignore what we do with these posts for the sake of this example. Instead, focus on “scaffolding.”

Model: Message log

We may consider saving all incoming messages before we start creating a view. Then we can diagnose problems, confirm their structure is documented and check what happens by recording all incoming messages.

We can use any data storage for the messages, but the simplest way is to use a database model. It combines the benefits of Django’s ORM with the reliability of our database server.

Because the messages are JSON, we can save them directly in JSONField. It has worked for the entire back end of the database since Django 3.1.

To increase query performance, we must also save and add the time we received the message. This will allow us to see the messages in chronological order. We may also use it to delete old posts, and prevent the table from growing indefinitely.

We obtain 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 is worth noting that we work with the modern approach of defining indices, models.index.

View

Our view should validate the request, receive the incoming message, store it, process it and respond successfully. These steps can be performed 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 Cross-Site Forgery (CSRF) default protection is disabled by @csrf_exempt. Normally we would not allow a POST request without a CSRF token because it could signal that a user has been tricked into submitting a malicious form to our site from another site. However, we use different authentication systems to check webhooks requests, which allows us to disable CSRF.

@non_atomic_requests

@non atomic requests disables the ATOMIC REQUESTS of this view (transaction per request). Adding transactions to your Django app with ATOMIC REQUESTS is usually a good idea and a simple process. We use direct transaction control here (the @atomic on process webhook load) to ensure that the CodeunderscoredWebhookMessage is saved for debugging if our business logic fails. As a result, we do not want a deal to be concentrated on the entire screen.

Codeunderscorede’s system uses a token in the Codeunderscored-Webhook-Token header to establish authentication. This title is compared to the token they need to use, is stored in an environment variable, and is called in our settings. We can reject the incoming message if the two do not match.

The comparison is made with secrets.compare_digest (). Unlike regular string comparisons, it will always take the same amount of time regardless of the string provided. This does not allow scheduled attacks to obtain our secret token.

Because webhook receivers are on the public internet, anyone can 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 fit this code.

We clear saved messages more than a week before saving the new message. This is a primary method of deleting old data.

If our webhook is used frequently, running this removal query each time can be costly. In this situation, similar to Django’s scans, we may move the deletion to a background process.

To load the request body, we use json.loads (). This is done without verifying the Content-Type header or resolving any errors if the body is not a valid JSON. If an error occurs, the display will crash, and our error reporting program (such as Sentry) will notify us.

For our purposes, this is an appropriate failure situation. If the body is not JSON, then it means something went wrong. As a result, we’ll want to know about it now that we’ve confirmed that the message is from Codeunderscored.

Before attempting to process the data, we save it in the CodeunderscoredWebhookMessage model. This ensures that even if later thawed, it will be recorded.

The handler of our business logic is what we call it. It has a stab realization, which is left blank in this example. We put code here in a real app. However, delivering the first version with an empty handler is helpful to ensure that messages are received properly.

Our view responds in a fine response in plain text. Since most webhook callers only check the status code, we can keep the body short.

website address

We use the regular 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 generated with a password manager and placed on the track. This URL is not given to anyone other than Codeunderscored. Thus it adds a layer of protection by fogging. At the very least, it prevents URL counting attempts from identifying our shelter.

The use of random URLs in strings does not provide real security. URLs are often copied to insecure locations like logs, emails, and sticky notes. However, this may be the best alternative because webhook-specific 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 change the token setting for each test in the test case, we use @override_settings. This implies that we should not use the true sensitive token, that we should not keep in our code base or set a value in our test settings.

We use the enforce_csrf_checks flag in our test client to verify the @csrf_exempt decorator. If we intentionally remove the designer from the view, the test client will encounter a CSRF problem.

Before we examine the success case of the display, we examine its various states of failure. Checking the missing and incorrect token scenarios for coverage is unnecessary, but is done to perfection if the code changes.

We adjust the HTTPStatus enum response status codes from the Python device directory on a regular basis. As a result, we have to use the rather inconvenient HTTP_ * syntax to pass the Codeunderscored-Webhook-Token header.

Tests

We can use Django’s test client to query our webhook view to test it:

Summary

Webhooks are a simple way to alert external services when a specific event occurs. This is a common technique for web application to get data. With an HTTP request, the external system sends data to yours.

Proper receipt and processing of webhook data can be critical to the success of your application.

Source

LEAVE A REPLY

Please enter your comment!
Please enter your name here

Popular Articles