[PYTHON] Implement JWT login functionality in Django REST framework

Articles up to the last time

1. Django REST framework + Vue.js "Let's make an EC site like SEIYU" 6 times (planned)
2. Create a SEIYU-style EC site (1) Requirements analysis and project initialization
3. Let's create a SEIYU-style EC site (2) Renew the management screen using Xadmin
4. Create a SEIYU-style EC site (3) Create an explosive API with the Django REST framework
5. Create a SEIYU-style EC site (4) Use Vue.js project initialization TypeScript

Even if you don't look at the contents so far, you can understand the contents of this time, so I hope you can keep in touch with me until the end. : relaxed:

What is JWT

It is an abbreviation of JSON Web Token, it is possible to hold arbitrary information (complaint) in the token, for example, the server generates a token containing the information "logged in as an administrator" to the client be able to. The client can use the token to prove that he is logged in as an administrator. -[Wikipedia] (https://ja.wikipedia.org/wiki/JSON_Web_Token)

Implemented by hand

You can easily implement JWT login by using the library django-rest-framework-jwt, It is easier to customize in various ways if you write it directly, so write it directly.

• If anyone wants to know how to use it, I will add it again. : point_up_tone1:

Create a new Django REST framework project

pip install Django
pip install djangorestframework
pip install markdown
pip install django-filter

django-admin startproject jwttest

cd jwttest
python manage.py runserver

If you start the server and see the rocket normally, it's OK.

Initial setting

Create a new app.

python manage.py startapp api

Add it to ʻINSTALLED_APPS along with rest_framework`.

jwttest/settings.py

INSTALLED_APPS = [
    ...
    'rest_framework',
    'api'
]

Create a user model for login.

jwtest/api/models.py

from django.db import models


class UserInfo(models.Model):
    username = models.CharField(max_length=50, unique=True, db_index=True)
    password = models.CharField(max_length=100, db_index=True)
    info = models.CharField(max_length=200)

Execute DB migration.

python manage.py makemigrations
python manage.py migrate

Implement login function

Install the library for JWT token generation.

pip install pyjwt

Add a class for login, Add the ʻutls folder under the ʻapi directory, and add a new ʻauth.py` file inside.

api/utils/auth.py

import time
import jwt
from jwttest.settings import SECRET_KEY
from rest_framework.authentication import BaseAuthentication
from rest_framework import exceptions

from api.models import UserInfo

class NormalAuthentication(BaseAuthentication):
    def authenticate(self, request):
        username = request._request.POST.get("username")
        password = request._request.POST.get("password")
        user_obj = UserInfo.objects.filter(username=username).first()
        if not user_obj:
            raise exceptions.AuthenticationFailed('Authentication failure')
        elif user_obj.password != password:
            raise exceptions.AuthenticationFailed('I don't have a password')
        token = generate_jwt(user_obj)
        return (token, None)

    def authenticate_header(self, request):
        pass

#Generate a Token with the jwt library you just installed
#The contents of the Token include user information and a timeout
#It is fixed that the timeout key is exp
#document: https://pyjwt.readthedocs.io/en/latest/usage.html?highlight=exp
def generate_jwt(user):
    timestamp = int(time.time()) + 60*60*24*7
    return jwt.encode(
        {"userid": user.pk, "username": user.username, "info": user.info, "exp": timestamp},
        SECRET_KEY).decode("utf-8")

Also add a view for login. If the login is successful, JWT is returned. Add the NormalAuthentication created earlier to ʻauthentication_classes`.

api/views.py

from rest_framework.views import APIView
from rest_framework.response import Response

from .utils.auth import NormalAuthentication


class Login(APIView):

    authentication_classes = [NormalAuthentication,]

    def post(self, request, *args, **kwargs):
        return Response({"token": request.user})

Add the url.

jwttest/urls.py

...
from api.views import Login

urlpatterns = [
    ...
    path('login/', Login.as_view()),
]

Add one user information

Start the server and try logging in. Let's parse the returned JWT with https://jwt.io/. The JWT contains the information you specified. Let's create a view that can only be seen by logging in and access it using this Token. First, add the authentication class for JWT to ʻapi / utils / auth.py`.

api/utils/auth.py

...
class JWTAuthentication(BaseAuthentication):
    keyword = 'JWT'
    model = None

    def authenticate(self, request):
        auth = get_authorization_header(request).split()

        if not auth or auth[0].lower() != self.keyword.lower().encode():
            return None

        if len(auth) == 1:
            msg = "Authorization disabled"
            raise exceptions.AuthenticationFailed(msg)
        elif len(auth) > 2:
            msg = "Authorization No invalid space"
            raise exceptions.AuthenticationFailed(msg)

        try:
            jwt_token = auth[1]
            jwt_info = jwt.decode(jwt_token, SECRET_KEY)
            userid = jwt_info.get("userid")
            try:
                user = UserInfo.objects.get(pk=userid)
                user.is_authenticated = True
                return (user, jwt_token)
            except:
                msg = "User does not exist"
                raise exceptions.AuthenticationFailed(msg)
        except jwt.ExpiredSignatureError:
            msg = "token timed out"
            raise exceptions.AuthenticationFailed(msg)

    def authenticate_header(self, request):
        pass
...

Added views to ʻapi / views.py` that can only be accessed by logging in.

api/views.py

...
from rest_framework.permissions import IsAuthenticated
...
class Something(APIView):
    authentication_classes = [JWTAuthentication, ]
    #Make it accessible only to logged-in users.
    permission_classes = [IsAuthenticated, ]

    def get(self, request, *args, **kwargs):
        return Response({"data": "It is the contents"})
...

I will try to access it after adding the url.

python:jwttest:urls.py

    path('data/', Something.as_view())

First, access without Token. It returned Authentication credentials were not provided.

After adding the Token, I was able to access it.

That's it, but I'd like to include some analysis of the Django REST framework login-related source code. Please read if you are interested. : relaxed:

DRF login source code analysis

Link with the authentication class to use

I will analyze it based on the code I wrote earlier, CBV (Class-based views) If used, dispatch will be executed. With that as the entrance, we will look at the source code. As a way of looking at it, add self.dispatch () to the login class I wrote earlier.

api/view.py

...
class Login(APIView):
    authentication_classes = [NormalAuthentication, ]
    def post(self, request, *args, **kwargs):
        #Add and follow the source code
        #When using PyCharm
        #command on mac+click
        #Alt in win+Should be a click
        self.dispatch() 
        return Response({"token": request.user})
...

The destination will be def dispatch (self, request, * args, ** kwargs): on line 481 of rest_framework / views.py. Let's look at the contents of the ʻinitialize_request` function on line 488.

rest_framework/views.py

    def dispatch(self, request, *args, **kwargs):
        ...
        #Let's look at the contents of this function
        request = self.initialize_request(request, *args, **kwargs)
        self.request = request

The definition of the contents is on line 381 of rest_framework / views.py.

rest_framework/views.py

    def initialize_request(self, request, *args, **kwargs):
        """
        Returns the initial request object.
        """
        parser_context = self.get_parser_context(request)

        return Request(
            request,
            parsers=self.get_parsers(),
            #here
            authenticators=self.get_authenticators(),
            negotiator=self.get_content_negotiator(),
            parser_context=parser_context
        )

And if you look at the contents of self.get_authenticators () on line 390, I have the following code, which is taken from self.authentication_classes to find out what authentication class the CBV is using.

rest_framework/views.py

    def get_authenticators(self):
        """
        Instantiates and returns the list of authenticators that this view can use.
        """
        return [auth() for auth in self.authentication_classes]

If you follow the definition of self.authentication_classes, Line 109 of rest_framework / views.py has the following definition:

rest_framework/views.py

class APIView(View):
   ...
   authentication_classes = api_settings.DEFAULT_AUTHENTICATION_CLASSES

Therefore, there are two places where you can directly define what to use for the authentication class.

1. Inside the CBV class inherited from ʻAPIView`

api/views.py

...
class Login(APIView):
    #here
    authentication_classes = [NormalAuthentication, ]

    def post(self, request, *args, **kwargs):
        return Response({"token": request.user})
...

2. Inside the settings for REST_FRAMEWORK in settings.py

REST_FRAMEWORK = {
  'DEFAULT_AUTHENTICATION_CLASSES': ['api.utils.auth.NormalAuthentication']
}

Once you understand the relationship between CBV and the authentication class, it's time to look at the features of the authentication class. We will also follow from dispatch,self.initial (request, * args, ** kwargs)on line 493 of rest_framework / views.py.

rest_framework/views.py

    def dispatch(self, request, *args, **kwargs):
        ...

        try:
            self.initial(request, *args, **kwargs)
        ...

I will follow the definition of it. Line 395 of rest_framework / views.py. There is perform_authentication, we will go further.

python:rest_framework.py/views.py

...
    def initial(self, request, *args, **kwargs):
        ...
        self.perform_authentication(request)
        ...

Beyond that is request.user.

rest_framework/views.py

...
    def perform_authentication(self, request):
        """
        Perform authentication on the incoming request.

        Note that if you override this and simply 'pass', then authentication
        will instead be performed lazily, the first time either
        `request.user` or `request.auth` is accessed.
        """
        request.user
...

Its definition is on line 213 of rest_framework / request.py.

rest_framework/request.py

...
    @property
    def user(self):
        """
        Returns the user associated with the current request, as authenticated
        by the authentication classes provided to the request.
        """
        if not hasattr(self, '_user'):
            with wrap_attributeerrors():
                self._authenticate()
        return self._user
...

Let's take a look at the definition of self._authenticate (). Line 366 of rest_framework / request.py. The content is to take a class from the CBV's authentication class list and execute the ʻauthenticatemethod for that class. The result of the execution is atuple with two elements, the first tuple is self.userand the second isself.auth`.

rest_framework/request.py

    def _authenticate(self):
        """
        Attempt to authenticate the request using each authentication instance
        in turn.
        """
        for authenticator in self.authenticators:
            try:
                user_auth_tuple = authenticator.authenticate(self)
            except exceptions.APIException:
                self._not_authenticated()
                raise

            if user_auth_tuple is not None:
                self._authenticator = authenticator
                self.user, self.auth = user_auth_tuple
                return

        self._not_authenticated()

With that in mind, let's take a look at the authentication class for JWT defined earlier.

api/utils/auth.py

class JWTAuthentication(BaseAuthentication):
    keyword = 'JWT'
    model = None

    def authenticate(self, request):
        auth = get_authorization_header(request).split()

        if not auth or auth[0].lower() != self.keyword.lower().encode():
            return None

        if len(auth) == 1:
            msg = "Authorization disabled"
            raise exceptions.AuthenticationFailed(msg)
        elif len(auth) > 2:
            msg = "Authorization No invalid space"
            raise exceptions.AuthenticationFailed(msg)

        try:
            jwt_token = auth[1]
            jwt_info = jwt.decode(jwt_token, SECRET_KEY)
            userid = jwt_info.get("userid")
            try:
                user = UserInfo.objects.get(pk=userid)
                user.is_authenticated = True
                return (user, jwt_token)
            except:
                msg = "User does not exist"
                raise exceptions.AuthenticationFailed(msg)
        except jwt.ExpiredSignatureError:
            msg = "token is timeout"
            raise exceptions.AuthenticationFailed(msg)

    def authenticate_header(self, request):
        pass

The class contains ʻauthenticate` method. If the authentication is successful, a tuple containing user information is returned. This concludes the analysis of login-related sources.

Thank you for staying with us until the end. : raised_hand_tone1: