feat: blocklist to revoke jwt tokens after logout

cms/djangoapps/contentstore/rest_api/v2/views/home.py

@@ -7,6 +7,7 @@
 from rest_framework.views import APIView
 from rest_framework.pagination import PageNumberPagination
 
+
 from openedx.core.lib.api.view_utils import view_auth_classes
 
 from cms.djangoapps.contentstore.utils import get_course_context_v2

cms/djangoapps/contentstore/rest_api/v2/views/tests/test_home.py

@@ -4,16 +4,23 @@
 
 from collections import OrderedDict
 from datetime import datetime, timedelta
+from unittest import mock
 
 import ddt
+import jwt
 import pytz
 from django.conf import settings
 from django.urls import reverse
 from rest_framework import status
+from rest_framework.test import APITestCase
 
 from cms.djangoapps.contentstore.tests.utils import CourseTestCase
 from cms.djangoapps.contentstore.utils import reverse_course_url
+from openedx.core.djangoapps.oauth_dispatch.jwt import create_jwt_for_user
 from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
+from openedx.core.djangolib.testing.utils import skip_unless_cms
+from common.djangoapps.student.tests.factories import UserFactory
+from openedx.core.djangoapps.user_authn.views import logout as logout_views
 
 
 @ddt.ddt
@@ -298,3 +305,47 @@ def test_if_empty_list_of_courses_non_staff(self, query_param, value):
 
         self.assertEqual(len(response.data["results"]["courses"]), 0)
         self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+
+@skip_unless_cms
+class HomePageCoursesViewV2TokenRevocationTests(APITestCase):
+    """Integration tests for JWT revocation behavior on the home courses endpoint."""
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory.create()
+        self.home_courses_url = reverse("cms.djangoapps.contentstore:v2:courses")
+        self.logout_url = reverse('logout')
+        self.token = create_jwt_for_user(self.user)
+        token_parts = self.token.split('.')
+        self.jwt_header_payload = '.'.join(token_parts[:2])
+        self.jwt_signature = token_parts[2]
+
+    def _set_jwt_cookies_on_client(self):
+        """Set the split JWT cookies expected by JwtAuthCookieMiddleware."""
+        self.client.cookies['edx-jwt-cookie-header-payload'] = self.jwt_header_payload
+        self.client.cookies['edx-jwt-cookie-signature'] = self.jwt_signature
+
+    @mock.patch('cms.djangoapps.contentstore.rest_api.v2.views.home.get_course_context_v2', return_value=([], []))
+    def test_home_page_rejects_revoked_jwt_after_logout(self, _mock_get_course_context):
+        """The same JWT cookies work before logout and fail with 401 after logout revocation."""
+        self._set_jwt_cookies_on_client()
+        response_before_logout = self.client.get(self.home_courses_url)
+        self.assertEqual(response_before_logout.status_code, status.HTTP_200_OK)
+
+        with mock.patch(
+            'openedx.core.djangoapps.user_authn.views.logout._blocklist_request_jwt',
+            wraps=logout_views._blocklist_request_jwt,  # pylint: disable=protected-access
+        ) as mock_blocklist_jwt:
+            logout_response = self.client.get(self.logout_url)
+
+        self.assertEqual(logout_response.status_code, status.HTTP_200_OK)
+        mock_blocklist_jwt.assert_called_once()
+
+        self._set_jwt_cookies_on_client()
+        response_after_logout = self.client.get(self.home_courses_url)
+        self.assertEqual(response_after_logout.status_code, status.HTTP_401_UNAUTHORIZED)
+
+        claims = jwt.decode(self.token, options={'verify_signature': False, 'verify_exp': False})
+        expected_cache_key = f"{logout_views.BLOCKLIST_KEY_PREFIX}{claims['sub']}:{claims['iat']}"
+        self.assertEqual(logout_views.cache.get(expected_cache_key), 'revoked')

openedx/core/djangoapps/user_authn/views/logout.py

@@ -4,11 +4,14 @@
 import re
 import urllib.parse as parse  # pylint: disable=import-error
 from urllib.parse import parse_qs, urlsplit, urlunsplit  # pylint: disable=import-error
+from time import time
 
 import nh3
+import jwt
 from django.conf import settings
 from django.contrib.auth import logout
 from django.shortcuts import redirect
+from django.core.cache import cache
 from django.utils.http import urlencode
 from django.views.generic import TemplateView
 from oauth2_provider.models import Application
@@ -19,6 +22,51 @@
 from common.djangoapps.third_party_auth import pipeline as tpa_pipeline
 
 
