Bài 7.2: Authentication in DRF - Phần 2 - Token Authentication Advanced

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

Mục Tiêu Bài 7.2

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

✅ Implement custom token models
✅ Add token expiration
✅ Handle token refresh
✅ Support multiple tokens per user
✅ Implement token revocation
✅ Apply security best practices
✅ Build custom authentication classes
✅ Test advanced authentication

Token Expiration

Why Token Expiration?

Problems with DRF's default TokenAuthentication:

❌ Tokens never expire
❌ Compromised tokens valid forever
❌ No refresh mechanism
❌ Single token per user

Solution: Add expiration

Custom Token Model

Create custom token with expiration:

# models.pyfrom django.db import modelsfrom django.conf import settingsfrom django.utils import timezonefrom datetime import timedeltaimport binasciiimport os class ExpiringToken(models.Model):    """    Token with expiration    """    key = models.CharField(max_length=40, primary_key=True)    user = models.ForeignKey(        settings.AUTH_USER_MODEL,        related_name='auth_tokens',        on_delete=models.CASCADE    )    created = models.DateTimeField(auto_now_add=True)    expires = models.DateTimeField()        class Meta:        verbose_name = "Token"        verbose_name_plural = "Tokens"        def save(self, *args, **kwargs):        if not self.key:            self.key = self.generate_key()        if not self.expires:            # Default: 7 days expiration            self.expires = timezone.now() + timedelta(days=7)        return super().save(*args, **kwargs)        @staticmethod    def generate_key():        return binascii.hexlify(os.urandom(20)).decode()        def is_expired(self):        return timezone.now() >= self.expires        def __str__(self):        return self.key

Migration:

python manage.py makemigrationspython manage.py migrate

Custom Authentication Class

# authentication.pyfrom rest_framework import exceptionsfrom rest_framework.authentication import TokenAuthentication as BaseTokenAuthenticationfrom django.utils import timezonefrom .models import ExpiringToken class ExpiringTokenAuthentication(BaseTokenAuthentication):    """    Token authentication with expiration    """    model = ExpiringToken        def authenticate_credentials(self, key):        try:            token = self.model.objects.select_related('user').get(key=key)        except self.model.DoesNotExist:            raise exceptions.AuthenticationFailed('Invalid token')                if not token.user.is_active:            raise exceptions.AuthenticationFailed('User inactive or deleted')                # Check expiration        if token.is_expired():            raise exceptions.AuthenticationFailed('Token has expired')                return (token.user, token)

Configure:

# settings.pyREST_FRAMEWORK = {    'DEFAULT_AUTHENTICATION_CLASSES': [        'myapp.authentication.ExpiringTokenAuthentication',    ]}

# views.pyfrom django.contrib.auth import authenticatefrom rest_framework.decorators import api_view, permission_classesfrom rest_framework.permissions import AllowAnyfrom rest_framework.response import Responsefrom datetime import timedeltafrom django.utils import timezonefrom .models import ExpiringToken @api_view(['POST'])@permission_classes([AllowAny])def login(request):    username = request.data.get('username')    password = request.data.get('password')        user = authenticate(username=username, password=password)        if not user:        return Response({'error': 'Invalid credentials'}, status=400)        # Delete old tokens    ExpiringToken.objects.filter(user=user).delete()        # Create new token    token = ExpiringToken.objects.create(user=user)        return Response({        'token': token.key,        'expires': token.expires,        'user': {            'id': user.id,            'username': user.username,            'email': user.email,        }    })

Token Refresh

Why Token Refresh?

Access Token + Refresh Token pattern:

Access Token:  Short-lived (15 min - 1 hour)Refresh Token: Long-lived (7 days - 30 days) Flow:1. Login → Get access + refresh tokens2. Use access token for API requests3. Access token expires → Use refresh token to get new access token4. Refresh token expires → Login again

Benefits:

✅ Short-lived access tokens (more secure)
✅ Long-lived refresh tokens (better UX)
✅ Can revoke refresh tokens
✅ Separate concerns

Dual Token Model

# models.pyfrom django.db import modelsfrom django.conf import settingsfrom django.utils import timezonefrom datetime import timedeltaimport secrets class AccessToken(models.Model):    """Short-lived access token"""    key = models.CharField(max_length=40, primary_key=True)    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)    created = models.DateTimeField(auto_now_add=True)    expires = models.DateTimeField()        def save(self, *args, **kwargs):        if not self.key:            self.key = secrets.token_urlsafe(30)        if not self.expires:            # 1 hour expiration            self.expires = timezone.now() + timedelta(hours=1)        return super().save(*args, **kwargs)        def is_expired(self):        return timezone.now() >= self.expires class RefreshToken(models.Model):    """Long-lived refresh token"""    key = models.CharField(max_length=40, primary_key=True)    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)    access_token = models.OneToOneField(AccessToken, on_delete=models.CASCADE, null=True)    created = models.DateTimeField(auto_now_add=True)    expires = models.DateTimeField()    revoked = models.BooleanField(default=False)        def save(self, *args, **kwargs):        if not self.key:            self.key = secrets.token_urlsafe(30)        if not self.expires:            # 7 days expiration            self.expires = timezone.now() + timedelta(days=7)        return super().save(*args, **kwargs)        def is_expired(self):        return timezone.now() >= self.expires        def is_valid(self):        return not self.revoked and not self.is_expired()

Authentication Class for Dual Tokens

# authentication.pyfrom rest_framework import exceptionsfrom rest_framework.authentication import BaseAuthenticationfrom .models import AccessToken class AccessTokenAuthentication(BaseAuthentication):    """    Authenticate with access token    """    def authenticate(self, request):        auth_header = request.META.get('HTTP_AUTHORIZATION', '')                if not auth_header.startswith('Bearer '):            return None                token_key = auth_header[7:]  # Remove 'Bearer '                try:            token = AccessToken.objects.select_related('user').get(key=token_key)        except AccessToken.DoesNotExist:            raise exceptions.AuthenticationFailed('Invalid token')                if not token.user.is_active:            raise exceptions.AuthenticationFailed('User inactive')                if token.is_expired():            raise exceptions.AuthenticationFailed('Token expired')                return (token.user, token)

# views.pyfrom django.contrib.auth import authenticatefrom rest_framework.decorators import api_view, permission_classesfrom rest_framework.permissions import AllowAny, IsAuthenticatedfrom rest_framework.response import Responsefrom rest_framework import statusfrom .models import AccessToken, RefreshToken @api_view(['POST'])@permission_classes([AllowAny])def login(request):    """Login and get access + refresh tokens"""    username = request.data.get('username')    password = request.data.get('password')        user = authenticate(username=username, password=password)        if not user:        return Response({'error': 'Invalid credentials'}, status=400)        # Create access token    access_token = AccessToken.objects.create(user=user)        # Create refresh token    refresh_token = RefreshToken.objects.create(        user=user,        access_token=access_token    )        return Response({        'access_token': access_token.key,        'refresh_token': refresh_token.key,        'expires_in': 3600,  # 1 hour        'token_type': 'Bearer',        'user': {            'id': user.id,            'username': user.username,            'email': user.email,        }    }) @api_view(['POST'])@permission_classes([AllowAny])def refresh_token(request):    """Refresh access token using refresh token"""    refresh_key = request.data.get('refresh_token')        if not refresh_key:        return Response({'error': 'refresh_token required'}, status=400)        try:        refresh = RefreshToken.objects.select_related('user').get(key=refresh_key)    except RefreshToken.DoesNotExist:        return Response({'error': 'Invalid refresh token'}, status=400)        # Check if refresh token is valid    if not refresh.is_valid():        return Response({'error': 'Refresh token expired or revoked'}, status=400)        # Delete old access token    if refresh.access_token:        refresh.access_token.delete()        # Create new access token    access_token = AccessToken.objects.create(user=refresh.user)        # Update refresh token    refresh.access_token = access_token    refresh.save()        return Response({        'access_token': access_token.key,        'expires_in': 3600,        'token_type': 'Bearer',    }) @api_view(['POST'])@permission_classes([IsAuthenticated])def logout(request):    """Logout: revoke refresh token and delete access token"""    # Delete access token    if hasattr(request, 'auth') and isinstance(request.auth, AccessToken):        access_token = request.auth                # Revoke refresh token        try:            refresh = RefreshToken.objects.get(access_token=access_token)            refresh.revoked = True            refresh.save()        except RefreshToken.DoesNotExist:            pass                # Delete access token        access_token.delete()        return Response({'message': 'Logged out successfully'})

