Add test for session state on authorization view

Author: lullis

oauth2_provider/settings.py

@@ -173,12 +173,7 @@ def import_from_string(val, setting_name):
     try:
         return import_string(val)
     except ImportError as e:
-        msg = "Could not import %r for setting %r. %s: %s." % (
-            val,
-            setting_name,
-            e.__class__.__name__,
-            e,
-        )
+        msg = "Could not import %r for setting %r. %s: %s." % (val, setting_name, e.__class__.__name__, e)
         raise ImportError(msg)
 
 

oauth2_provider/urls.py

@@ -17,11 +17,7 @@
 management_urlpatterns = [
     # Application management views
     path("applications/", views.ApplicationList.as_view(), name="list"),
-    path(
-        "applications/register/",
-        views.ApplicationRegistration.as_view(),
-        name="register",
-    ),
+    path("applications/register/", views.ApplicationRegistration.as_view(), name="register"),
     path("applications/<slug:pk>/", views.ApplicationDetail.as_view(), name="detail"),
     path("applications/<slug:pk>/delete/", views.ApplicationDelete.as_view(), name="delete"),
     path("applications/<slug:pk>/update/", views.ApplicationUpdate.as_view(), name="update"),

oauth2_provider/views/base.py

@@ -137,10 +137,7 @@ def form_valid(self, form):
 
         try:
             uri, headers, body, status = self.create_authorization_response(
-                request=self.request,
-                scopes=scopes,
-                credentials=credentials,
-                allow=allow,
+                request=self.request, scopes=scopes, credentials=credentials, allow=allow
             )
         except OAuthToolkitError as error:
             return self.error_response(error, application)
@@ -160,7 +157,7 @@ def form_valid(self, form):
             salt = secrets.token_urlsafe(16)
             encoded = " ".join(
                 [
-                    self.client.client_id,
+                    credentials["client_id"],
                     client_origin,
                     session_management_state_key(self.request),
                     salt,
@@ -231,20 +228,15 @@ def get(self, request, *args, **kwargs):
             # are already approved.
             if application.skip_authorization:
                 uri, headers, body, status = self.create_authorization_response(
-                    request=self.request,
-                    scopes=" ".join(scopes),
-                    credentials=credentials,
-                    allow=True,
+                    request=self.request, scopes=" ".join(scopes), credentials=credentials, allow=True
                 )
                 return self.redirect(uri, application)
 
             elif require_approval == "auto":
                 tokens = (
                     get_access_token_model()
                     .objects.filter(
-                        user=request.user,
-                        application=kwargs["application"],
-                        expires__gt=timezone.now(),
+                        user=request.user, application=kwargs["application"], expires__gt=timezone.now()
                     )
                     .all()
                 )
@@ -253,10 +245,7 @@ def get(self, request, *args, **kwargs):
                 for token in tokens:
                     if token.allow_scopes(scopes):
                         uri, headers, body, status = self.create_authorization_response(
-                            request=self.request,
-                            scopes=" ".join(scopes),
-                            credentials=credentials,
-                            allow=True,
+                            request=self.request, scopes=" ".join(scopes), credentials=credentials, allow=True
                         )
                         return self.redirect(uri, application)
 

tests/presets.py

@@ -37,6 +37,8 @@
 OIDC_SETTINGS_RP_LOGOUT_DENY_EXPIRED["OIDC_RP_INITIATED_LOGOUT_ACCEPT_EXPIRED_TOKENS"] = False
 OIDC_SETTINGS_RP_LOGOUT_KEEP_TOKENS = deepcopy(OIDC_SETTINGS_RP_LOGOUT)
 OIDC_SETTINGS_RP_LOGOUT_KEEP_TOKENS["OIDC_RP_INITIATED_LOGOUT_DELETE_TOKENS"] = False
+OIDC_SETTINGS_SESSION_MANAGEMENT = deepcopy(OIDC_SETTINGS_RW)
+OIDC_SETTINGS_SESSION_MANAGEMENT["OIDC_SESSION_MANAGEMENT_ENABLED"] = True
 REST_FRAMEWORK_SCOPES = {
     "SCOPES": {
         "read": "Read scope",

tests/test_oidc_views.py

@@ -1,5 +1,5 @@
 import pytest
-from django.contrib.auth import get_user
+from django.contrib.auth import get_user, get_user_model
 from django.contrib.auth.models import AnonymousUser
 from django.test import RequestFactory
 from django.urls import reverse
@@ -12,10 +12,19 @@
     InvalidOIDCClientError,
     InvalidOIDCRedirectURIError,
 )
-from oauth2_provider.models import get_access_token_model, get_id_token_model, get_refresh_token_model
+from oauth2_provider.models import (
+    get_access_token_model,
+    get_application_model,
+    get_id_token_model,
+    get_refresh_token_model,
+)
 from oauth2_provider.oauth2_validators import OAuth2Validator
 from oauth2_provider.settings import oauth2_settings
-from oauth2_provider.views.oidc import RPInitiatedLogoutView, _load_id_token, _validate_claims
+from oauth2_provider.views.oidc import (
+    RPInitiatedLogoutView,
+    _load_id_token,
+    _validate_claims,
+)
 
 from . import presets
 from .common_testing import OAuth2ProviderTestCase as TestCase
@@ -44,7 +53,10 @@ def test_get_connect_discovery_info(self):
             ],
             "subject_types_supported": ["public"],
             "id_token_signing_alg_values_supported": ["RS256", "HS256"],
-            "token_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
+            "token_endpoint_auth_methods_supported": [
+                "client_secret_post",
+                "client_secret_basic",
+            ],
             "code_challenge_methods_supported": ["plain", "S256"],
             "claims_supported": ["sub"],
         }
@@ -71,7 +83,10 @@ def test_get_connect_discovery_info_deprecated(self):
             ],
             "subject_types_supported": ["public"],
             "id_token_signing_alg_values_supported": ["RS256", "HS256"],
-            "token_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
+            "token_endpoint_auth_methods_supported": [
+                "client_secret_post",
+                "client_secret_basic",
+            ],
             "code_challenge_methods_supported": ["plain", "S256"],
             "claims_supported": ["sub"],
         }
