refactor for consolidated auth code request

mozilla_django_oidc/auth.py

@@ -47,13 +47,13 @@ class OIDCAuthenticationBackend(ModelBackend):
 
     def __init__(self, *args, **kwargs):
         """Initialize settings."""
-        self.OIDC_OP_TOKEN_ENDPOINT = self.get_settings("OIDC_OP_TOKEN_ENDPOINT")
-        self.OIDC_OP_USER_ENDPOINT = self.get_settings("OIDC_OP_USER_ENDPOINT")
-        self.OIDC_OP_JWKS_ENDPOINT = self.get_settings("OIDC_OP_JWKS_ENDPOINT", None)
-        self.OIDC_RP_CLIENT_ID = self.get_settings("OIDC_RP_CLIENT_ID")
-        self.OIDC_RP_CLIENT_SECRET = self.get_settings("OIDC_RP_CLIENT_SECRET")
-        self.OIDC_RP_SIGN_ALGO = self.get_settings("OIDC_RP_SIGN_ALGO", "HS256")
-        self.OIDC_RP_IDP_SIGN_KEY = self.get_settings("OIDC_RP_IDP_SIGN_KEY", None)
+        self.OIDC_OP_TOKEN_ENDPOINT = import_from_settings("OIDC_OP_TOKEN_ENDPOINT")
+        self.OIDC_OP_USER_ENDPOINT = import_from_settings("OIDC_OP_USER_ENDPOINT")
+        self.OIDC_OP_JWKS_ENDPOINT = import_from_settings("OIDC_OP_JWKS_ENDPOINT", None)
+        self.OIDC_RP_CLIENT_ID = import_from_settings("OIDC_RP_CLIENT_ID")
+        self.OIDC_RP_CLIENT_SECRET = import_from_settings("OIDC_RP_CLIENT_SECRET")
+        self.OIDC_RP_SIGN_ALGO = import_from_settings("OIDC_RP_SIGN_ALGO", "HS256")
+        self.OIDC_RP_IDP_SIGN_KEY = import_from_settings("OIDC_RP_IDP_SIGN_KEY", None)
 
         if (
             self.OIDC_RP_SIGN_ALGO.startswith("RS")
@@ -66,10 +66,6 @@ def __init__(self, *args, **kwargs):
 
         self.UserModel = get_user_model()
 
-    @staticmethod
-    def get_settings(attr, *args):
-        return import_from_settings(attr, *args)
-
     def describe_user_by_claims(self, claims):
         email = claims.get("email")
         return "email {}".format(email)
@@ -85,7 +81,7 @@ def verify_claims(self, claims):
         """Verify the provided claims to decide if authentication should be allowed."""
 
         # Verify claims required by default configuration
-        scopes = self.get_settings("OIDC_RP_SCOPES", "openid email")
+        scopes = import_from_settings("OIDC_RP_SCOPES", "openid email")
         if "email" in scopes.split():
             return "email" in claims
 
@@ -107,7 +103,7 @@ def get_username(self, claims):
         # bluntly stolen from django-browserid
         # https://github.com/mozilla/django-browserid/blob/master/django_browserid/auth.py
 
-        username_algo = self.get_settings("OIDC_USERNAME_ALGO", None)
+        username_algo = import_from_settings("OIDC_USERNAME_ALGO", None)
 
         if username_algo:
             if isinstance(username_algo, str):
@@ -159,9 +155,9 @@ def retrieve_matching_jwk(self, token):
         """Get the signing key by exploring the JWKS endpoint of the OP."""
         response_jwks = requests.get(
             self.OIDC_OP_JWKS_ENDPOINT,
-            verify=self.get_settings("OIDC_VERIFY_SSL", True),
-            timeout=self.get_settings("OIDC_TIMEOUT", None),
-            proxies=self.get_settings("OIDC_PROXY", None),
+            verify=import_from_settings("OIDC_VERIFY_SSL", True),
+            timeout=import_from_settings("OIDC_TIMEOUT", None),
+            proxies=import_from_settings("OIDC_PROXY", None),
         )
         response_jwks.raise_for_status()
         jwks = response_jwks.json()
@@ -186,7 +182,7 @@ def retrieve_matching_jwk(self, token):
 
     def get_payload_data(self, token, key):
         """Helper method to get the payload of the JWT token."""
-        if self.get_settings("OIDC_ALLOW_UNSECURED_JWT", False):
+        if import_from_settings("OIDC_ALLOW_UNSECURED_JWT", False):
             header, payload_data, signature = token.split(b".")
             header = json.loads(smart_str(b64decode(header)))
 
@@ -224,7 +220,7 @@ def verify_token(self, token, **kwargs):
         payload = json.loads(payload_data.decode("utf-8"))
         token_nonce = payload.get("nonce")
 
-        if self.get_settings("OIDC_USE_NONCE", True) and nonce != token_nonce:
+        if import_from_settings("OIDC_USE_NONCE", True) and nonce != token_nonce:
             msg = "JWT Nonce verification failed."
             raise SuspiciousOperation(msg)
         return payload
@@ -233,7 +229,7 @@ def get_token(self, payload):
         """Return token object as a dictionary."""
 
         auth = None
-        if self.get_settings("OIDC_TOKEN_USE_BASIC_AUTH", False):
+        if import_from_settings("OIDC_TOKEN_USE_BASIC_AUTH", False):
             # When Basic auth is defined, create the Auth Header and remove secret from payload.
             user = payload.get("client_id")
             pw = payload.get("client_secret")
@@ -245,9 +241,9 @@ def get_token(self, payload):
             self.OIDC_OP_TOKEN_ENDPOINT,
             data=payload,
             auth=auth,
-            verify=self.get_settings("OIDC_VERIFY_SSL", True),
-            timeout=self.get_settings("OIDC_TIMEOUT", None),
-            proxies=self.get_settings("OIDC_PROXY", None),
+            verify=import_from_settings("OIDC_VERIFY_SSL", True),
+            timeout=import_from_settings("OIDC_TIMEOUT", None),
+            proxies=import_from_settings("OIDC_PROXY", None),
         )
         self.raise_token_response_error(response)
         return response.json()