+BLOCKLIST_KEY_PREFIX = 'blocklist:'
+
+
+def _get_authorization_token(request):
+    """Return the JWT from cookies."""
+
+    # In browser-based requests, Open edX stores JWT in two cookies.
+    header_payload = request.COOKIES.get('edx-jwt-cookie-header-payload')
+    signature = request.COOKIES.get('edx-jwt-cookie-signature')
+    if header_payload and signature:
+        return f'{header_payload}.{signature}'
+
+    return None
+
+
+def _blocklist_request_jwt(request):
+    """Store the current JWT in Redis until it expires."""
+
+    token = _get_authorization_token(request)
+    if not token:
+        return
+
+    try:
+        claims = jwt.decode(token, options={'verify_signature': False, 'verify_exp': False})
+    except jwt.PyJWTError:
+        return
+
+    token_subject = claims.get('sub')
+    token_issued_at = claims.get('iat')
+    token_expires_at = claims.get('exp')
+    if token_subject is None or token_issued_at is None or token_expires_at is None:
+        return
+
+    try:
+        ttl = int(token_expires_at) - int(time())
+    except (TypeError, ValueError):
+        return
+
+    if ttl <= 0:
+        return
+
+    cache_key = f'{BLOCKLIST_KEY_PREFIX}{token_subject}:{token_issued_at}'
+    cache.set(cache_key, 'revoked', timeout=ttl)
+
+
 class LogoutView(TemplateView):
     """
     Logs out user and redirects.
@@ -76,6 +124,9 @@ def dispatch(self, request, *args, **kwargs):
         # Get third party auth provider's logout url
         self.tpa_logout_url = tpa_pipeline.get_idp_logout_url_from_running_pipeline(request)
 
+        # Blocklist the JWT before the session is cleared so the token can no longer be used for API calls.
+        _blocklist_request_jwt(request)
+
         logout(request)
 
         response = super().dispatch(request, *args, **kwargs)

openedx/core/djangoapps/user_authn/views/tests/test_logout.py

@@ -6,12 +6,16 @@
 from unittest import mock
 import ddt
 import nh3
+from django.core.cache import cache
+from django.test import RequestFactory, TestCase
 from django.conf import settings
-from django.test import TestCase
 from django.test.utils import override_settings
 from django.urls import reverse
+from rest_framework.exceptions import AuthenticationFailed
 
+from openedx.core.djangoapps.user_authn.views import logout as logout_views
 from openedx.core.djangoapps.oauth_dispatch.tests.factories import ApplicationFactory
+from openedx.core.djangolib.default_auth_classes import BlocklistJwtAuthentication, DefaultJwtAuthentication
 from openedx.core.djangolib.testing.utils import skip_unless_lms
 from common.djangoapps.student.tests.factories import UserFactory
 
@@ -240,3 +244,86 @@ def test_logout_redirect_failure_with_xss_vulnerability(self, redirect_url, host
             'target': nh3.clean(urllib.parse.unquote(redirect_url)),
         }
         self.assertDictContainsSubset(expected, response.context_data)
+
+
+class AuthenticationAndLogoutBlocklistTests(TestCase):
+    """Tests for JWT blocklisting during logout and JWT authentication."""
+
+    def setUp(self):
+        super().setUp()
+        cache.clear()
+        self.user = UserFactory.create()
+        self.request_factory = RequestFactory()
+        self.token = 'header.payload.signature'
+        self.jwt_claims = {
+            'sub': 'user-uuid',
+            'iat': 1700000000,
+            'exp': 1800000000,
+            'user_id': self.user.id,
+        }
+
+    def test_blocklist_request_jwt_stores_decoded_claims_in_cache(self):
+        """Logout blocklists JWTs using blocklist:<sub>:<iat> cache keys."""
+        request = self.request_factory.get('/logout')
+        request.COOKIES = {
+            'edx-jwt-cookie-header-payload': 'header.payload',
+            'edx-jwt-cookie-signature': 'signature',
+        }
+
+        with mock.patch(
+            'openedx.core.djangoapps.user_authn.views.logout.jwt.decode',
+            return_value=self.jwt_claims,
+        ) as mock_decode, mock.patch(
+            'openedx.core.djangoapps.user_authn.views.logout.time',
+            return_value=1700000001,
+        ), mock.patch(
+            'openedx.core.djangoapps.user_authn.views.logout.cache.set',
+        ) as mock_cache_set:
+            # pylint: disable=protected-access
+            logout_views._blocklist_request_jwt(request)
+
+        expected_cache_key = f"{logout_views.BLOCKLIST_KEY_PREFIX}{self.jwt_claims['sub']}:{self.jwt_claims['iat']}"
+        mock_cache_set.assert_called_once_with(expected_cache_key, 'revoked', timeout=99999999)
+        mock_decode.assert_called_once_with(
+            self.token,
+            options={'verify_signature': False, 'verify_exp': False},
+        )
+
+    def test_blocklist_jwt_authentication_allows_non_revoked_token(self):
+        """JWT authentication succeeds when the token is not in the revocation cache."""
+        request = self.request_factory.get('/api/contentstore/v2/home/courses')
+        auth = BlocklistJwtAuthentication()
+
+        with mock.patch.object(
+            DefaultJwtAuthentication,
+            'authenticate',
+            return_value=(self.user, self.token),
+        ), mock.patch(
+            'openedx.core.djangolib.default_auth_classes.jwt.decode',
+            return_value=self.jwt_claims,
+        ):
+            user_and_token = auth.authenticate(request)
+
+        assert user_and_token == (self.user, self.token)
+
+    def test_blocklist_jwt_authentication_rejects_revoked_token(self):
+        """JWT authentication raises AuthenticationFailed for revoked tokens."""
+        request = self.request_factory.get('/api/contentstore/v2/home/courses')
+        auth = BlocklistJwtAuthentication()
+        cache_key = f"{auth.blocklist_key_prefix}{self.jwt_claims['sub']}:{self.jwt_claims['iat']}"
+
+        with mock.patch.object(
+            DefaultJwtAuthentication,
+            'authenticate',
+            return_value=(self.user, self.token),
+        ), mock.patch(
+            'openedx.core.djangolib.default_auth_classes.jwt.decode',
+            return_value=self.jwt_claims,
+        ), mock.patch(
+            'openedx.core.djangolib.default_auth_classes.cache.get',
+            return_value='revoked',
+        ) as mock_cache_get:
+            with self.assertRaises(AuthenticationFailed):
+                auth.authenticate(request)
+
+        mock_cache_get.assert_called_once_with(cache_key)

openedx/core/djangolib/default_auth_classes.py

@@ -2,9 +2,12 @@
 Default Authentication classes that are ONLY meant to be used by
 DEFAULT_AUTHENTICATION_CLASSES for observability purposes.
 """
