feat(auth): Revoke refresh token on password change

Author: mahdirahimi1999

rest_framework_simplejwt/serializers.py

@@ -10,6 +10,7 @@
 from .models import TokenUser
 from .settings import api_settings
 from .tokens import RefreshToken, SlidingToken, Token, UntypedToken
+from .utils import get_md5_hash_password
 
 AuthUser = TypeVar("AuthUser", AbstractBaseUser, TokenUser)
 
@@ -111,18 +112,6 @@ class TokenRefreshSerializer(serializers.Serializer):
     def validate(self, attrs: dict[str, Any]) -> dict[str, str]:
         refresh = self.token_class(attrs["refresh"])
 
-        user_id = refresh.payload.get(api_settings.USER_ID_CLAIM, None)
-        if user_id and (
-            user := get_user_model().objects.get(
-                **{api_settings.USER_ID_FIELD: user_id}
-            )
-        ):
-            if not api_settings.USER_AUTHENTICATION_RULE(user):
-                raise AuthenticationFailed(
-                    self.error_messages["no_active_account"],
-                    "no_active_account",
-                )
-
         data = {"access": str(refresh.access_token)}
 
         if api_settings.ROTATE_REFRESH_TOKENS:
@@ -142,6 +131,39 @@ def validate(self, attrs: dict[str, Any]) -> dict[str, str]:
 
             data["refresh"] = str(refresh)
 
+        # 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:
+                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 = refresh.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:
+                            refresh.blacklist()
+                        except AttributeError:
+                            pass
+
+                    raise AuthenticationFailed(
+                        _("The user's password has been changed."), code="password_changed"
+                    )
+
         return data
 
 

tests/test_serializers.py

@@ -286,10 +286,11 @@ def test_it_should_raise_error_for_deleted_users(self):
 
         s = TokenRefreshSerializer(data={"refresh": str(refresh)})
 
-        with self.assertRaises(django_exceptions.ObjectDoesNotExist) as e:
+        # It should raise AuthenticationFailed instead of ObjectDoesNotExist
+        with self.assertRaises(drf_exceptions.AuthenticationFailed) as e:
             s.is_valid()
 
-        self.assertIn("does not exist", str(e.exception))
+        self.assertEqual(e.exception.get_codes(), "no_active_account")
 
     def test_it_should_raise_error_for_inactive_users(self):
         refresh = RefreshToken.for_user(self.user)
@@ -483,6 +484,55 @@ def test_blacklist_app_not_installed_should_pass(self):
         reload(tokens)
         reload(serializers)
 
+    @override_api_settings(
+        CHECK_REVOKE_TOKEN=True,
+        REVOKE_TOKEN_CLAIM="hash_password",
+        BLACKLIST_AFTER_ROTATION=False,
+    )
+    def test_refresh_token_should_fail_after_password_change(self):
+        """
+        Tests that token refresh fails if CHECK_REVOKE_TOKEN is True and the
+        user's password has changed.
+        """
+        refresh = RefreshToken.for_user(self.user)
+        self.user.set_password("new_password")
+        self.user.save()
+
+        s = TokenRefreshSerializer(data={"refresh": str(refresh)})
+
+        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_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()
+
+        s = TokenRefreshSerializer(data={"refresh": str(refresh)})
+        with self.assertRaises(drf_exceptions.AuthenticationFailed):
+            s.is_valid(raise_exception=True)
+
+        # Check that the token is now in the blacklist
+        jti = refresh[api_settings.JTI_CLAIM]
+        self.assertTrue(OutstandingToken.objects.filter(jti=jti).exists())
+        self.assertTrue(BlacklistedToken.objects.filter(token__jti=jti).exists())
+
 
 class TestTokenVerifySerializer(TestCase):
     def test_it_should_raise_token_error_if_token_invalid(self):