Authenticating Background Tasks with Django Rest Framework SimpleJWT

Situation

You have an API built with Django Rest Framework
You are using SimpleJWT as your authentication backend
You have integrated Celery with your Django application and are using it to run background tasks
Some of your tasks need to make authenticated API calls

Possible Solution

You can authenticate your task using the SimpleJWT authentication endpoints, just like a normal user

Alternate Solution

Leverage the SimpleJWT API (not HTTP endpoint) to authenticate the task

Example

from celery import shared_task
from django.contrib.auth import get_user_model
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer 
import os

@shared_task
def some_task():
    username = os.getenv("SVC_ACCT_USER", "")
    password = os.getenv("SVC_ACCT_PASS", "")
    endpoint = "some_url"

    svc_acct = {
        f"{get_user_model().USERNAME_FIELD}": username,
        "password": password,
    }

    TokenGenerator = TokenObtainPairSerializer(data=svc_acct)
    TokenGenerator.is_valid()

    headers = {
        "Authorization": f"JWT {TokenGenerator.validated_data['access']}"
    }

    res = requests.get(url=endpoint, headers=headers)

    # Logic based on Response

    return foo

Explanation

We get the username and password from environmental variables. These may be set in production by injecting secrets into the environment using the CICD pipeline, a .env file, manually, and so on. The point here is that we're not hardcoding credentials.
We create a dictionary with the credentials for the service account. In Django, it's a common practice to write a custom user model. Therefore, we're dynamically setting the username field based on the user model being used (rather than hardcoding it). This eliminates potential tech debt if the user model ever changes, or if we use this task in multiple applications.
Next, we initialize a TokenGenerator object from the TokenObtainPairSerializer in the SimpleJWT module. We pass the service account dictionary as the data keyword argument when initializing the TokenObtainPairSerializer class.
As a best practice, we validate the data using the TokenObtainPairSerializer class' validator.
We then create an Authentication header using the access token of the TokenGenerator class.
Finally, we make our request to the API and execute the remaining business logic.

Improvements

Currently, the authentication happens every time the task is run. If this is a periodic task (for example, a crontab) it may make sense to create a SvcAccount class and encapsulate the authentication logic inside of it. This object can be initialized when the application is started, and then the tokens can simply be refreshed when they're expired.
Depending on your use case, it may make sense to not make HTTP endpoint requests at all. Instead, you can have the task execute the desired logic encapsulated in the HTTP endpoint. This may vary depending on security and operational needs. That is to say, your security policy may expect HTTP logs anytime this logic is executed. In that case, you should probably execute HTTP requests. Otherwise, it might be better to execute the logic and run logger statements locally inside the task.

Authenticating Background Tasks with Django Rest Framework SimpleJWT

Comments

More from this blog

Testing API Key Protected Endpoints in Django

A Practical Application of dnstwist

What's the difference between "is" and "==" in Python?

My Non-Linear Journey Into Information Security

Situation

Possible Solution

Alternate Solution

Example

Explanation

Improvements

Command Palette

Comments

More from this blog

Situation

Possible Solution

Alternate Solution

Example

Explanation

Improvements