✨(oidc) store and refresh tokens

This provides a way to to refresh the OIDC access token.
The OIDC token will be used to request data to a resource server.
This code is highly related to
mozilla/mozilla-django-oidc#377

Author: qbey

src/backend/core/authentication/backends.py

@@ -10,12 +10,25 @@
 from mozilla_django_oidc.auth import (
     OIDCAuthenticationBackend as MozillaOIDCAuthenticationBackend,
 )
+from mozilla_django_oidc.utils import import_from_settings
 
 from core.models import DuplicateEmailError, User
 
 logger = logging.getLogger(__name__)
 
 
+def store_tokens(session, access_token, id_token, refresh_token):
+    """Store tokens in the session if enabled in settings."""
+    if import_from_settings("OIDC_STORE_ACCESS_TOKEN", False):
+        session["oidc_access_token"] = access_token
+
+    if import_from_settings("OIDC_STORE_ID_TOKEN", False):
+        session["oidc_id_token"] = id_token
+
+    if import_from_settings("OIDC_STORE_REFRESH_TOKEN", False):
+        session["oidc_refresh_token"] = refresh_token
+
+
 class OIDCAuthenticationBackend(MozillaOIDCAuthenticationBackend):
     """Custom OpenID Connect (OIDC) Authentication Backend.
 

src/backend/core/authentication/decorators.py

@@ -0,0 +1,12 @@
+"""
+Decorators for the authentication app.
+
+We don't want (yet) to enforce the OIDC access token to be "fresh" for all
+views, so we provide a decorator to refresh the access token only when needed.
+"""
+
+from django.utils.decorators import decorator_from_middleware
+
+from .middleware import RefreshOIDCAccessToken
+
+refresh_oidc_access_token = decorator_from_middleware(RefreshOIDCAccessToken)

src/backend/core/authentication/middleware.py

@@ -0,0 +1,199 @@
+"""
+Module to declare a RefreshOIDCAccessToken middleware that extends the
+mozilla_django_oidc.middleware.SessionRefresh middleware to refresh the
+access token when it expires, based on the OIDC provided refresh token.
+
+This is based on https://github.com/mozilla/mozilla-django-oidc/pull/377
+which is still not merged.
+"""
+
+import json
+import logging
+import time
+from urllib.parse import quote, urlencode
+
+from django.http import JsonResponse
+from django.urls import reverse
+from django.utils.crypto import get_random_string
+
+import requests
+from mozilla_django_oidc.middleware import SessionRefresh
+
+try:
+    from mozilla_django_oidc.middleware import (  # pylint: disable=unused-import
+        RefreshOIDCAccessToken as MozillaRefreshOIDCAccessToken,
+    )
+
+    # If the import is successful, raise an error to notify the user that the
+    # version of mozilla_django_oidc added the expected middleware, and we don't need
+    # our implementation anymore.
+    # See https://github.com/mozilla/mozilla-django-oidc/pull/377
+    raise RuntimeError("This version of mozilla_django_oidc has RefreshOIDCAccessToken")
+except ImportError:
+    pass
+
+from mozilla_django_oidc.utils import (
+    absolutify,
+    add_state_and_verifier_and_nonce_to_session,
+    import_from_settings,
+)
+
+from core.authentication.backends import store_tokens
+
+logger = logging.getLogger(__name__)
+
+
+class RefreshOIDCAccessToken(SessionRefresh):
+    """
+    A middleware that will refresh the access token following proper OIDC protocol:
+    https://auth0.com/docs/tokens/refresh-token/current
+
+    This is based on https://github.com/mozilla/mozilla-django-oidc/pull/377
+    but limited to our needs (YAGNI/KISS).
+    """
+
+    def _prepare_reauthorization(self, request):
+        """
+        Constructs a new authorization grant request to refresh the session.
+        Besides constructing the request, the state and nonce included in the
+        request are registered in the current session in preparation for the
+        client following through with the authorization flow.
+        """
+        auth_url = self.OIDC_OP_AUTHORIZATION_ENDPOINT
+        client_id = self.OIDC_RP_CLIENT_ID
+        state = get_random_string(self.OIDC_STATE_SIZE)
+
+        # Build the parameters as if we were doing a real auth handoff, except
+        # we also include prompt=none.
+        auth_params = {
+            "response_type": "code",
+            "client_id": client_id,
+            "redirect_uri": absolutify(
+                request, reverse(self.OIDC_AUTHENTICATION_CALLBACK_URL)
+            ),
+            "state": state,
+            "scope": self.OIDC_RP_SCOPES,
+            "prompt": "none",
+        }
+
+        if self.OIDC_USE_NONCE:
+            nonce = get_random_string(self.OIDC_NONCE_SIZE)
+            auth_params.update({"nonce": nonce})
+
+        # Register the one-time parameters in the session
+        add_state_and_verifier_and_nonce_to_session(request, state, auth_params)
+        request.session["oidc_login_next"] = request.get_full_path()
+
+        query = urlencode(auth_params, quote_via=quote)
+        return f"{auth_url}?{query}"
+
+    def is_expired(self, request):
+        """Check whether the access token is expired and needs to be refreshed."""
+        if not self.is_refreshable_url(request):
+            logger.debug("request is not refreshable")
+            return False
+
+        expiration = request.session.get("oidc_token_expiration", 0)
+        now = time.time()
+        if expiration > now:
+            # The id_token is still valid, so we don't have to do anything.
+            logger.debug("id token is still valid (%s > %s)", expiration, now)
+            return False
+
+        return True
+
+    def finish(self, request, prompt_reauth=True):
+        """Finish request handling and handle sending downstream responses for XHR.
+        This function should only be run if the session is determind to
+        be expired.
+        Almost all XHR request handling in client-side code struggles
+        with redirects since redirecting to a page where the user
+        is supposed to do something is extremely unlikely to work
+        in an XHR request. Make a special response for these kinds
+        of requests.
+        The use of 403 Forbidden is to match the fact that this
+        middleware doesn't really want the user in if they don't
+        refresh their session.
+
+        WARNING: this varies from the original implementation:
+         - to return a 401 status code
+         - to consider all requests as XHR requests
+        """
+        xhr_response_json = {"error": "the authentication session has expired"}
+        if prompt_reauth:
+            # The id_token has expired, so we have to re-authenticate silently.
+            refresh_url = self._prepare_reauthorization(request)
+            xhr_response_json["refresh_url"] = refresh_url
+
+        xhr_response = JsonResponse(xhr_response_json, status=401)
+        if "refresh_url" in xhr_response_json:
+            xhr_response["refresh_url"] = xhr_response_json["refresh_url"]
+        return xhr_response
+
+    def process_request(self, request):  # noqa: PLR0911 # pylint: disable=too-many-return-statements
+        """Process the request and refresh the access token if necessary."""
+        if not self.is_expired(request):
+            return None
+
+        token_url = self.get_settings("OIDC_OP_TOKEN_ENDPOINT")
+        client_id = self.get_settings("OIDC_RP_CLIENT_ID")
+        client_secret = self.get_settings("OIDC_RP_CLIENT_SECRET")
+        refresh_token = request.session.get("oidc_refresh_token")
+
+        if not refresh_token:
+            logger.debug("no refresh token stored")
+            return self.finish(request, prompt_reauth=True)
+
+        token_payload = {
+            "grant_type": "refresh_token",
+            "client_id": client_id,
+            "client_secret": client_secret,
+            "refresh_token": refresh_token,
+        }
+
+        req_auth = None
+        if self.get_settings("OIDC_TOKEN_USE_BASIC_AUTH", False):
+            # supported in https://github.com/mozilla/mozilla-django-oidc/pull/377
+            # but we don't need it, so enforce error here.
+            raise RuntimeError("OIDC_TOKEN_USE_BASIC_AUTH is not supported")
+
+        try:
+            response = requests.post(
+                token_url,
+                auth=req_auth,
+                data=token_payload,
+                verify=import_from_settings("OIDC_VERIFY_SSL", True),
+                timeout=import_from_settings("OIDC_TIMEOUT", 3),
+            )
+            response.raise_for_status()
+            token_info = response.json()
+        except requests.exceptions.Timeout:
+            logger.debug("timed out refreshing access token")
+            # Don't prompt for reauth as this could be a temporary problem
+            return self.finish(request, prompt_reauth=False)
+        except requests.exceptions.HTTPError as exc:
+            status_code = exc.response.status_code
+            logger.debug("http error %s when refreshing access token", status_code)
+            # OAuth error response will be a 400 for various situations, including
+            # an expired token. https://datatracker.ietf.org/doc/html/rfc6749#section-5.2
+            return self.finish(request, prompt_reauth=status_code == 400)
+        except json.JSONDecodeError:
+            logger.debug("malformed response when refreshing access token")
+            # Don't prompt for reauth as this could be a temporary problem
+            return self.finish(request, prompt_reauth=False)
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.exception(
+                "unknown error occurred when refreshing access token: %s", exc
+            )
+            # Don't prompt for reauth as this could be a temporary problem
+            return self.finish(request, prompt_reauth=False)
+
+        # Until we can properly validate an ID token on the refresh response
+        # per the spec[1], we intentionally drop the id_token.
+        # [1]: https://openid.net/specs/openid-connect-core-1_0.html#RefreshTokenResponse
+        id_token = None
+        access_token = token_info.get("access_token")
+        refresh_token = token_info.get("refresh_token")
+        store_tokens(request.session, access_token, id_token, refresh_token)
+
+        return None

src/backend/core/tests/authentication/test_decorators.py

@@ -0,0 +1,55 @@
+"""Tests for the refresh_oidc_access_token decorator in core app."""
+
+from unittest.mock import patch
+
+from django.http import HttpResponse
+from django.test import RequestFactory
+from django.utils.decorators import method_decorator
+from django.views import View
+
+from core.authentication.decorators import refresh_oidc_access_token
+
+
+class RefreshOIDCAccessTokenView(View):
+    """
+    A Django view that uses the refresh_oidc_access_token decorator to refresh
+    the OIDC access token before processing the request.
+    """
+
+    @method_decorator(refresh_oidc_access_token)
+    def dispatch(self, request, *args, **kwargs):
+        """
+        Overrides the dispatch method to apply the refresh_oidc_access_token decorator.
+        """
+        return super().dispatch(request, *args, **kwargs)
+
+    def get(self, request, *args, **kwargs):
+        """
+        Handles GET requests.
+
+        Returns:
+            HttpResponse: A simple HTTP response with "OK" as the content.
+        """
+        return HttpResponse("OK")
+
+
+def test_refresh_oidc_access_token_decorator():
+    """
+    Tests the refresh_oidc_access_token decorator is called on RefreshOIDCAccessTokenView access.
+
+    The test creates a mock request and patches the dispatch method to verify that it is called
+    with the correct request object.
+    """
+    # Create a test request
+    factory = RequestFactory()
+    request = factory.get("/")
+
+    # Mock the OIDC refresh functionality
+    with patch(
+        "core.authentication.middleware.RefreshOIDCAccessToken.process_request"
+    ) as mock_refresh:
+        # Call the decorated view
+        RefreshOIDCAccessTokenView.as_view()(request)
+
+        # Assert that the refresh method was called
+        mock_refresh.assert_called_once_with(request)