feat(serializers): Add full user validation to sliding token refresh

Implements the same user validation logic (active status, password change)
in  to ensure consistent behavior with the
standard .

Author: mahdirahimi1999

rest_framework_simplejwt/serializers.py

@@ -113,9 +113,6 @@ class TokenRefreshSerializer(serializers.Serializer):
     def validate(self, attrs: dict[str, Any]) -> dict[str, str]:
         refresh = self.token_class(attrs["refresh"])
 
-        data = {"access": str(refresh.access_token)}
-
-        # We handle user-related validation in a single, efficient block.
         user_id = refresh.payload.get(api_settings.USER_ID_CLAIM, None)
         if user_id:
             try:
@@ -154,6 +151,7 @@ def validate(self, attrs: dict[str, Any]) -> dict[str, str]:
                         code="password_changed",
                     )
 
+        data = {"access": str(refresh.access_token)}
         if api_settings.ROTATE_REFRESH_TOKENS:
             if api_settings.BLACKLIST_AFTER_ROTATION:
                 try:
@@ -178,8 +176,50 @@ class TokenRefreshSlidingSerializer(serializers.Serializer):
     token = serializers.CharField()
     token_class = SlidingToken
 
+    default_error_messages = {
+        "no_active_account": _("No active account found for the given token."),
+        "password_changed": _("The user's password has been changed."),
+    }
+
     def validate(self, attrs: dict[str, Any]) -> dict[str, str]:
         token = self.token_class(attrs["token"])
+        user_id = token.payload.get(api_settings.USER_ID_CLAIM, None)
+        if user_id:
+            try:
+                user = get_user_model().objects.get(
+                    **{api_settings.USER_ID_FIELD: user_id}
+                )
+            except get_user_model().DoesNotExist:
+                # This handles the case where the user has been deleted.
+                raise AuthenticationFailed(
+                    self.error_messages["no_active_account"], "no_active_account"
+                )
+
+            if not api_settings.USER_AUTHENTICATION_RULE(user):
+                raise AuthenticationFailed(
+                    self.error_messages["no_active_account"], "no_active_account"
+                )
+
+            if api_settings.CHECK_REVOKE_TOKEN:
+                token_hash = token.payload.get(api_settings.REVOKE_TOKEN_CLAIM)
+                user_hash = get_md5_hash_password(user.password)
+
+                if token_hash != user_hash:
+                    # If the password has changed, we blacklist the token
+                    # to prevent any further use.
+                    if (
+                        "rest_framework_simplejwt.token_blacklist"
+                        in settings.INSTALLED_APPS
+                    ):
+                        try:
+                            token.blacklist()
+                        except AttributeError:
+                            pass
+
+                    raise AuthenticationFailed(
+                        self.error_messages["password_changed"],
+                        code="password_changed",
+                    )
 
         # Check that the timestamp in the "refresh_exp" claim has not
         # passed

tests/test_serializers.py

@@ -188,6 +188,15 @@ def test_it_should_produce_a_json_web_token_when_valid(self):
 
 
 class TestTokenRefreshSlidingSerializer(TestCase):
+    def setUp(self):
+        self.username = "test_user"
+        self.password = "test_password"
+
+        self.user = User.objects.create_user(
+            username=self.username,
+            password=self.password,
+        )
+
     def test_it_should_not_validate_if_token_invalid(self):
         token = SlidingToken()
         del token["exp"]
@@ -269,6 +278,74 @@ def test_it_should_update_token_exp_claim_if_everything_ok(self):
 
         self.assertTrue(old_exp < new_exp)
 
+    def test_it_should_raise_error_for_deleted_users(self):
+        token = SlidingToken.for_user(self.user)
+        self.user.delete()
+
+        s = TokenRefreshSlidingSerializer(data={"token": str(token)})
+
+        # It should raise AuthenticationFailed instead of ObjectDoesNotExist
+        with self.assertRaises(drf_exceptions.AuthenticationFailed) as e:
+            s.is_valid()
+
+        self.assertEqual(e.exception.get_codes(), "no_active_account")
+
+    def test_it_should_raise_error_for_inactive_users(self):
+        token = SlidingToken.for_user(self.user)
+        self.user.is_active = False
+        self.user.save()
+
+        s = TokenRefreshSlidingSerializer(data={"token": str(token)})
+
+        with self.assertRaises(drf_exceptions.AuthenticationFailed) as e:
+            s.is_valid()
+
+        self.assertEqual(e.exception.get_codes(), "no_active_account")
+
+    @override_api_settings(
+        CHECK_REVOKE_TOKEN=True,
+        REVOKE_TOKEN_CLAIM="hash_password",
+        BLACKLIST_AFTER_ROTATION=False,
+    )
+    def test_sliding_token_should_fail_after_password_change(self):
+        """
+        Tests that sliding token refresh fails if CHECK_REVOKE_TOKEN is True and the
+        user's password has changed.
+        """
+        token = SlidingToken.for_user(self.user)
+        self.user.set_password("new_password")
+        self.user.save()
+
+        s = TokenRefreshSlidingSerializer(data={"token": str(token)})
+
+        with self.assertRaises(drf_exceptions.AuthenticationFailed) as e:
+            s.is_valid(raise_exception=True)
+
+        self.assertEqual(e.exception.get_codes(), "password_changed")
+
+    @override_api_settings(
+        CHECK_REVOKE_TOKEN=True,
+        REVOKE_TOKEN_CLAIM="hash_password",
+        BLACKLIST_AFTER_ROTATION=True,
+    )
+    def test_sliding_token_should_blacklist_after_password_change(self):
+        """
+        Tests that if sliding token refresh fails due to a password change, the
+        offending token is blacklisted.
+        """
+        token = SlidingToken.for_user(self.user)
+        self.user.set_password("new_password")
+        self.user.save()
+
+        s = TokenRefreshSlidingSerializer(data={"token": str(token)})
+        with self.assertRaises(drf_exceptions.AuthenticationFailed):
+            s.is_valid(raise_exception=True)
+
+        # Check that the token is now in the blacklist
+        jti = token[api_settings.JTI_CLAIM]
+        self.assertTrue(OutstandingToken.objects.filter(jti=jti).exists())
+        self.assertTrue(BlacklistedToken.objects.filter(token__jti=jti).exists())
+
 
 class TestTokenRefreshSerializer(TestCase):
     def setUp(self):
@@ -515,11 +592,6 @@ def test_refresh_token_should_blacklist_after_password_change(self):
         Tests that if token refresh fails due to a password change, the
         offending refresh token is blacklisted.
         """
-        from rest_framework_simplejwt.token_blacklist.models import (
-            BlacklistedToken,
-            OutstandingToken,
-        )
-
         refresh = RefreshToken.for_user(self.user)
         self.user.set_password("new_password")
         self.user.save()