URLs:

# urls.pyfrom django.urls import pathfrom . import views urlpatterns = [    path('auth/login/', views.login),    path('auth/refresh/', views.refresh_token),    path('auth/logout/', views.logout),]

Usage:

# 1. Logincurl -X POST http://localhost:8000/api/auth/login/ \  -H "Content-Type: application/json" \  -d '{"username": "john", "password": "secret"}' # Response:# {#   "access_token": "abc123...",#   "refresh_token": "xyz789...",#   "expires_in": 3600# } # 2. Use access tokencurl http://localhost:8000/api/articles/ \  -H "Authorization: Bearer abc123..." # 3. When access token expires, refresh itcurl -X POST http://localhost:8000/api/auth/refresh/ \  -H "Content-Type: application/json" \  -d '{"refresh_token": "xyz789..."}' # Response:# {#   "access_token": "def456...",#   "expires_in": 3600# }

Multiple Tokens Per User

Why Multiple Tokens?

Use cases:

Different devices (phone, tablet, web)
Different clients (mobile app, web app)
Selective revocation

Token Model with Device Info

# models.pyfrom django.db import modelsfrom django.conf import settingsimport secrets class UserToken(models.Model):    """Multiple tokens per user"""    key = models.CharField(max_length=40, unique=True)    user = models.ForeignKey(        settings.AUTH_USER_MODEL,        related_name='tokens',        on_delete=models.CASCADE    )    device_name = models.CharField(max_length=100, blank=True)    device_type = models.CharField(max_length=20, blank=True)  # ios, android, web    ip_address = models.GenericIPAddressField(null=True, blank=True)    user_agent = models.TextField(blank=True)    created = models.DateTimeField(auto_now_add=True)    last_used = models.DateTimeField(auto_now=True)    expires = models.DateTimeField()    revoked = models.BooleanField(default=False)        class Meta:        ordering = ['-last_used']        def save(self, *args, **kwargs):        if not self.key:            self.key = secrets.token_urlsafe(32)        if not self.expires:            from datetime import timedelta            from django.utils import timezone            self.expires = timezone.now() + timedelta(days=30)        return super().save(*args, **kwargs)        def is_valid(self):        from django.utils import timezone        return not self.revoked and self.expires > timezone.now()        def __str__(self):        return f"{self.user.username} - {self.device_name or 'Unknown'}"

# views.py@api_view(['POST'])@permission_classes([AllowAny])def login(request):    username = request.data.get('username')    password = request.data.get('password')    device_name = request.data.get('device_name', '')    device_type = request.data.get('device_type', '')        user = authenticate(username=username, password=password)        if not user:        return Response({'error': 'Invalid credentials'}, status=400)        # Get client info    ip_address = get_client_ip(request)    user_agent = request.META.get('HTTP_USER_AGENT', '')        # Create token    token = UserToken.objects.create(        user=user,        device_name=device_name,        device_type=device_type,        ip_address=ip_address,        user_agent=user_agent    )        return Response({        'token': token.key,        'expires': token.expires,        'user': {            'id': user.id,            'username': user.username,            'email': user.email,        }    }) def get_client_ip(request):    """Get client IP address"""    x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')    if x_forwarded_for:        ip = x_forwarded_for.split(',')[0]    else:        ip = request.META.get('REMOTE_ADDR')    return ip

