refactor: Rework JWT authentication.

- add invalidation of previously issued tokens for a given user upon successful authentication.

Author: RandomProgramm3r

promo_code/core/tests.py

@@ -6,5 +6,5 @@
 
 class StaticURLTests(django.test.TestCase):
     def test_ping_endpoint(self):
-        response = self.client.get(django.urls.reverse('core:ping'))
+        response = self.client.get(django.urls.reverse('api-core:ping'))
         self.assertEqual(response.status_code, http.HTTPStatus.OK)

promo_code/core/urls.py

@@ -1,7 +1,7 @@
 import core.views
 import django.urls
 
-app_name = 'core'
+app_name = 'api-core'
 
 
 urlpatterns = [
@@ -10,4 +10,9 @@
         core.views.PingView.as_view(),
         name='ping',
     ),
+    django.urls.path(
+        'protected-endpoint/',
+        core.views.MyProtectedView.as_view(),
+        name='protected',
+    ),
 ]

promo_code/core/views.py

@@ -1,7 +1,18 @@
 import django.http
 import django.views
+import rest_framework.permissions
+import rest_framework.response
+import rest_framework.views
 
 
 class PingView(django.views.View):
     def get(self, request, *args, **kwargs):
         return django.http.HttpResponse('PROOOOOOOOOOOOOOOOOD', status=200)
+
+
+class MyProtectedView(rest_framework.views.APIView):
+    permission_classes = [rest_framework.permissions.IsAuthenticated]
+
+    def get(self, request, format=None):
+        content = {'status': 'request was permitted'}
+        return rest_framework.response.Response(content)

promo_code/promo_code/settings.py

@@ -48,47 +48,55 @@ def load_bool(name, default):
 AUTH_USER_MODEL = 'user.User'
 
 REST_FRAMEWORK = {
-    'DEFAULT_RENDERER_CLASSES': ('rest_framework.renderers.JSONRenderer',),
     'DEFAULT_AUTHENTICATION_CLASSES': [
-        'user.authentication.CustomJWTAuthentication',
+        'rest_framework_simplejwt.authentication.JWTAuthentication',
     ],
 }
 
 SIMPLE_JWT = {
-    'ACCESS_TOKEN_LIFETIME': datetime.timedelta(hours=1),
+    'ACCESS_TOKEN_LIFETIME': datetime.timedelta(minutes=60),
     'REFRESH_TOKEN_LIFETIME': datetime.timedelta(days=1),
     'ROTATE_REFRESH_TOKENS': True,
     'BLACKLIST_AFTER_ROTATION': True,
-    'UPDATE_LAST_LOGIN': False,  # !
-    #
+    'UPDATE_LAST_LOGIN': False,
     'ALGORITHM': 'HS256',
-    'SIGNING_KEY': SECRET_KEY,
-    'VERIFYING_KEY': None,
+    'VERIFYING_KEY': '',
     'AUDIENCE': None,
     'ISSUER': None,
     'JSON_ENCODER': 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',
+        '.default_user_authentication_rule'
     ),
-    #
+    'AUTH_TOKEN_CLASSES': ('rest_framework_simplejwt.tokens.AccessToken',),
     'TOKEN_TYPE_CLAIM': 'token_type',
     'TOKEN_USER_CLASS': 'rest_framework_simplejwt.models.TokenUser',
-    #
     'JTI_CLAIM': 'jti',
-    #
     'SLIDING_TOKEN_REFRESH_EXP_CLAIM': 'refresh_exp',
     'SLIDING_TOKEN_LIFETIME': datetime.timedelta(minutes=5),
     'SLIDING_TOKEN_REFRESH_LIFETIME': datetime.timedelta(days=1),
-    #
-    'ACCESS_TOKEN_CLASS': 'user.tokens.CustomAccessToken',
+    'TOKEN_OBTAIN_SERIALIZER': 'user.serializers.SignInSerializer',
+    'TOKEN_REFRESH_SERIALIZER': (
+        'rest_framework_simplejwt.serializers.TokenRefreshSerializer'
+    ),
+    'TOKEN_VERIFY_SERIALIZER': (
+        'rest_framework_simplejwt.serializers.TokenVerifySerializer'
+    ),
+    'TOKEN_BLACKLIST_SERIALIZER': (
+        'rest_framework_simplejwt.serializers.TokenBlacklistSerializer'
+    ),
+    'SLIDING_TOKEN_OBTAIN_SERIALIZER': (
+        'rest_framework_simplejwt.serializers.TokenObtainSlidingSerializer'
+    ),
+    'SLIDING_TOKEN_REFRESH_SERIALIZER': (
+        'rest_framework_simplejwt.serializers.TokenRefreshSlidingSerializer'
+    ),
 }
 
 MIDDLEWARE = [
@@ -99,6 +107,7 @@ def load_bool(name, default):
     'django.contrib.auth.middleware.AuthenticationMiddleware',
     'django.contrib.messages.middleware.MessageMiddleware',
     'django.middleware.clickjacking.XFrameOptionsMiddleware',
+    'user.middleware.TokenVersionMiddleware',
 ]
 
 ROOT_URLCONF = 'promo_code.urls'

