fix: update blacklist() type hint, blacklist_family() type hint and return values, and TokenVerifySerializer check. resolves #911

Author: juanbailon

rest_framework_simplejwt/serializers.py

@@ -14,12 +14,10 @@
 from .tokens import RefreshToken, SlidingToken, Token, UntypedToken, FamilyMixin
 from .utils import get_md5_hash_password
 from .cache import blacklist_cache
+from .token_blacklist.models import BlacklistedToken
 
 AuthUser = TypeVar("AuthUser", AbstractBaseUser, TokenUser)
 
-if api_settings.BLACKLIST_AFTER_ROTATION:
-    from .token_blacklist.models import BlacklistedToken
-
 
 class PasswordField(serializers.CharField):
     def __init__(self, *args, **kwargs) -> None:
@@ -263,7 +261,7 @@ def validate(self, attrs: dict[str, None]) -> dict[Any, Any]:
         token = UntypedToken(attrs["token"])
 
         if (
-            api_settings.BLACKLIST_AFTER_ROTATION
+            token.get(api_settings.TOKEN_TYPE_CLAIM) == RefreshToken.token_type
             and "rest_framework_simplejwt.token_blacklist" in settings.INSTALLED_APPS
         ):
             jti = token.get(api_settings.JTI_CLAIM)

rest_framework_simplejwt/tokens.py

@@ -288,7 +288,7 @@ def check_blacklist(self) -> None:
             if BlacklistedToken.objects.filter(token__jti=jti).exists():
                 raise RefreshTokenBlacklistedError(_("Token is blacklisted"))
 
-        def blacklist(self) -> BlacklistedToken:
+        def blacklist(self) -> tuple[BlacklistedToken, bool]:
             """
             Ensures this token is included in the outstanding token list and
             adds it to the blacklist.
@@ -389,7 +389,7 @@ def verify(self, *args, **kwargs) -> None:
 
             super().verify(*args, **kwargs)  # type: ignore
 
-        def blacklist_family(self) -> BlacklistedTokenFamily:
+        def blacklist_family(self) -> tuple[BlacklistedTokenFamily, bool]:
             """
             Blacklists the token family.
             """
@@ -413,7 +413,7 @@ def blacklist_family(self) -> BlacklistedTokenFamily:
             if blacklist_cache.is_families_cache_enabled:
                 blacklist_cache.add_token_family(family_id)
 
-            return blacklisted_fam
+            return blacklisted_fam, created
 
         def get_family_id(self) -> Optional[str]:
             return self.payload.get(api_settings.TOKEN_FAMILY_CLAIM, None)

tests/test_token_blacklist.py

@@ -312,7 +312,7 @@ def setUp(self):
         super().setUp()
 
     @override_api_settings(BLACKLIST_AFTER_ROTATION=True)
-    def test_token_verify_serializer_should_honour_blacklist_if_blacklisting_enabled(
+    def test_token_verify_serializer_should_honour_blacklist_if_rotation_enabled(
         self,
     ):
         refresh_token = RefreshToken.for_user(self.user)
@@ -322,14 +322,14 @@ def test_token_verify_serializer_should_honour_blacklist_if_blacklisting_enabled
         self.assertFalse(serializer.is_valid())
 
     @override_api_settings(BLACKLIST_AFTER_ROTATION=False)
-    def test_token_verify_serializer_should_not_honour_blacklist_if_blacklisting_not_enabled(
+    def test_token_verify_serializer_should_honour_blacklist_if_rotation_not_enabled(
         self,
     ):
         refresh_token = RefreshToken.for_user(self.user)
         refresh_token.blacklist()
 
         serializer = TokenVerifySerializer(data={"token": str(refresh_token)})
-        self.assertTrue(serializer.is_valid())
+        self.assertFalse(serializer.is_valid())
 
 
 class TestBigAutoFieldIDMigration(MigrationTestCase):

tests/test_token_family.py

@@ -243,12 +243,13 @@ def test_token_family_can_be_manually_blacklisted(self):
         self.assertEqual(TokenFamily.objects.count(), 1)
 
         # Add family to blacklist
-        blacklisted_fam = token.blacklist_family()
+        blacklisted_fam, created = token.blacklist_family()
 
         # Should not add family to tokenfamily list if already present
         self.assertEqual(TokenFamily.objects.count(), 1)
 
         # Should return blacklist record
+        self.assertTrue(created)
         self.assertEqual(blacklisted_fam.family.family_id, token.get_family_id())
 
         with self.assertRaises(TokenError) as cm: