OpenConceptLab/ocl_online#535 | Percent-encode the query values in the Keycloak login, sign-up, reset and logout URLs

core/services/auth/openid.py

@@ -1,6 +1,7 @@
 import base64
 import logging
 import time
+from urllib.parse import quote, urlencode
 
 import requests
 from django.conf import settings
@@ -30,54 +31,69 @@ class OpenIDAuthService(AbstractAuthService):
     PAGE_SIZE = 200
     OIDP_ADMIN_TOKEN_URL = settings.OIDC_SERVER_INTERNAL_URL + '/realms/master/protocol/openid-connect/token'
 
+    @staticmethod
+    def build_url(endpoint, params, code_challenge=None, code_challenge_method=None):
+        # Every value is percent-encoded: Django has already decoded the query these come from, so a
+        # redirect_uri carrying ?, & or # would otherwise split into extra params or lose its tail.
+        if code_challenge and code_challenge_method:
+            params = {**params, 'code_challenge': code_challenge, 'code_challenge_method': code_challenge_method}
+        return f"{endpoint}?{urlencode(params, quote_via=quote)}"
+
     @staticmethod
     def get_login_redirect_url(client_id, redirect_uri, state, nonce, code_challenge=None, code_challenge_method=None):  # pylint: disable=line-too-long,too-many-arguments
         # response_type=code only (no id_token): a pure code flow never puts the id_token in the
         # callback URL/access logs. scope=openid is requested explicitly so the token-exchange
         # response still includes id_token in its body, where the frontend now reads it from.
-        url = f"{settings.OIDC_OP_AUTHORIZATION_ENDPOINT}?" \
-              f"response_type=code&" \
-              f"scope=openid profile email&" \
-              f"client_id={client_id}&" \
-              f"state={state}&" \
-              f"nonce={nonce}&" \
-              f"redirect_uri={redirect_uri}"
-        if code_challenge and code_challenge_method:
-            url += f"&code_challenge={code_challenge}&code_challenge_method={code_challenge_method}"
-        return url
+        return OpenIDAuthService.build_url(
+            settings.OIDC_OP_AUTHORIZATION_ENDPOINT,
+            {
+                'response_type': 'code',
+                'scope': 'openid profile email',
+                'client_id': client_id,
+                'state': state,
+                'nonce': nonce,
+                'redirect_uri': redirect_uri,
+            },
+            code_challenge, code_challenge_method
+        )
 
     @staticmethod
     def get_reset_password_redirect_url(client_id, redirect_uri, code_challenge=None, code_challenge_method=None):
-        url = f"{settings.OIDC_OP_AUTHORIZATION_ENDPOINT}?" \
-              f"scope=openid&" \
-              f"kc_action=UPDATE_PASSWORD&" \
-              f"response_type=code&" \
-              f"client_id={client_id}&" \
-              f"redirect_uri={redirect_uri}"
-        if code_challenge and code_challenge_method:
-            url += f"&code_challenge={code_challenge}&code_challenge_method={code_challenge_method}"
-        return url
+        return OpenIDAuthService.build_url(
+            settings.OIDC_OP_AUTHORIZATION_ENDPOINT,
+            {
+                'scope': 'openid',
+                'kc_action': 'UPDATE_PASSWORD',
+                'response_type': 'code',
+                'client_id': client_id,
+                'redirect_uri': redirect_uri,
+            },
+            code_challenge, code_challenge_method
+        )
 
     @staticmethod
     def get_registration_redirect_url(client_id, redirect_uri, state, nonce, code_challenge=None, code_challenge_method=None):  # pylint: disable=line-too-long,too-many-arguments
         # See get_login_redirect_url: pure code flow + explicit openid scope, id_token comes
         # from the token-exchange response body instead of the callback URL.
-        url = f"{settings.OIDC_OP_REGISTRATION_ENDPOINT}?" \
-              f"response_type=code&" \
-              f"scope=openid profile email&" \
-              f"client_id={client_id}&" \
-              f"state={state}&" \
-              f"nonce={nonce}&" \
-              f"redirect_uri={redirect_uri}"
-        if code_challenge and code_challenge_method:
-            url += f"&code_challenge={code_challenge}&code_challenge_method={code_challenge_method}"
-        return url
+        return OpenIDAuthService.build_url(
+            settings.OIDC_OP_REGISTRATION_ENDPOINT,
+            {
+                'response_type': 'code',
+                'scope': 'openid profile email',
+                'client_id': client_id,
+                'state': state,
+                'nonce': nonce,
+                'redirect_uri': redirect_uri,
+            },
+            code_challenge, code_challenge_method
+        )
 
     @staticmethod
     def get_logout_redirect_url(id_token_hint, redirect_uri):
