Bài 7.3: Authentication in DRF - Phần 3 - JWT Authentication

**Series Navigation:** - [Bài 7.1: Authentication Basics](./django-advance/authentication-drf-phan-1) - [Bài 7.2: Token Authentication Advanced](/django-advance/authentication-drf-phan-2) - **Bài 7.3 (bài này): JWT Authentication** 👈 - [Bài 7.4: OAuth2 & Social Auth](/django-advance/authentication-drf-phan-4)

Mục Tiêu Bài 7.3

Sau khi hoàn thành Bài 7.3 này, bạn sẽ:

✅ Hiểu JWT (JSON Web Tokens)
✅ Setup djangorestframework-simplejwt
✅ Generate và validate JWT tokens
✅ Implement access + refresh JWT
✅ Customize JWT claims
✅ Implement token blacklisting
✅ Apply JWT security best practices
✅ Test JWT authentication

JWT Overview

What is JWT?

JWT (JSON Web Token) = Self-contained token format.

Structure:

header.payload.signature eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJ1c2VybmFtZSI6ImpvaG4iLCJleHAiOjE2MzI0NTY3ODl9.abc123xyz456...

Three parts:

Header - Algorithm and token type
Payload - Claims (user data)
Signature - Verification signature

JWT vs Traditional Tokens

Traditional Token (database token):

Pros:✅ Can revoke immediately✅ Can track usage✅ Centralized control Cons:❌ Database lookup on every request❌ Not self-contained❌ Scalability issues

JWT:

Pros:✅ No database lookup needed✅ Self-contained (includes user data)✅ Stateless✅ Better scalability Cons:❌ Cannot revoke (unless blacklisted)❌ Larger size❌ Harder to invalidate

JWT Structure Decoded

Example JWT:

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJ1c2VybmFtZSI6ImpvaG4iLCJleHAiOjE2MzI0NTY3ODl9.abc123xyz456

Decoded:

Header:

{  "alg": "HS256",  "typ": "JWT"}

Payload (Claims):

{  "user_id": 1,  "username": "john",  "exp": 1632456789,  "iat": 1632453189,  "jti": "abc-123-def-456"}

Signature:

HMACSHA256(  base64UrlEncode(header) + "." + base64UrlEncode(payload),  secret_key)

Standard JWT Claims

Registered claims:

iss (Issuer) - Who issued the token
sub (Subject) - Who the token is about
aud (Audience) - Who should accept the token
exp (Expiration) - When token expires
nbf (Not Before) - Token not valid before this time
iat (Issued At) - When token was issued
jti (JWT ID) - Unique identifier

Custom claims:

{  "user_id": 1,  "username": "john",  "email": "john@example.com",  "role": "admin",  "permissions": ["read", "write"]}

djangorestframework-simplejwt

Installation

pip install djangorestframework-simplejwt

Basic Setup

# settings.pyINSTALLED_APPS = [    'rest_framework',    'rest_framework_simplejwt',] REST_FRAMEWORK = {    'DEFAULT_AUTHENTICATION_CLASSES': [        'rest_framework_simplejwt.authentication.JWTAuthentication',    ],} from datetime import timedelta SIMPLE_JWT = {    'ACCESS_TOKEN_LIFETIME': timedelta(minutes=5),    'REFRESH_TOKEN_LIFETIME': timedelta(days=1),    'ROTATE_REFRESH_TOKENS': False,    'BLACKLIST_AFTER_ROTATION': False,    'UPDATE_LAST_LOGIN': False,     'ALGORITHM': 'HS256',    'SIGNING_KEY': SECRET_KEY,    'VERIFYING_KEY': None,    'AUDIENCE': None,    'ISSUER': None,    'JWK_URL': None,    'LEEWAY': 0,     'AUTH_HEADER_TYPES': ('Bearer',),    'AUTH_HEADER_NAME': 'HTTP_AUTHORIZATION',    'USER_ID_FIELD': 'id',    'USER_ID_CLAIM': 'user_id',    'USER_AUTHENTICATION_RULE': 'rest_framework_simplejwt.authentication.default_user_authentication_rule',     'AUTH_TOKEN_CLASSES': ('rest_framework_simplejwt.tokens.AccessToken',),    'TOKEN_TYPE_CLAIM': 'token_type',     'JTI_CLAIM': 'jti',     'SLIDING_TOKEN_REFRESH_EXP_CLAIM': 'refresh_exp',    'SLIDING_TOKEN_LIFETIME': timedelta(minutes=5),    'SLIDING_TOKEN_REFRESH_LIFETIME': timedelta(days=1),}

