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)
 

oauth2_provider/views/oidc.py

@@ -83,10 +83,7 @@ def get(self, request, *args, **kwargs):
 
         signing_algorithms = [Application.HS256_ALGORITHM]
         if oauth2_settings.OIDC_RSA_PRIVATE_KEY:
-            signing_algorithms = [
-                Application.RS256_ALGORITHM,
-                Application.HS256_ALGORITHM,
-            ]
+            signing_algorithms = [Application.RS256_ALGORITHM, Application.HS256_ALGORITHM]
 
         validator_class = oauth2_settings.OAUTH2_VALIDATOR_CLASS
         validator = validator_class()
@@ -251,10 +248,7 @@ class RPInitiatedLogoutView(OIDCLogoutOnlyMixin, FormView):
     form_class = ConfirmLogoutForm
     # Only delete tokens for Application whose client type and authorization
     # grant type are in the respective lists.
-    token_deletion_client_types = [
-        Application.CLIENT_PUBLIC,
-        Application.CLIENT_CONFIDENTIAL,
-    ]
+    token_deletion_client_types = [Application.CLIENT_PUBLIC, Application.CLIENT_CONFIDENTIAL]
     token_deletion_grant_types = [
         Application.GRANT_AUTHORIZATION_CODE,
         Application.GRANT_IMPLICIT,
@@ -458,13 +452,7 @@ def must_prompt(self, token_user):
         """ We didn't find a reason to prompt the user """
         return False
 
-    def do_logout(
-        self,
-        application=None,
-        post_logout_redirect_uri=None,
-        state=None,
-        token_user=None,
-    ):
+    def do_logout(self, application=None, post_logout_redirect_uri=None, state=None, token_user=None):
         user = token_user or self.request.user
         # Delete Access Tokens if a user was found
         if oauth2_settings.OIDC_RP_INITIATED_LOGOUT_DELETE_TOKENS and not isinstance(user, AnonymousUser):
@@ -501,8 +489,7 @@ def do_logout(
                 return OAuth2ResponseRedirect(post_logout_redirect_uri, application.get_allowed_schemes())
         else:
             return OAuth2ResponseRedirect(
-                self.request.build_absolute_uri("/"),
-                oauth2_settings.ALLOWED_REDIRECT_URI_SCHEMES,
+                self.request.build_absolute_uri("/"), oauth2_settings.ALLOWED_REDIRECT_URI_SCHEMES
             )
 
     def error_response(self, error):

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,7 +12,12 @@
     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
@@ -132,7 +137,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 +214,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 +379,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/"
@@ -467,10 +512,7 @@ def test_rp_initiated_logout_expired_tokens_accept(logged_in_client, application
     # Accepting expired (but otherwise valid and signed by us) tokens is enabled. Logout should go through.
     rsp = logged_in_client.get(
         reverse("oauth2_provider:rp-initiated-logout"),
-        data={
-            "id_token_hint": expired_id_token,
-            "client_id": application.client_id,
-        },
+        data={"id_token_hint": expired_id_token, "client_id": application.client_id},
     )
     assert rsp.status_code == 302
     assert not is_logged_in(logged_in_client)
@@ -482,10 +524,7 @@ def test_rp_initiated_logout_expired_tokens_deny(logged_in_client, application,
     # Expired tokens should not be accepted by default.
     rsp = logged_in_client.get(
         reverse("oauth2_provider:rp-initiated-logout"),
-        data={
-            "id_token_hint": expired_id_token,
-            "client_id": application.client_id,
-        },
+        data={"id_token_hint": expired_id_token, "client_id": application.client_id},
     )
     assert rsp.status_code == 400
     assert is_logged_in(logged_in_client)