@@ -274,9 +270,9 @@ def get_userinfo(self, access_token, id_token, payload):
         user_response = requests.get(
             self.OIDC_OP_USER_ENDPOINT,
             headers={"Authorization": "Bearer {0}".format(access_token)},
-            verify=self.get_settings("OIDC_VERIFY_SSL", True),
-            timeout=self.get_settings("OIDC_TIMEOUT", None),
-            proxies=self.get_settings("OIDC_PROXY", None),
+            verify=import_from_settings("OIDC_VERIFY_SSL", True),
+            timeout=import_from_settings("OIDC_TIMEOUT", None),
+            proxies=import_from_settings("OIDC_PROXY", None),
         )
         user_response.raise_for_status()
         return user_response.json()
@@ -296,7 +292,7 @@ def authenticate(self, request, **kwargs):
         if not code or not state:
             return None
 
-        reverse_url = self.get_settings(
+        reverse_url = import_from_settings(
             "OIDC_AUTHENTICATION_CALLBACK_URL", "oidc_authentication_callback"
         )
 
@@ -334,10 +330,10 @@ def store_tokens(self, access_token, id_token):
         """Store OIDC tokens."""
         session = self.request.session
 
-        if self.get_settings("OIDC_STORE_ACCESS_TOKEN", False):
+        if import_from_settings("OIDC_STORE_ACCESS_TOKEN", False):
             session["oidc_access_token"] = access_token
 
-        if self.get_settings("OIDC_STORE_ID_TOKEN", False):
+        if import_from_settings("OIDC_STORE_ID_TOKEN", False):
             session["oidc_id_token"] = id_token
 
     def get_or_create_user(self, access_token, id_token, payload):
@@ -361,7 +357,7 @@ def get_or_create_user(self, access_token, id_token, payload):
             # bail. Randomly selecting one seems really wrong.
             msg = "Multiple users returned"
             raise SuspiciousOperation(msg)
-        elif self.get_settings("OIDC_CREATE_USER", True):
+        elif import_from_settings("OIDC_CREATE_USER", True):
             user = self.create_user(user_info)
             return user
         else:

mozilla_django_oidc/middleware.py

@@ -1,20 +1,18 @@
 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,
+    get_url_for_authorization_code_request,
     import_from_settings,
 )
 