URLs

# urls.pyfrom django.urls import pathfrom rest_framework_simplejwt.views import (    TokenObtainPairView,    TokenRefreshView,    TokenVerifyView,) urlpatterns = [    path('api/token/', TokenObtainPairView.as_view(), name='token_obtain_pair'),    path('api/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),    path('api/token/verify/', TokenVerifyView.as_view(), name='token_verify'),]

Usage

1. Obtain tokens (login):

curl -X POST http://localhost:8000/api/token/ \  -H "Content-Type: application/json" \  -d '{"username": "john", "password": "secret123"}'

Response:

{  "access": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...",  "refresh": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."}

2. Use access token:

curl http://localhost:8000/api/articles/ \  -H "Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."

3. Refresh access token:

curl -X POST http://localhost:8000/api/token/refresh/ \  -H "Content-Type: application/json" \  -d '{"refresh": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."}'

Response:

{  "access": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."}

4. Verify token:

curl -X POST http://localhost:8000/api/token/verify/ \  -H "Content-Type: application/json" \  -d '{"token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."}'

Custom JWT Claims

Customize Token Claims

Create custom view:

# views.pyfrom rest_framework_simplejwt.serializers import TokenObtainPairSerializerfrom rest_framework_simplejwt.views import TokenObtainPairView class CustomTokenObtainPairSerializer(TokenObtainPairSerializer):    @classmethod    def get_token(cls, user):        token = super().get_token(user)         # Add custom claims        token['username'] = user.username        token['email'] = user.email        token['is_staff'] = user.is_staff        token['full_name'] = user.get_full_name()                # Add more custom data        if hasattr(user, 'profile'):            token['avatar'] = user.profile.avatar_url            token['bio'] = user.profile.bio         return token class CustomTokenObtainPairView(TokenObtainPairView):    serializer_class = CustomTokenObtainPairSerializer

Update URLs:

# urls.pyfrom .views import CustomTokenObtainPairView urlpatterns = [    path('api/token/', CustomTokenObtainPairView.as_view(), name='token_obtain_pair'),]

Token payload now includes:

{  "token_type": "access",  "exp": 1632456789,  "iat": 1632453189,  "jti": "abc123",  "user_id": 1,  "username": "john",  "email": "john@example.com",  "is_staff": true,  "full_name": "John Doe",  "avatar": "https://...",  "bio": "Developer"}

Add Custom Response Data

# serializers.pyfrom rest_framework_simplejwt.serializers import TokenObtainPairSerializer class CustomTokenObtainPairSerializer(TokenObtainPairSerializer):    def validate(self, attrs):        data = super().validate(attrs)                # Add custom response data        data['user'] = {            'id': self.user.id,            'username': self.user.username,            'email': self.user.email,            'full_name': self.user.get_full_name(),            'is_staff': self.user.is_staff,        }                return data

Response:

{  "access": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...",  "refresh": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...",  "user": {    "id": 1,    "username": "john",    "email": "john@example.com",    "full_name": "John Doe",    "is_staff": true  }}

Accessing JWT Claims in Views

Get User from JWT

# views.pyfrom rest_framework.decorators import api_view, permission_classesfrom rest_framework.permissions import IsAuthenticatedfrom rest_framework.response import Response @api_view(['GET'])@permission_classes([IsAuthenticated])def profile(request):    # request.user is automatically set from JWT    user = request.user        # Access JWT token    token = request.auth  # JWT token object        return Response({        'user_id': user.id,        'username': user.username,        'email': user.email,        'token_jti': token['jti'],        'token_exp': token['exp'],    })

Custom Claims in View

@api_view(['GET'])@permission_classes([IsAuthenticated])def check_permission(request):    # Access custom claims from token    token = request.auth        is_admin = token.get('is_staff', False)    roles = token.get('roles', [])    permissions = token.get('permissions', [])        if 'admin' in roles:        return Response({'message': 'Admin access granted'})        return Response({'message': 'Regular user'})

Token Blacklisting

Why Blacklisting?

JWT problem: Can't invalidate tokens before expiration.

Solutions:

Short expiration (5-15 minutes)
Token blacklisting
Token versioning

Setup Blacklist

Install:

# Already included in djangorestframework-simplejwt

Configure:

# settings.pyINSTALLED_APPS = [    'rest_framework_simplejwt',    'rest_framework_simplejwt.token_blacklist',  # Add this] SIMPLE_JWT = {    'ROTATE_REFRESH_TOKENS': True,    'BLACKLIST_AFTER_ROTATION': True,    # ...}

Migrate:

python manage.py migrate

Creates tables:

token_blacklist_outstandingtokentoken_blacklist_blacklistedtoken

Logout with Blacklist

# views.pyfrom rest_framework.decorators import api_view, permission_classesfrom rest_framework.permissions import IsAuthenticatedfrom rest_framework.response import Responsefrom rest_framework import statusfrom rest_framework_simplejwt.tokens import RefreshToken @api_view(['POST'])@permission_classes([IsAuthenticated])def logout(request):    try:        refresh_token = request.data.get('refresh')        token = RefreshToken(refresh_token)        token.blacklist()                return Response({'message': 'Logout successful'}, status=status.HTTP_200_OK)    except Exception as e:        return Response({'error': str(e)}, status=status.HTTP_400_BAD_REQUEST)

Usage:

curl -X POST http://localhost:8000/api/logout/ \  -H "Authorization: Bearer <access_token>" \  -H "Content-Type: application/json" \  -d '{"refresh": "<refresh_token>"}'

Blacklist All User Tokens

# views.pyfrom rest_framework_simplejwt.token_blacklist.models import OutstandingToken, BlacklistedToken @api_view(['POST'])@permission_classes([IsAuthenticated])def logout_all_devices(request):    """Blacklist all refresh tokens for user"""    tokens = OutstandingToken.objects.filter(user=request.user)        for token in tokens:        BlacklistedToken.objects.get_or_create(token=token)        return Response({'message': 'Logged out from all devices'})

JWT Security Configuration

Recommended Settings

# settings.pyfrom datetime import timedelta SIMPLE_JWT = {    # Token lifetimes    'ACCESS_TOKEN_LIFETIME': timedelta(minutes=15),  # Short-lived    'REFRESH_TOKEN_LIFETIME': timedelta(days=7),     # Long-lived        # Rotation and blacklisting    'ROTATE_REFRESH_TOKENS': True,    'BLACKLIST_AFTER_ROTATION': True,        # Update last login    'UPDATE_LAST_LOGIN': True,        # Algorithm    'ALGORITHM': 'HS256',        # Signing key (use different key in production)    'SIGNING_KEY': SECRET_KEY,        # Token types    'AUTH_HEADER_TYPES': ('Bearer',),        # Claims    'USER_ID_FIELD': 'id',    'USER_ID_CLAIM': 'user_id',        # Sliding tokens (alternative to refresh)    'SLIDING_TOKEN_REFRESH_EXP_CLAIM': 'refresh_exp',    'SLIDING_TOKEN_LIFETIME': timedelta(minutes=5),    'SLIDING_TOKEN_REFRESH_LIFETIME': timedelta(days=1),}

Use Separate Signing Key

# settings.pyimport os # Separate JWT signing keyJWT_SIGNING_KEY = os.environ.get('JWT_SIGNING_KEY', SECRET_KEY) SIMPLE_JWT = {    'SIGNING_KEY': JWT_SIGNING_KEY,    # ...}

Audience and Issuer Claims

# settings.pySIMPLE_JWT = {    'ISSUER': 'https://myapi.example.com',    'AUDIENCE': ['https://myapp.example.com', 'https://mobile.example.com'],    # ...}

Advanced JWT Patterns

Sliding Tokens

Alternative to access + refresh:

# settings.pySIMPLE_JWT = {    'AUTH_TOKEN_CLASSES': ('rest_framework_simplejwt.tokens.SlidingToken',),    'SLIDING_TOKEN_LIFETIME': timedelta(minutes=5),    'SLIDING_TOKEN_REFRESH_LIFETIME': timedelta(days=1),} # urls.pyfrom rest_framework_simplejwt.views import (    TokenObtainSlidingView,    TokenRefreshSlidingView,) urlpatterns = [    path('api/token/', TokenObtainSlidingView.as_view()),    path('api/token/refresh/', TokenRefreshSlidingView.as_view()),]

Custom Token Class

# tokens.pyfrom rest_framework_simplejwt.tokens import RefreshToken class CustomRefreshToken(RefreshToken):    @classmethod    def for_user(cls, user):        token = super().for_user(user)                # Add custom claims        token['email'] = user.email        token['is_staff'] = user.is_staff                # Add permissions        token['permissions'] = list(user.user_permissions.values_list('codename', flat=True))                return token # views.pyfrom .tokens import CustomRefreshToken @api_view(['POST'])def custom_token(request):    username = request.data.get('username')    password = request.data.get('password')        user = authenticate(username=username, password=password)        if user:        refresh = CustomRefreshToken.for_user(user)        return Response({            'refresh': str(refresh),            'access': str(refresh.access_token),        })        return Response({'error': 'Invalid credentials'}, status=400)

Token Versioning

# models.pyclass UserProfile(models.Model):    user = models.OneToOneField(User, on_delete=models.CASCADE)    token_version = models.IntegerField(default=0)        def invalidate_tokens(self):        """Increment version to invalidate all tokens"""        self.token_version += 1        self.save() # tokens.pyclass VersionedRefreshToken(RefreshToken):    @classmethod    def for_user(cls, user):        token = super().for_user(user)        token['token_version'] = user.profile.token_version        return token # authentication.pyfrom rest_framework_simplejwt.authentication import JWTAuthentication class VersionedJWTAuthentication(JWTAuthentication):    def authenticate(self, request):        result = super().authenticate(request)                if result is None:            return None                user, validated_token = result                # Check token version        token_version = validated_token.get('token_version', 0)        current_version = user.profile.token_version                if token_version != current_version:            raise exceptions.AuthenticationFailed('Token version mismatch')                return (user, validated_token)

Complete JWT Example

Models

# models.pyfrom django.db import modelsfrom django.contrib.auth.models import User class UserProfile(models.Model):    user = models.OneToOneField(User, on_delete=models.CASCADE, related_name='profile')    avatar_url = models.URLField(blank=True)    bio = models.TextField(blank=True)    phone = models.CharField(max_length=20, blank=True)    token_version = models.IntegerField(default=0)        def __str__(self):        return f"{self.user.username}'s profile"

Serializers

# serializers.pyfrom rest_framework import serializersfrom rest_framework_simplejwt.serializers import TokenObtainPairSerializerfrom django.contrib.auth.models import User class UserSerializer(serializers.ModelSerializer):    class Meta:        model = User        fields = ['id', 'username', 'email', 'first_name', 'last_name'] class CustomTokenObtainPairSerializer(TokenObtainPairSerializer):    def validate(self, attrs):        data = super().validate(attrs)                # Add user data to response        data['user'] = UserSerializer(self.user).data                return data        @classmethod    def get_token(cls, user):        token = super().get_token(user)                # Add custom claims        token['username'] = user.username        token['email'] = user.email        token['is_staff'] = user.is_staff                if hasattr(user, 'profile'):            token['avatar'] = user.profile.avatar_url            token['token_version'] = user.profile.token_version                return token

Views

# views.pyfrom rest_framework import viewsetsfrom rest_framework.decorators import api_view, permission_classes, actionfrom rest_framework.permissions import IsAuthenticated, AllowAnyfrom rest_framework.response import Responsefrom rest_framework import statusfrom rest_framework_simplejwt.views import TokenObtainPairViewfrom rest_framework_simplejwt.tokens import RefreshTokenfrom .serializers import CustomTokenObtainPairSerializer, UserSerializer class CustomTokenObtainPairView(TokenObtainPairView):    serializer_class = CustomTokenObtainPairSerializer @api_view(['POST'])@permission_classes([IsAuthenticated])def logout(request):    """Blacklist refresh token"""    try:        refresh_token = request.data.get('refresh')        token = RefreshToken(refresh_token)        token.blacklist()        return Response({'message': 'Logout successful'})    except Exception as e:        return Response({'error': str(e)}, status=status.HTTP_400_BAD_REQUEST) @api_view(['GET'])@permission_classes([IsAuthenticated])def profile(request):    """Get current user profile"""    serializer = UserSerializer(request.user)    return Response(serializer.data) @api_view(['POST'])@permission_classes([IsAuthenticated])def invalidate_all_tokens(request):    """Invalidate all tokens by incrementing version"""    if hasattr(request.user, 'profile'):        request.user.profile.token_version += 1        request.user.profile.save()        return Response({'message': 'All tokens invalidated'})    return Response({'error': 'Profile not found'}, status=400)

URLs

# urls.pyfrom django.urls import pathfrom rest_framework_simplejwt.views import TokenRefreshView, TokenVerifyViewfrom . import views urlpatterns = [    # JWT endpoints    path('api/token/', views.CustomTokenObtainPairView.as_view(), name='token_obtain_pair'),    path('api/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),    path('api/token/verify/', TokenVerifyView.as_view(), name='token_verify'),        # Auth endpoints    path('api/logout/', views.logout, name='logout'),    path('api/profile/', views.profile, name='profile'),    path('api/invalidate-tokens/', views.invalidate_all_tokens, name='invalidate_tokens'),]

Testing JWT Authentication

# tests.pyfrom rest_framework.test import APITestCasefrom django.contrib.auth.models import Userfrom rest_framework_simplejwt.tokens import RefreshToken class JWTAuthenticationTest(APITestCase):    def setUp(self):        self.user = User.objects.create_user('testuser', 'test@test.com', 'password')        self.refresh = RefreshToken.for_user(self.user)        self.access = str(self.refresh.access_token)        def test_obtain_token(self):        """Test obtaining JWT tokens"""        response = self.client.post('/api/token/', {            'username': 'testuser',            'password': 'password'        })                self.assertEqual(response.status_code, 200)        self.assertIn('access', response.data)        self.assertIn('refresh', response.data)        def test_access_with_token(self):        """Test accessing protected endpoint with JWT"""        self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access}')        response = self.client.get('/api/profile/')                self.assertEqual(response.status_code, 200)        self.assertEqual(response.data['username'], 'testuser')        def test_refresh_token(self):        """Test refreshing access token"""        response = self.client.post('/api/token/refresh/', {            'refresh': str(self.refresh)        })                self.assertEqual(response.status_code, 200)        self.assertIn('access', response.data)        def test_blacklist_token(self):        """Test token blacklisting"""        # Logout (blacklist)        response = self.client.post('/api/logout/', {            'refresh': str(self.refresh)        })        self.assertEqual(response.status_code, 200)                # Try to refresh with blacklisted token        response = self.client.post('/api/token/refresh/', {            'refresh': str(self.refresh)        })        self.assertEqual(response.status_code, 401)        def test_expired_token(self):        """Test expired token is rejected"""        # Create expired token        from datetime import timedelta        from django.utils import timezone                refresh = RefreshToken.for_user(self.user)        refresh.set_exp(lifetime=timedelta(seconds=-1))  # Already expired        access = str(refresh.access_token)                self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {access}')        response = self.client.get('/api/profile/')                self.assertEqual(response.status_code, 401)

Tóm Tắt Bài 7.3

Trong Bài 7.3 này, bạn đã học:

✅ JWT Basics:

JWT structure (header.payload.signature)
JWT vs traditional tokens
Standard JWT claims

✅ djangorestframework-simplejwt:

Installation và setup
Obtain, refresh, verify tokens
Configuration options

✅ Custom JWT Claims:

Add custom claims to tokens
Customize response data
Access claims in views

✅ Token Blacklisting:

Why blacklisting is needed
Setup blacklist
Logout implementation
Blacklist all user tokens

✅ Security Best Practices:

Token lifetimes
Separate signing keys
Audience and issuer claims
Token rotation

✅ Advanced Patterns:

Sliding tokens
Custom token classes
Token versioning
Complete example

✅ Testing:

Test token obtain
Test authentication
Test refresh
Test blacklisting

Next Steps

Trong Bài 7.4, bạn sẽ học:

OAuth2 fundamentals
Social authentication (Google, Facebook, GitHub)
django-allauth integration
dj-rest-auth với social auth
Custom OAuth2 providers
Security considerations

Navigation:

← Bài 7.2: Token Authentication Advanced
Bài 7.3: JWT Authentication (current)
Bài 7.4: OAuth2 & Social Auth →

#jwt

Authentication in DRF - Phần 3 - JWT Authentication

OAuth2, JWT, và Social Authentication