List and Revoke Tokens

# views.pyfrom rest_framework.decorators import api_viewfrom rest_framework.permissions import IsAuthenticatedfrom rest_framework.response import Response @api_view(['GET'])@permission_classes([IsAuthenticated])def list_tokens(request):    """List all user's tokens"""    tokens = UserToken.objects.filter(        user=request.user,        revoked=False    ).values(        'key',        'device_name',        'device_type',        'created',        'last_used',        'ip_address'    )        return Response({'tokens': list(tokens)}) @api_view(['POST'])@permission_classes([IsAuthenticated])def revoke_token(request):    """Revoke specific token"""    token_key = request.data.get('token')        if not token_key:        return Response({'error': 'token required'}, status=400)        try:        token = UserToken.objects.get(user=request.user, key=token_key)        token.revoked = True        token.save()        return Response({'message': 'Token revoked'})    except UserToken.DoesNotExist:        return Response({'error': 'Token not found'}, status=404) @api_view(['POST'])@permission_classes([IsAuthenticated])def revoke_all_tokens(request):    """Revoke all tokens except current"""    current_token = request.auth        UserToken.objects.filter(        user=request.user,        revoked=False    ).exclude(key=current_token.key).update(revoked=True)        return Response({'message': 'All other tokens revoked'})

Serializer for token list:

# serializers.pyfrom rest_framework import serializers class TokenSerializer(serializers.Serializer):    key = serializers.CharField()    device_name = serializers.CharField()    device_type = serializers.CharField()    created = serializers.DateTimeField()    last_used = serializers.DateTimeField()    ip_address = serializers.IPAddressField()    is_current = serializers.BooleanField()

Token Security Best Practices

1. Use HTTPS Always

# settings.pyif not DEBUG:    SECURE_SSL_REDIRECT = True    SESSION_COOKIE_SECURE = True    CSRF_COOKIE_SECURE = True

2. Hash Tokens in Database

# models.pyimport hashlibimport secrets class HashedToken(models.Model):    hashed_key = models.CharField(max_length=64, unique=True)    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)    prefix = models.CharField(max_length=8)  # For identification    created = models.DateTimeField(auto_now_add=True)        @staticmethod    def generate_token():        """Generate token and return (token, hashed)"""        token = secrets.token_urlsafe(32)        hashed = hashlib.sha256(token.encode()).hexdigest()        prefix = token[:8]        return token, hashed, prefix        @classmethod    def create_for_user(cls, user):        token, hashed, prefix = cls.generate_token()        cls.objects.create(user=user, hashed_key=hashed, prefix=prefix)        return token  # Return unhashed token to client        @classmethod    def get_user_from_token(cls, token):        hashed = hashlib.sha256(token.encode()).hexdigest()        try:            token_obj = cls.objects.select_related('user').get(hashed_key=hashed)            return token_obj.user        except cls.DoesNotExist:            return None

3. Rate Limiting

# throttling.pyfrom rest_framework.throttling import AnonRateThrottle, UserRateThrottle class LoginRateThrottle(AnonRateThrottle):    rate = '5/hour'  # 5 login attempts per hour # views.pyfrom rest_framework.decorators import throttle_classes @api_view(['POST'])@permission_classes([AllowAny])@throttle_classes([LoginRateThrottle])def login(request):    # Login logic    pass

4. Token Rotation

# Rotate token on each use (optional, more secure but complex)def rotate_token(user, old_token):    # Delete old token    old_token.delete()        # Create new token    new_token = AccessToken.objects.create(user=user)        return new_token

