refactor for consolidated auth code request

Author: escattone

mozilla_django_oidc/middleware.py

@@ -1,27 +1,22 @@
 import logging
 import time
 from re import Pattern as re_Pattern
-from urllib.parse import quote, urlencode
+from urllib.parse import quote
 
 from django.contrib.auth import BACKEND_SESSION_KEY
 from django.http import HttpResponseRedirect, JsonResponse
 from django.urls import reverse
-from django.utils.crypto import get_random_string
 from django.utils.deprecation import MiddlewareMixin
 from django.utils.functional import cached_property
 from django.utils.module_loading import import_string
 
 from mozilla_django_oidc.auth import OIDCAuthenticationBackend
-from mozilla_django_oidc.utils import (
-    absolutify,
-    add_state_and_verifier_and_nonce_to_session,
-    import_from_settings,
-)
+from mozilla_django_oidc.utils import AuthorizationCodeRequestMixin
 
 LOGGER = logging.getLogger(__name__)
 
 
-class SessionRefresh(MiddlewareMixin):
+class SessionRefresh(MiddlewareMixin, AuthorizationCodeRequestMixin):
     """Refreshes the session with the OIDC RP after expiry seconds
 
     For users authenticated with the OIDC RP, verify tokens are still valid and
@@ -30,24 +25,9 @@ class SessionRefresh(MiddlewareMixin):
     """
 
     def __init__(self, get_response):
-        super(SessionRefresh, self).__init__(get_response)
+        super().__init__(get_response)
+        self.init_settings()
         self.OIDC_EXEMPT_URLS = self.get_settings("OIDC_EXEMPT_URLS", [])
-        self.OIDC_OP_AUTHORIZATION_ENDPOINT = self.get_settings(
-            "OIDC_OP_AUTHORIZATION_ENDPOINT"
-        )
-        self.OIDC_RP_CLIENT_ID = self.get_settings("OIDC_RP_CLIENT_ID")
-        self.OIDC_STATE_SIZE = self.get_settings("OIDC_STATE_SIZE", 32)
-        self.OIDC_AUTHENTICATION_CALLBACK_URL = self.get_settings(
-            "OIDC_AUTHENTICATION_CALLBACK_URL",
-            "oidc_authentication_callback",
-        )
-        self.OIDC_RP_SCOPES = self.get_settings("OIDC_RP_SCOPES", "openid email")
-        self.OIDC_USE_NONCE = self.get_settings("OIDC_USE_NONCE", True)
-        self.OIDC_NONCE_SIZE = self.get_settings("OIDC_NONCE_SIZE", 32)
-
-    @staticmethod
-    def get_settings(attr, *args):
-        return import_from_settings(attr, *args)
 
     @cached_property
     def exempt_urls(self):
@@ -115,6 +95,11 @@ def is_refreshable_url(self, request):
             and not any(pat.match(request.path) for pat in self.exempt_url_patterns)
         )
 
+    def get_extra_params(self, request):
+        extra = super().get_extra_params(request)
+        extra.update(prompt="none")
+        return extra
+
     def process_request(self, request):
         if not self.is_refreshable_url(request):
             LOGGER.debug("request is not refreshable")
@@ -129,35 +114,11 @@ def process_request(self, request):
 
         LOGGER.debug("id token has expired")
         # The id_token has expired, so we have to re-authenticate silently.
-        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.
-        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",
-        }
-
-        params.update(self.get_settings("OIDC_AUTH_REQUEST_EXTRA_PARAMS", {}))
-
-        if self.OIDC_USE_NONCE:
-            nonce = get_random_string(self.OIDC_NONCE_SIZE)
-            params.update({"nonce": nonce})
-
-        add_state_and_verifier_and_nonce_to_session(request, state, params)
-
+        redirect_url = self.get_url_for_authorization_code_request(
+            request, quote_via=quote
+        )
         request.session["oidc_login_next"] = request.get_full_path()
 
-        query = urlencode(params, quote_via=quote)
-        redirect_url = "{url}?{query}".format(url=auth_url, query=query)
         if request.headers.get("x-requested-with") == "XMLHttpRequest":
             # Almost all XHR request handling in client-side code struggles
             # with redirects since redirecting to a page where the user

mozilla_django_oidc/utils.py

@@ -2,12 +2,15 @@
 import time
 import warnings
 from hashlib import sha256
+from urllib.parse import urlencode
 from urllib.request import parse_http_list, parse_keqv_list
 
 # Make it obvious that these aren't the usual base64 functions
 import josepy.b64
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
+from django.urls import reverse
+from django.utils.crypto import get_random_string
 
 LOGGER = logging.getLogger(__name__)
 