promo_code/user/authentication.py

@@ -0,0 +1,25 @@
+import django.http
+import rest_framework_simplejwt.authentication
+
+
+class TokenVersionMiddleware:
+    def __init__(self, get_response):
+        self.get_response = get_response
+
+    def __call__(self, request):
+        auth = rest_framework_simplejwt.authentication.JWTAuthentication()
+        auth_result = auth.authenticate(request)
+
+        if auth_result is None:
+            return self.get_response(request)
+
+        user, token = auth_result
+        if user:
+            token_version = token.payload.get('token_version', 0)
+            if token_version != user.token_version:
+                return django.http.JsonResponse(
+                    {'error': 'Token invalid'},
+                    status=401,
+                )
+
+        return self.get_response(request)

promo_code/user/middleware.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.2b1 on 2025-03-14 19:46
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('user', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='user',
+            name='token_version',
+            field=models.IntegerField(default=0),
+        ),
+    ]

promo_code/user/migrations/0002_user_token_version.py

@@ -46,6 +46,8 @@ class User(
     )
     other = django.db.models.JSONField(default=dict)
 
+    token_version = django.db.models.IntegerField(default=0)
+
     is_active = django.db.models.BooleanField(default=True)
     is_staff = django.db.models.BooleanField(default=False)
     last_login = django.db.models.DateTimeField(null=True, blank=True)

promo_code/user/models.py

@@ -4,6 +4,8 @@
 import rest_framework.exceptions
 import rest_framework.serializers
 import rest_framework.status
+import rest_framework_simplejwt.serializers
+import rest_framework_simplejwt.token_blacklist.models as tb_models
 import rest_framework_simplejwt.tokens
 
 import user.models as user_models
@@ -69,7 +71,9 @@ def create(self, validated_data):
             raise rest_framework.serializers.ValidationError(e.messages)
 
 
-class SignInSerializer(rest_framework.serializers.Serializer):
+class SignInSerializer(
+    rest_framework_simplejwt.serializers.TokenObtainPairSerializer,
+):
     email = rest_framework.serializers.EmailField(required=True)
     password = rest_framework.serializers.CharField(
         required=True,
@@ -97,10 +101,51 @@ def validate(self, data):
                 code='authorization',
             )
 
-        data['user'] = user
+        authenticate_kwargs = {
+            self.username_field: data[self.username_field],
+            'password': data['password'],
+        }
+        try:
+            authenticate_kwargs['request'] = self.context['request']
+        except KeyError:
+            pass
+
+        self.user = django.contrib.auth.authenticate(**authenticate_kwargs)
+
+        if not getattr(self.user, 'is_active', None):
+            raise rest_framework.exceptions.AuthenticationFailed(
+                self.error_messages['no_active_account'],
+                'no_active_account',
+            )
+
+        self.user.token_version += 1
+        self.user.save()
+
+        refresh = self.get_token(self.user)
+        data = {
+            'refresh': str(refresh),
+            'access': str(refresh.access_token),
+        }
+
+        current_jti = refresh['jti']
+
+        tokens_qs = tb_models.OutstandingToken.objects.filter(
+            user=self.user,
+        )
+
+        outstanding_tokens = tokens_qs.exclude(jti=current_jti)
+
+        for token in outstanding_tokens:
+            (
+                tb_models.BlacklistedToken.objects.get_or_create(
+                    token=token,
+                )
+            )
+
+        data['token_version'] = self.user.token_version
         return data
 
-    def get_token(self):
-        user = self.validated_data['user']
-        refresh = rest_framework_simplejwt.tokens.RefreshToken.for_user(user)
-        return {'token': str(refresh.access_token)}
+    def get_token(self, user):
+        token = super().get_token(user)
+        token['token_version'] = user.token_version
+        return token

promo_code/user/serializers.py

@@ -5,7 +5,6 @@
 import rest_framework.test
 
 import user.models
-import user.tokens
 
 
 class AuthTestCase(rest_framework.test.APITestCase):
@@ -18,7 +17,6 @@ def tearDown(self):
         super(AuthTestCase, self).tearDown()
 
     def test_valid_registration_and_email_duplication(self):
-        # Successful registration
         valid_data = {
             'name': 'Emma',
             'surname': 'Thompson',
@@ -36,7 +34,6 @@ def test_valid_registration_and_email_duplication(self):
             rest_framework.status.HTTP_200_OK,
         )
 
-        # Duplicate email registration attempt
         duplicate_data = {
             'name': 'Lui',
             'surname': 'Jomalone',