5. IP Whitelisting

# models.pyclass IPWhitelistedToken(models.Model):    key = models.CharField(max_length=40, unique=True)    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)    allowed_ips = models.JSONField(default=list)  # ['192.168.1.1', '10.0.0.1']        def is_ip_allowed(self, ip):        return ip in self.allowed_ips # authentication.pyclass IPWhitelistedAuthentication(BaseAuthentication):    def authenticate(self, request):        # ... get token ...                client_ip = get_client_ip(request)        if not token.is_ip_allowed(client_ip):            raise exceptions.AuthenticationFailed('IP not allowed')                return (token.user, token)

6. Automatic Token Cleanup

# management/commands/cleanup_tokens.pyfrom django.core.management.base import BaseCommandfrom django.utils import timezonefrom myapp.models import AccessToken, RefreshToken class Command(BaseCommand):    help = 'Clean up expired tokens'        def handle(self, *args, **options):        now = timezone.now()                # Delete expired access tokens        deleted_access = AccessToken.objects.filter(expires__lt=now).delete()                # Delete expired/revoked refresh tokens        deleted_refresh = RefreshToken.objects.filter(            models.Q(expires__lt=now) | models.Q(revoked=True)        ).delete()                self.stdout.write(            f'Deleted {deleted_access[0]} access tokens, '            f'{deleted_refresh[0]} refresh tokens'        )

Cron job:

# Run daily at 2 AM0 2 * * * cd /path/to/project && python manage.py cleanup_tokens

Custom Authentication Class

Complete Custom Authentication

# authentication.pyfrom rest_framework import exceptionsfrom rest_framework.authentication import BaseAuthenticationfrom django.utils import timezonefrom .models import UserToken class CustomTokenAuthentication(BaseAuthentication):    """    Custom token authentication with:    - Expiration check    - Revocation check    - Last used tracking    - IP validation    """    keyword = 'Bearer'    model = UserToken        def authenticate(self, request):        auth_header = self.get_authorization_header(request)                if not auth_header:            return None                token_key = self.get_token_from_header(auth_header)                if not token_key:            return None                return self.authenticate_credentials(token_key, request)        def get_authorization_header(self, request):        """Get Authorization header"""        return request.META.get('HTTP_AUTHORIZATION', '')        def get_token_from_header(self, auth_header):        """Extract token from header"""        parts = auth_header.split()                if len(parts) == 0:            return None                if parts[0].lower() != self.keyword.lower():            return None                if len(parts) == 1:            raise exceptions.AuthenticationFailed('Invalid token header. No credentials provided.')                if len(parts) > 2:            raise exceptions.AuthenticationFailed('Invalid token header. Token string should not contain spaces.')                return parts[1]        def authenticate_credentials(self, key, request):        """Validate token and return user"""        try:            token = self.model.objects.select_related('user').get(key=key)        except self.model.DoesNotExist:            raise exceptions.AuthenticationFailed('Invalid token')                # Check user active        if not token.user.is_active:            raise exceptions.AuthenticationFailed('User inactive or deleted')                # Check expiration        if token.expires < timezone.now():            raise exceptions.AuthenticationFailed('Token has expired')                # Check revocation        if token.revoked:            raise exceptions.AuthenticationFailed('Token has been revoked')                # Optional: Check IP address        client_ip = self.get_client_ip(request)        if token.ip_address and token.ip_address != client_ip:            # Log suspicious activity            self.log_ip_mismatch(token, client_ip)                # Update last used        token.last_used = timezone.now()        token.save(update_fields=['last_used'])                return (token.user, token)        def authenticate_header(self, request):        """Return WWW-Authenticate header for 401 responses"""        return self.keyword        @staticmethod    def get_client_ip(request):        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')        if x_forwarded_for:            ip = x_forwarded_for.split(',')[0]        else:            ip = request.META.get('REMOTE_ADDR')        return ip        @staticmethod    def log_ip_mismatch(token, new_ip):        """Log IP address mismatch"""        import logging        logger = logging.getLogger(__name__)        logger.warning(            f'IP mismatch for token {token.prefix}. '            f'Expected: {token.ip_address}, Got: {new_ip}'        )