+import jwt
+from django.core.cache import cache
 from edx_django_utils.monitoring import set_custom_attribute
 from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
 from rest_framework.authentication import SessionAuthentication
+from rest_framework.exceptions import AuthenticationFailed
 
 
 class DefaultSessionAuthentication(SessionAuthentication):
@@ -53,3 +56,34 @@ def authenticate(self, request):
         # includes a jwt_auth_result custom attribute, so we do not need to
         # reimplement that observability in this class.
         return super().authenticate(request)
+
+
+class BlocklistJwtAuthentication(DefaultJwtAuthentication):
+    """
+    Default JwtAuthentication with Redis-based token revocation support.
+    """
+
+    blocklist_key_prefix = 'blocklist:'
+
+    def authenticate(self, request):
+        user_and_token = super().authenticate(request)
+        if not user_and_token:
+            return user_and_token
+
+        user, token = user_and_token
+        try:
+            claims = jwt.decode(token, options={'verify_signature': False, 'verify_exp': False})
+        except jwt.PyJWTError:
+            return user, token
+
+        token_subject = claims.get('sub')
+        token_issued_at = claims.get('iat')
+
+        if token_subject is None or token_issued_at is None:
+            return user, token
+
+        cache_key = f'{self.blocklist_key_prefix}{token_subject}:{token_issued_at}'
+        if cache.get(cache_key):
+            raise AuthenticationFailed('JWT has been revoked.')
+
+        return user, token

openedx/core/lib/api/view_utils.py

@@ -8,7 +8,7 @@
 from django.core.exceptions import NON_FIELD_ERRORS, ObjectDoesNotExist, ValidationError
 from django.http import Http404, HttpResponseBadRequest
 from django.utils.translation import gettext as _
-from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
+from openedx.core.djangolib.default_auth_classes import BlocklistJwtAuthentication
 from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser
 from opaque_keys import InvalidKeyError
 from opaque_keys.edx.keys import CourseKey
@@ -120,7 +120,7 @@ def _decorator(func_or_class):
         If is_user is True, also requires username in URL matches the request user.
         """
         func_or_class.authentication_classes = (
-            JwtAuthentication,
+            BlocklistJwtAuthentication,
             BearerAuthenticationAllowInactiveUser,
             SessionAuthenticationAllowInactiveUser
         )