-        return f"{settings.OIDC_OP_LOGOUT_ENDPOINT}?" \
-               f"id_token_hint={id_token_hint}&" \
-               f"post_logout_redirect_uri={redirect_uri}"
+        return OpenIDAuthService.build_url(
+            settings.OIDC_OP_LOGOUT_ENDPOINT,
+            {'id_token_hint': id_token_hint, 'post_logout_redirect_uri': redirect_uri}
+        )
 
     @staticmethod
     def credential_representation_from_hash(hash_, temporary=False):

core/services/auth/tests.py

@@ -1,4 +1,5 @@
 from unittest.mock import patch, Mock, ANY
+from urllib.parse import parse_qs, urlparse
 
 from core.services.auth.django import DjangoAuthService
 from core.services.auth.openid import OpenIDAuthService
@@ -25,17 +26,51 @@ class OpenIDAuthServiceTest(OCLTestCase):
     def test_get_login_redirect_url(self):
         self.assertEqual(
             OpenIDAuthService.get_login_redirect_url('client-id', 'http://localhost:4000', 'state', 'nonce'),
-            '/realms/ocl/protocol/openid-connect/auth?response_type=code&scope=openid profile email&'
-            'client_id=client-id&state=state&nonce=nonce&redirect_uri=http://localhost:4000'
+            '/realms/ocl/protocol/openid-connect/auth?response_type=code&scope=openid%20profile%20email&'
+            'client_id=client-id&state=state&nonce=nonce&redirect_uri=http%3A%2F%2Flocalhost%3A4000'
+        )
+        self.assertEqual(
+            OpenIDAuthService.get_login_redirect_url(
+                'client-id', 'http://localhost:4000', 'state', 'nonce', 'challenge', 'S256'),
+            '/realms/ocl/protocol/openid-connect/auth?response_type=code&scope=openid%20profile%20email&'
+            'client_id=client-id&state=state&nonce=nonce&redirect_uri=http%3A%2F%2Flocalhost%3A4000&'
+            'code_challenge=challenge&code_challenge_method=S256'
         )
 
     def test_get_logout_redirect_url(self):
         self.assertEqual(
             OpenIDAuthService.get_logout_redirect_url('id-token-hint', 'http://localhost:4000'),
             '/realms/ocl/protocol/openid-connect/logout?id_token_hint=id-token-hint&'
-            'post_logout_redirect_uri=http://localhost:4000'
+            'post_logout_redirect_uri=http%3A%2F%2Flocalhost%3A4000'
         )
 
+    def test_redirect_urls_keep_query_values_whole(self):
+        redirect_uri = 'http://localhost:4000/#/signin?returnTo=http%3A%2F%2Flocalhost%3A4000%2Fa&b=1'
+
+        def query(url):
+            parsed = urlparse(url)
+            self.assertEqual(parsed.fragment, '')
+            return parse_qs(parsed.query)
+
+        logout = query(OpenIDAuthService.get_logout_redirect_url('id-token', redirect_uri))
+        self.assertEqual(logout, {'id_token_hint': ['id-token'], 'post_logout_redirect_uri': [redirect_uri]})
+
+        for url in [
+                OpenIDAuthService.get_login_redirect_url('client', redirect_uri, 's&t', 'n#o', 'c', 'S256'),
+                OpenIDAuthService.get_registration_redirect_url('client', redirect_uri, 's&t', 'n#o', 'c', 'S256'),
+        ]:
+            params = query(url)
+            self.assertEqual(params['redirect_uri'], [redirect_uri])
+            self.assertEqual(params['state'], ['s&t'])
+            self.assertEqual(params['nonce'], ['n#o'])
+            self.assertEqual(params['scope'], ['openid profile email'])
+            self.assertEqual(params['code_challenge'], ['c'])
+
+        reset = query(OpenIDAuthService.get_reset_password_redirect_url('client', redirect_uri))
+        self.assertEqual(reset['redirect_uri'], [redirect_uri])
+        self.assertEqual(reset['kc_action'], ['UPDATE_PASSWORD'])
+        self.assertNotIn('code_challenge', reset)
+
     @patch('requests.post')
     def test_exchange_code_for_token(self, post_mock):
         post_mock.return_value = Mock(json=Mock(return_value={'token': 'token', 'foo': 'bar'}))