Testing Advanced Authentication

# tests.pyfrom rest_framework.test import APITestCasefrom django.contrib.auth.models import Userfrom django.utils import timezonefrom datetime import timedeltafrom .models import UserToken, AccessToken, RefreshToken class TokenAuthenticationTest(APITestCase):    def setUp(self):        self.user = User.objects.create_user('testuser', 'test@test.com', 'password')        def test_login_creates_tokens(self):        """Test login creates access and refresh tokens"""        response = self.client.post('/api/auth/login/', {            'username': 'testuser',            'password': 'password'        })                self.assertEqual(response.status_code, 200)        self.assertIn('access_token', response.data)        self.assertIn('refresh_token', response.data)        def test_expired_token_rejected(self):        """Test expired token is rejected"""        # Create expired token        token = AccessToken.objects.create(            user=self.user,            expires=timezone.now() - timedelta(hours=1)        )                self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}')        response = self.client.get('/api/articles/')                self.assertEqual(response.status_code, 401)        self.assertIn('expired', response.data['detail'].lower())        def test_refresh_token_works(self):        """Test refresh token generates new access token"""        # Login        login_response = self.client.post('/api/auth/login/', {            'username': 'testuser',            'password': 'password'        })                refresh_token = login_response.data['refresh_token']                # Refresh        refresh_response = self.client.post('/api/auth/refresh/', {            'refresh_token': refresh_token        })                self.assertEqual(refresh_response.status_code, 200)        self.assertIn('access_token', refresh_response.data)        def test_multiple_tokens_per_user(self):        """Test user can have multiple tokens"""        # Login from device 1        response1 = self.client.post('/api/auth/login/', {            'username': 'testuser',            'password': 'password',            'device_name': 'iPhone'        })                # Login from device 2        response2 = self.client.post('/api/auth/login/', {            'username': 'testuser',            'password': 'password',            'device_name': 'iPad'        })                # Both should work        self.assertEqual(response1.status_code, 200)        self.assertEqual(response2.status_code, 200)                # User should have 2 tokens        self.assertEqual(UserToken.objects.filter(user=self.user).count(), 2)        def test_token_revocation(self):        """Test token can be revoked"""        token = UserToken.objects.create(user=self.user)                # Use token        self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}')        response = self.client.get('/api/articles/')        self.assertEqual(response.status_code, 200)                # Revoke token        token.revoked = True        token.save()                # Token should not work        response = self.client.get('/api/articles/')        self.assertEqual(response.status_code, 401)

Tóm Tắt Bài 7.2

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

✅ Token Expiration:

Custom token model with expiration
Custom authentication class
Expiration checking

✅ Token Refresh:

Access + Refresh token pattern
Dual token models
Refresh endpoint implementation

✅ Multiple Tokens:

Multiple tokens per user
Device tracking
Token management (list, revoke)

✅ Security Best Practices:

HTTPS enforcement
Token hashing
Rate limiting
Token rotation
IP whitelisting
Automatic cleanup

✅ Custom Authentication:

Complete custom authentication class
Advanced validation
Error handling

✅ Testing:

Testing token expiration
Testing refresh flow
Testing multiple tokens
Testing revocation

Next Steps

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

JWT (JSON Web Tokens) introduction
djangorestframework-simplejwt
JWT structure và claims
Access + Refresh JWT
Custom JWT claims
JWT security best practices

Navigation:

← Bài 7.1: Authentication Basics
Bài 7.2: Token Authentication Advanced (current)
Bài 7.3: JWT Authentication →

#custom-auth

Authentication in DRF - Phần 2 - Token Authentication Advanced