👌 PR feedback

mozilla_django_oidc_db/config.py

@@ -15,6 +15,7 @@
 from mozilla_django_oidc.utils import import_from_settings
 from typing_extensions import Self, TypedDict, Unpack
 
+from .constants import CONFIG_CLASS_SESSION_KEY
 from .models import OpenIDConnectConfigBase
 
 
@@ -111,22 +112,19 @@ def store_config(request: HttpRequest) -> None:
     """
     # Attempt to retrieve the config_class from the session, this only works for users
     # that are actually logged in as Django users
-    if request.session.get("OIDCDB_CONFIG_CLASS", ""):
-        config_class = request.session.get("OIDCDB_CONFIG_CLASS", "")
-    else:
-        # The config_class key is added to the state in the OIDCInit.get method.
-        # TODO: verify that the state query param is present for error flows! Need to check
-        # the OAUTH2 spec for this, but according to ChatGeePeeTee if the request contains
-        # it, the callback must have it too.
-        state_key = request.GET.get("state")
-        if not state_key or state_key not in (
-            states := request.session.get("oidc_states", [])
-        ):
-            raise BadRequest("Could not look up the referenced config.")
-
+    # The config_class key is added to the state in the OIDCInit.get method.
+    # TODO: verify that the state query param is present for error flows! Need to check
+    # the OAUTH2 spec for this, but according to ChatGeePeeTee if the request contains
+    # it, the callback must have it too.
+    config_class = ""
+    state_key = request.GET.get("state")
+    if state_key and state_key in (states := request.session.get("oidc_states", [])):
         state = states[state_key]
         config_class = state.get("config_class", "")
 
+    if not config_class and request.session.get(CONFIG_CLASS_SESSION_KEY, ""):
+        config_class = request.session.get(CONFIG_CLASS_SESSION_KEY, "")
+
     try:
         config = apps.get_model(config_class)
     except (LookupError, ValueError) as exc:

mozilla_django_oidc_db/constants.py

@@ -9,3 +9,5 @@
 }
 
 OPEN_ID_CONFIG_PATH = ".well-known/openid-configuration"
+
+CONFIG_CLASS_SESSION_KEY = "_OIDCDB_CONFIG_CLASS"

mozilla_django_oidc_db/signals.py

@@ -1,12 +1,14 @@
 from django.contrib.auth.signals import user_logged_in
 from django.dispatch import receiver
 
+from .constants import CONFIG_CLASS_SESSION_KEY
 
-@receiver([user_logged_in], dispatch_uid="oidc.set_config_class")
+
+@receiver([user_logged_in], dispatch_uid="oidcdb.set_config_class")
 def set_oidcdb_config_class_on_session(sender, user, request, **kwargs):
     """
     Record the OIDC config class on the session, this is needed so the callback view
     can retrieve the config in case of a SessionRefresh flow
     """
     if hasattr(user, "_oidcdb_config_class"):
-        request.session["OIDCDB_CONFIG_CLASS"] = user._oidcdb_config_class
+        request.session[CONFIG_CLASS_SESSION_KEY] = user._oidcdb_config_class

pyproject.toml

@@ -51,6 +51,7 @@ tests = [
     "pytest-recording",
     "requests-mock",
     "factory-boy",
+    "freezegun",
     "pyquery",
     "tox",
     "isort",

tests/conftest.py

@@ -21,6 +21,10 @@ def mock_state_and_nonce(mocker):
         "mozilla_django_oidc.views.get_random_string",
         return_value="not-a-random-string",
     )
+    mocker.patch(
+        "mozilla_django_oidc.middleware.get_random_string",
+        return_value="not-a-random-string",
+    )
 
 
 @pytest.fixture

tests/test_callback_flow.py

@@ -8,6 +8,7 @@
 import pytest
 from pytest_django.asserts import assertContains, assertRedirects
 
+from mozilla_django_oidc_db.constants import CONFIG_CLASS_SESSION_KEY
 from mozilla_django_oidc_db.models import (
     OpenIDConnectConfig,
     UserInformationClaimsSources,
@@ -313,44 +314,6 @@ def test_set_config_class_on_session_after_successful_login(
     assert user.username == "some_username"
 
     assert (
-        callback_client.session["OIDCDB_CONFIG_CLASS"]
+        callback_client.session[CONFIG_CLASS_SESSION_KEY]
         == "mozilla_django_oidc_db.OpenIDConnectConfig"
     )
-
-
-@pytest.mark.mock_backend_claims(
-    {
-        "email": "nocollision@example.com",
-        "sub": "some_username",
-    }
-)
-@pytest.mark.auth_request(next="/admin/")
-@pytest.mark.oidcconfig(
-    enabled=True,
-    userinfo_claims_source=UserInformationClaimsSources.id_token,
-    make_users_staff=True,
-)
-def test_use_config_class_on_session_on_session_refresh(
-    dummy_config: OpenIDConnectConfig,
-    callback_request: HttpRequest,
-    callback_client: Client,
-    mock_auth_backend: MockBackend,
-):
-    """
-    Verify that it's possible to retrieve the OIDC db config class directly from the session
-    instead of relying on it being stored in the OIDC states. The latter does not work if
-    the user is redirected to the callback view in case of SessionRefresh
-    """
-    callback_url = reverse("oidc_authentication_callback")
-
-    session = callback_client.session
-    session["OIDCDB_CONFIG_CLASS"] = "mozilla_django_oidc_db.OpenIDConnectConfig"
-    del session["oidc_states"]
-    session.save()
-
-    # Should not crash when retrieving config
-    callback_response = callback_client.get(callback_url, {**callback_request.GET})
-
-    assertRedirects(
-        callback_response, "/admin/login/failure/", fetch_redirect_response=False
-    )

tests/test_integration_oidc_flow_variants.py

@@ -1,7 +1,13 @@
+from unittest.mock import patch
+from urllib.parse import parse_qs, urlparse
+
 from django.urls import reverse
 
 import pytest
+import requests
+from freezegun import freeze_time
 
+from mozilla_django_oidc_db.middleware import SessionRefresh
 from mozilla_django_oidc_db.models import (
     OpenIDConnectConfig,
     UserInformationClaimsSources,
@@ -110,3 +116,68 @@ def test_return_jwt_from_userinfo_endpoint(
 
     # a user was created
     assert django_user_model.objects.count() == 1
+
+
+@freeze_time("2024-01-01T12:00:00")
+@pytest.mark.vcr
+def test_session_refresh(
+    keycloak_config, settings, mock_state_and_nonce, client, django_user_model, vcr
+):
+    settings.MIDDLEWARE = settings.MIDDLEWARE + [
+        "mozilla_django_oidc_db.middleware.SessionRefresh"
+    ]
+    settings.OIDC_RENEW_ID_TOKEN_EXPIRY_SECONDS = 60
+    login_url = reverse("login")
+    django_login_response = client.get(login_url)
+    assert django_login_response.status_code == 302
+
+    # simulate login to Keycloak
+    redirect_uri = keycloak_login(django_login_response["Location"])
+
+    # complete the login flow on our end
+    callback_response = client.get(redirect_uri)
+
+    assert callback_response.status_code == 302
+    assert callback_response["Location"] == "/admin/"
+
+    # a user was created
+    assert django_user_model.objects.count() == 1
+
+    # check that the token request was performed as expected
+    token_request = next(
+        req
+        for req in vcr.requests
+        if req.uri == f"{KEYCLOAK_BASE_URL}protocol/openid-connect/token"
+        and req.method == "POST"
+    )
+    assert token_request is not None
+    assert b"client_id=testid" in token_request.body
+    assert b"secret=7DB3KUAAizYCcmZufpHRVOcD0TOkNO3I" in token_request.body
+    assert "Authorization" not in token_request.headers
+
+    # SessionRefresh should be called and should redirect user to Keycloak
+    with freeze_time("2024-01-01T13:00:00"):
+        admin_response = client.get(callback_response["Location"])
+
+        assert django_login_response.status_code == 302
+
+        redirect_url = urlparse(admin_response["Location"])
+        query_params = parse_qs(redirect_url.query)
+
+        assert redirect_url.path == "/realms/test/protocol/openid-connect/auth"
+        assert query_params["client_id"] == ["testid"]
+        assert query_params["response_type"] == ["code"]
+
+        keycloak_response = requests.get(
+            admin_response["Location"], allow_redirects=False
+        )
+
+        assert keycloak_response.status_code == 302
+        assert (
+            keycloak_response.headers["Location"]
+            == "http://testserver/oidc/callback/?error=login_required&state=not-a-random-string"
+        )
+
+        app_response = client.get(keycloak_response.headers["Location"])
+
+        assert app_response.status_code == 200