@@ -98,7 +113,10 @@ def expect_json_response_with_rp_logout(self, base):
             ],
             "subject_types_supported": ["public"],
             "id_token_signing_alg_values_supported": ["RS256", "HS256"],
-            "token_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
+            "token_endpoint_auth_methods_supported": [
+                "client_secret_post",
+                "client_secret_basic",
+            ],
             "code_challenge_methods_supported": ["plain", "S256"],
             "claims_supported": ["sub"],
             "end_session_endpoint": f"{base}/logout/",
@@ -132,7 +150,10 @@ def test_get_connect_discovery_info_without_issuer_url(self):
             ],
             "subject_types_supported": ["public"],
             "id_token_signing_alg_values_supported": ["RS256", "HS256"],
-            "token_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
+            "token_endpoint_auth_methods_supported": [
+                "client_secret_post",
+                "client_secret_basic",
+            ],
             "code_challenge_methods_supported": ["plain", "S256"],
             "claims_supported": ["sub"],
         }
@@ -206,6 +227,42 @@ def test_get_jwks_info_multiple_rsa_keys(self):
         assert response.json() == expected_response
 
 
+@pytest.mark.usefixtures("oauth2_settings")
+@pytest.mark.oauth2_settings(presets.OIDC_SETTINGS_SESSION_MANAGEMENT)
+class TestAuthorizationView(TestCase):
+    def test_session_state_is_present_in_url(self):
+        User = get_user_model()
+        Application = get_application_model()
+
+        User.objects.create_user("test_user", "[email protected]", "123456")
+        dev_user = User.objects.create_user("dev_user", "[email protected]", "123456")
+
+        application = Application.objects.create(
+            name="Test Application",
+            redirect_uris=(
+                "http://localhost http://example.com http://example.org custom-scheme://example.com"
+            ),
+            user=dev_user,
+            client_type=Application.CLIENT_CONFIDENTIAL,
+            authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+            client_secret="1234567890qwertyuiop",
+        )
+        self.client.login(username="test_user", password="123456")
+        response = self.client.post(
+            reverse("oauth2_provider:authorize"),
+            {
+                "client_id": application.client_id,
+                "response_type": "code",
+                "state": "random_state_string",
+                "scope": "read write",
+                "redirect_uri": "http://example.org",
+                "allow": True,
+            },
+        )
+        self.assertEqual(response.status_code, 302)
+        self.assertTrue("session_state" in response["Location"])
+
+
 def mock_request():
     """
     Dummy request with an AnonymousUser attached.
@@ -335,7 +392,8 @@ def test_rp_initiated_logout_get(logged_in_client, rp_settings):
 @pytest.mark.django_db(databases=retrieve_current_databases())
 def test_rp_initiated_logout_get_id_token(logged_in_client, oidc_tokens, rp_settings):
     rsp = logged_in_client.get(
-        reverse("oauth2_provider:rp-initiated-logout"), data={"id_token_hint": oidc_tokens.id_token}
+        reverse("oauth2_provider:rp-initiated-logout"),
+        data={"id_token_hint": oidc_tokens.id_token},
     )
     assert rsp.status_code == 302
     assert rsp["Location"] == "http://testserver/"
@@ -347,7 +405,8 @@ def test_rp_initiated_logout_get_revoked_id_token(logged_in_client, oidc_tokens,
     validator = oauth2_settings.OAUTH2_VALIDATOR_CLASS()
     validator._load_id_token(oidc_tokens.id_token).revoke()
     rsp = logged_in_client.get(
-        reverse("oauth2_provider:rp-initiated-logout"), data={"id_token_hint": oidc_tokens.id_token}
+        reverse("oauth2_provider:rp-initiated-logout"),
+        data={"id_token_hint": oidc_tokens.id_token},
     )
     assert rsp.status_code == 400
     assert is_logged_in(logged_in_client)
@@ -357,7 +416,10 @@ def test_rp_initiated_logout_get_revoked_id_token(logged_in_client, oidc_tokens,
 def test_rp_initiated_logout_get_id_token_redirect(logged_in_client, oidc_tokens, rp_settings):
     rsp = logged_in_client.get(
         reverse("oauth2_provider:rp-initiated-logout"),
-        data={"id_token_hint": oidc_tokens.id_token, "post_logout_redirect_uri": "http://example.org"},
+        data={
+            "id_token_hint": oidc_tokens.id_token,
+            "post_logout_redirect_uri": "http://example.org",
+        },
     )
     assert rsp.status_code == 302
     assert rsp["Location"] == "http://example.org"
@@ -385,7 +447,10 @@ def test_rp_initiated_logout_get_id_token_missmatch_client_id(
 ):
     rsp = logged_in_client.get(
         reverse("oauth2_provider:rp-initiated-logout"),
-        data={"id_token_hint": oidc_tokens.id_token, "client_id": public_application.client_id},
+        data={
+            "id_token_hint": oidc_tokens.id_token,
+            "client_id": public_application.client_id,
+        },
     )
     assert rsp.status_code == 400
     assert is_logged_in(logged_in_client)
@@ -427,7 +492,8 @@ def test_rp_initiated_logout_public_client_strict_redirect_client_id(
 @pytest.mark.django_db(databases=retrieve_current_databases())
 def test_rp_initiated_logout_get_client_id(logged_in_client, oidc_tokens, rp_settings):
     rsp = logged_in_client.get(
-        reverse("oauth2_provider:rp-initiated-logout"), data={"client_id": oidc_tokens.application.client_id}
+        reverse("oauth2_provider:rp-initiated-logout"),
+        data={"client_id": oidc_tokens.application.client_id},
     )
     assert rsp.status_code == 200
     assert is_logged_in(logged_in_client)