@@ -159,3 +162,88 @@ def add_state_and_verifier_and_nonce_to_session(
         "nonce": nonce,
         "added_on": time.time(),
     }
+
+
+class AuthorizationCodeRequestMixin:
+    """
+    Class that encapsulates the functionality required to make an authorization code request.
+    """
+
+    @staticmethod
+    def get_settings(attr, *args):
+        return import_from_settings(attr, *args)
+
+    def init_settings(self):
+        self.OIDC_OP_AUTH_ENDPOINT = self.get_settings("OIDC_OP_AUTHORIZATION_ENDPOINT")
+        self.OIDC_OP_AUTHORIZATION_ENDPOINT = self.OIDC_OP_AUTH_ENDPOINT
+        self.OIDC_RP_CLIENT_ID = self.get_settings("OIDC_RP_CLIENT_ID")
+        self.OIDC_STATE_SIZE = self.get_settings("OIDC_STATE_SIZE", 32)
+        self.OIDC_AUTHENTICATION_CALLBACK_URL = self.get_settings(
+            "OIDC_AUTHENTICATION_CALLBACK_URL",
+            "oidc_authentication_callback",
+        )
+        self.OIDC_RP_SCOPES = self.get_settings("OIDC_RP_SCOPES", "openid email")
+        self.OIDC_USE_NONCE = self.get_settings("OIDC_USE_NONCE", True)
+        self.OIDC_NONCE_SIZE = self.get_settings("OIDC_NONCE_SIZE", 32)
+        self.OIDC_USE_PKCE = self.get_settings("OIDC_USE_PKCE", False)
+        self.OIDC_PKCE_CODE_VERIFIER_SIZE = self.get_settings(
+            "OIDC_PKCE_CODE_VERIFIER_SIZE", 64
+        )
+
+        if not (43 <= self.OIDC_PKCE_CODE_VERIFIER_SIZE <= 128):
+            # Check that OIDC_PKCE_CODE_VERIFIER_SIZE is between the min and max length
+            # defined in https://datatracker.ietf.org/doc/html/rfc7636#section-4.1
+            raise ImproperlyConfigured(
+                "OIDC_PKCE_CODE_VERIFIER_SIZE must be between 43 and 128"
+            )
+
+        self.OIDC_PKCE_CODE_CHALLENGE_METHOD = self.get_settings(
+            "OIDC_PKCE_CODE_CHALLENGE_METHOD", "S256"
+        )
+
+        if self.OIDC_PKCE_CODE_CHALLENGE_METHOD not in ("plain", "S256"):
+            raise ImproperlyConfigured(
+                "OIDC_PKCE_CODE_CHALLENGE_METHOD must be 'plain' or 'S256'"
+            )
+
+    def get_extra_params(self, request):
+        return self.get_settings("OIDC_AUTH_REQUEST_EXTRA_PARAMS", {})
+
+    def get_url_for_authorization_code_request(self, request, **urlencode_kwargs):
+        """
+        Builds and returns the URL required for the authorization code request, and
+        also adds the state, nonce, and code verifier (if using PKCE) to the session.
+        """
+        state = get_random_string(self.OIDC_STATE_SIZE)
+
+        params = {
+            "response_type": "code",
+            "scope": self.OIDC_RP_SCOPES,
+            "client_id": self.OIDC_RP_CLIENT_ID,
+            "redirect_uri": absolutify(
+                request, reverse(self.OIDC_AUTHENTICATION_CALLBACK_URL)
+            ),
+            "state": state,
+        }
+
+        params.update(self.get_extra_params(request))
+
+        if self.OIDC_USE_NONCE:
+            params.update(nonce=get_random_string(self.OIDC_NONCE_SIZE))
+
+        if self.OIDC_USE_PKCE:
+            code_verifier = get_random_string(self.OIDC_PKCE_CODE_VERIFIER_SIZE)
+            params.update(
+                code_challenge=generate_code_challenge(
+                    code_verifier, self.OIDC_PKCE_CODE_CHALLENGE_METHOD
+                ),
+                code_challenge_method=self.OIDC_PKCE_CODE_CHALLENGE_METHOD,
+            )
+        else:
+            code_verifier = None
+
+        add_state_and_verifier_and_nonce_to_session(
+            request, state, params, code_verifier
+        )
+
+        return f"{self.OIDC_OP_AUTHORIZATION_ENDPOINT}?{urlencode(params, **urlencode_kwargs)}"

mozilla_django_oidc/views.py

@@ -1,20 +1,15 @@
 import time
