Check for blocked tokens depending on the setting for token ids.

If all valid tokens should have a token id ('require'), then all still-valid blacklisted tokens should have been recorded with a token_id, and we can purely look up using the id. Since the migration adding this column backfills the token_id values, this should be safe to rely on immediately after the migration completes.

If tokens are being generated without ids ('off'), then we look up the blacklist record by the whole token value as before. This means that when altering the config from 'require' to 'off' existing blacklisted tokens would no longer match. I doubt that is going to be a common transition in practice. If someone needed to do that carefully, they could shift from 'require' to 'include' for one token expiration horizon, and then to 'off', but I think this is sufficiently fringe to ignore in practice.

For the default setting ('include'), we check for a blacklist entry with either the token id or the exact token value. This should ensure that for people who have kept the defaults, things work in as straightforward manner as possible. Refreshed tokens are invalidated if any token from the same chain of refreshes is invalidated, which still improves logout security for the default case, even though we still storing the full token values, which has some downsides.

Author: ashokdelphia

src/rest_framework_jwt/authentication.py

@@ -82,7 +82,7 @@ def authenticate(self, request):
 
         if apps.is_installed('rest_framework_jwt.blacklist'):
             from rest_framework_jwt.blacklist.models import BlacklistedToken
-            if BlacklistedToken.is_blocked(token):
+            if BlacklistedToken.is_blocked(token, payload):
                 msg = _('Token is blacklisted.')
                 raise exceptions.PermissionDenied(msg)
 

src/rest_framework_jwt/blacklist/models.py

@@ -7,6 +7,8 @@
 from django.utils import timezone
 from django.utils.encoding import force_str
 
+from rest_framework_jwt.settings import api_settings
+
 
 class BlacklistedTokenManager(models.Manager):
     def delete_stale_tokens(self):
@@ -38,5 +40,19 @@ def __str__(self):
 
 
     @staticmethod
-    def is_blocked(token):
-        return BlacklistedToken.objects.filter(token=force_str(token)).exists()
+    def is_blocked(token, payload):
+        token = force_str(token)
+
+        # For invalidated tokens that have an original token id (orig_jti),
+        # we record that in the list, so that the whole family of tokens
+        # refreshed from the same initial token is rejected.
+        token_id = payload.get('orig_jti') or payload.get('jti')
+
+        if api_settings.JWT_TOKEN_ID == 'require':
+            query = Q(token_id=token_id)
+        elif api_settings.JWT_TOKEN_ID == 'off':
+            query = Q(token=token)
+        else:
+            query = Q(token__isnull=False, token=token) | Q(token_id__isnull=False, token_id=token_id)
+
+        return BlacklistedToken.objects.filter(query).exists()

src/rest_framework_jwt/blacklist/permissions.py

@@ -1,3 +1,5 @@
+import jwt
+
 from rest_framework.permissions import BasePermission
 
 from rest_framework_jwt.authentication import JSONWebTokenAuthentication
@@ -15,4 +17,6 @@ def has_permission(self, request, view):
         if token is None:
             return True
 
-        return not BlacklistedToken.is_blocked(token)
+        # The token should already be validated before we call this.
+        payload = jwt.decode(token, None, False)
+        return not BlacklistedToken.is_blocked(token, payload)

src/rest_framework_jwt/utils.py

@@ -217,7 +217,7 @@ def check_payload(token):
 
     if apps.is_installed('rest_framework_jwt.blacklist'):
         from rest_framework_jwt.blacklist.models import BlacklistedToken
-        if BlacklistedToken.is_blocked(token):
+        if BlacklistedToken.is_blocked(token, payload):
             msg = _('Token is blacklisted.')
             raise serializers.ValidationError(msg)