@@ -31,23 +29,7 @@ class SessionRefresh(MiddlewareMixin):
 
     def __init__(self, get_response):
         super(SessionRefresh, self).__init__(get_response)
-        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)
+        self.OIDC_EXEMPT_URLS = import_from_settings("OIDC_EXEMPT_URLS", [])
 
     @cached_property
     def exempt_urls(self):
@@ -98,7 +80,6 @@ def is_refreshable_url(self, request):
         :arg HttpRequest request:
 
         :returns: boolean
-
         """
         # Do not attempt to refresh the session if the OIDC backend is not used
         backend_session = request.session.get(BACKEND_SESSION_KEY)
@@ -129,35 +110,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)
-
         request.session["oidc_login_next"] = request.get_full_path()
+        redirect_url = get_url_for_authorization_code_request(
+            request, prompt=False, quote_params_via=quote
+        )
 
-        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,16 @@
 import time
 import warnings
 from hashlib import sha256
+from urllib.parse import quote_plus, 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 +163,73 @@ def add_state_and_verifier_and_nonce_to_session(
         "nonce": nonce,
         "added_on": time.time(),
     }
+
+
+def get_url_for_authorization_code_request(
+    request, prompt=True, quote_params_via=quote_plus
+):
+    """
+    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.
+    """
+    OIDC_AUTHENTICATION_CALLBACK_URL = import_from_settings(
+        "OIDC_AUTHENTICATION_CALLBACK_URL",
+        "oidc_authentication_callback",
+    )
+
+    state = get_random_string(import_from_settings("OIDC_STATE_SIZE", 32))
+
+    params = {
+        "response_type": "code",
+        "client_id": import_from_settings("OIDC_RP_CLIENT_ID"),
+        "scope": import_from_settings("OIDC_RP_SCOPES", "openid email"),
+        "redirect_uri": absolutify(request, reverse(OIDC_AUTHENTICATION_CALLBACK_URL)),
+        "state": state,
+    }
+
+    if not prompt:
+        params.update(prompt="none")
+
+    params.update(import_from_settings("OIDC_AUTH_REQUEST_EXTRA_PARAMS", {}))
+
+    if import_from_settings("OIDC_USE_NONCE", True):
+        params.update(
+            nonce=get_random_string(import_from_settings("OIDC_NONCE_SIZE", 32))
+        )
+
+    if import_from_settings("OIDC_USE_PKCE", False):
+        OIDC_PKCE_CODE_VERIFIER_SIZE = import_from_settings(
+            "OIDC_PKCE_CODE_VERIFIER_SIZE", 64
+        )
+
+        if not (43 <= 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"
+            )
+
+        OIDC_PKCE_CODE_CHALLENGE_METHOD = import_from_settings(
+            "OIDC_PKCE_CODE_CHALLENGE_METHOD", "S256"
+        )
+
+        if OIDC_PKCE_CODE_CHALLENGE_METHOD not in ("plain", "S256"):
+            raise ImproperlyConfigured(
+                "OIDC_PKCE_CODE_CHALLENGE_METHOD must be 'plain' or 'S256'"
+            )
+
+        code_verifier = get_random_string(OIDC_PKCE_CODE_VERIFIER_SIZE)
+        params.update(
+            code_challenge=generate_code_challenge(
+                code_verifier, OIDC_PKCE_CODE_CHALLENGE_METHOD
+            ),
+            code_challenge_method=OIDC_PKCE_CODE_CHALLENGE_METHOD,
+        )
+    else:
+        code_verifier = None
+
+    add_state_and_verifier_and_nonce_to_session(request, state, params, code_verifier)
+
+    query_params = urlencode(params, quote_via=quote_params_via)
+
+    return f"{import_from_settings('OIDC_OP_AUTHORIZATION_ENDPOINT')}?{query_params}"