-from urllib.parse import urlencode
 
 from django.contrib import auth
 from django.core.exceptions import SuspiciousOperation
 from django.http import HttpResponseNotAllowed, HttpResponseRedirect
 from django.shortcuts import resolve_url
-from django.urls import reverse
-from django.utils.crypto import get_random_string
 from django.utils.http import url_has_allowed_host_and_scheme
 from django.utils.module_loading import import_string
 from django.views.generic import View
 
 from mozilla_django_oidc.utils import (
-    absolutify,
-    add_state_and_verifier_and_nonce_to_session,
-    generate_code_challenge,
+    AuthorizationCodeRequestMixin,
     import_from_settings,
 )
 
@@ -159,85 +154,26 @@ def get_next_url(request, redirect_field_name):
     return None
 
 
-class OIDCAuthenticationRequestView(View):
+class OIDCAuthenticationRequestView(View, AuthorizationCodeRequestMixin):
     """OIDC client authentication HTTP endpoint"""
 
     http_method_names = ["get"]
 
     def __init__(self, *args, **kwargs):
-        super(OIDCAuthenticationRequestView, self).__init__(*args, **kwargs)
-
-        self.OIDC_OP_AUTH_ENDPOINT = self.get_settings("OIDC_OP_AUTHORIZATION_ENDPOINT")
-        self.OIDC_RP_CLIENT_ID = self.get_settings("OIDC_RP_CLIENT_ID")
-
-    @staticmethod
-    def get_settings(attr, *args):
-        return import_from_settings(attr, *args)
+        super().__init__(*args, **kwargs)
+        self.init_settings()
+        self.OIDC_REDIRECT_FIELD_NAME = self.get_settings(
+            "OIDC_REDIRECT_FIELD_NAME", "next"
+        )
 
     def get(self, request):
         """OIDC client authentication initialization HTTP endpoint"""
-        state = get_random_string(self.get_settings("OIDC_STATE_SIZE", 32))
-        redirect_field_name = self.get_settings("OIDC_REDIRECT_FIELD_NAME", "next")
-        reverse_url = self.get_settings(
-            "OIDC_AUTHENTICATION_CALLBACK_URL", "oidc_authentication_callback"
-        )
-
-        params = {
-            "response_type": "code",
-            "scope": self.get_settings("OIDC_RP_SCOPES", "openid email"),
-            "client_id": self.OIDC_RP_CLIENT_ID,
-            "redirect_uri": absolutify(request, reverse(reverse_url)),
-            "state": state,
-        }
-
-        params.update(self.get_extra_params(request))
-
-        if self.get_settings("OIDC_USE_NONCE", True):
-            nonce = get_random_string(self.get_settings("OIDC_NONCE_SIZE", 32))
-            params.update({"nonce": nonce})
-
-        if self.get_settings("OIDC_USE_PKCE", False):
-            code_verifier_length = self.get_settings("OIDC_PKCE_CODE_VERIFIER_SIZE", 64)
-            # Check that code_verifier_length is between the min and max length
-            # defined in https://datatracker.ietf.org/doc/html/rfc7636#section-4.1
-            if not (43 <= code_verifier_length <= 128):
-                raise ValueError("code_verifier_length must be between 43 and 128")
-
-            # Generate code_verifier and code_challenge pair
-            code_verifier = get_random_string(code_verifier_length)
-            code_challenge_method = self.get_settings(
-                "OIDC_PKCE_CODE_CHALLENGE_METHOD", "S256"
-            )
-            code_challenge = generate_code_challenge(
-                code_verifier, code_challenge_method
-            )
-
-            # Append code_challenge to authentication request parameters
-            params.update(
-                {
-                    "code_challenge": code_challenge,
-                    "code_challenge_method": code_challenge_method,
-                }
-            )
-
-        else:
-            code_verifier = None
-
-        add_state_and_verifier_and_nonce_to_session(
-            request, state, params, code_verifier
-        )
-
-        request.session["oidc_login_next"] = get_next_url(request, redirect_field_name)
-
-        query = urlencode(params)
-        redirect_url = "{url}?{query}".format(
-            url=self.OIDC_OP_AUTH_ENDPOINT, query=query
+        redirect_url = self.get_url_for_authorization_code_request(request)
+        request.session["oidc_login_next"] = get_next_url(
+            request, self.OIDC_REDIRECT_FIELD_NAME
         )
         return HttpResponseRedirect(redirect_url)
 
-    def get_extra_params(self, request):
-        return self.get_settings("OIDC_AUTH_REQUEST_EXTRA_PARAMS", {})
-
 
 class OIDCLogoutView(View):
     """Logout helper view"""