OIDC: Implement PKCE in client implementation

src/pretix/base/customersso/oidc.py

@@ -148,7 +148,7 @@ def oidc_validate_and_complete_config(config):
     return config
 
 
-def oidc_authorize_url(provider, state, redirect_uri):
+def oidc_authorize_url(provider, state, redirect_uri, pkce_code_verifier):
     endpoint = provider.configuration['provider_config']['authorization_endpoint']
     params = {
         # https://datatracker.ietf.org/doc/html/rfc6749#section-4.1.1
@@ -163,10 +163,14 @@ def oidc_authorize_url(provider, state, redirect_uri):
     if "query_parameters" in provider.configuration and provider.configuration["query_parameters"]:
         params.update(parse_qsl(provider.configuration["query_parameters"]))
 
+    if pkce_code_verifier:
+        params["code_challenge"] = base64.urlsafe_b64encode(hashlib.sha256(pkce_code_verifier.encode()).digest()).decode().rstrip("=")
+        params["code_challenge_method"] = "S256"
+
     return endpoint + '?' + urlencode(params)
 
 
-def oidc_validate_authorization(provider, code, redirect_uri):
+def oidc_validate_authorization(provider, code, redirect_uri, pkce_code_verifier):
     endpoint = provider.configuration['provider_config']['token_endpoint']
 
     # Wall of shame and RFC ignorant IDPs
@@ -188,6 +192,9 @@ def oidc_validate_authorization(provider, code, redirect_uri):
         'redirect_uri': redirect_uri,
     }
 
+    if pkce_code_verifier:
+        params["code_verifier"] = pkce_code_verifier
+
     if token_endpoint_auth_method == 'client_secret_post':
         params['client_id'] = provider.configuration['client_id']
         params['client_secret'] = provider.configuration['client_secret']

src/pretix/presale/views/customer.py

@@ -676,6 +676,8 @@ class SSOLoginView(RedirectBackMixin, View):
                 popup_origin = None
 
         nonce = get_random_string(32)
+        pkce_code_verifier = get_random_string(64)
+        request.session[f'pretix_customerauth_{self.provider.pk}_pkce_code_verifier'] = pkce_code_verifier
         request.session[f'pretix_customerauth_{self.provider.pk}_nonce'] = nonce
         request.session[f'pretix_customerauth_{self.provider.pk}_popup_origin'] = popup_origin
         request.session[f'pretix_customerauth_{self.provider.pk}_cross_domain_requested'] = self.request.GET.get("request_cross_domain_customer_auth") == "true"
@@ -684,7 +686,7 @@ class SSOLoginView(RedirectBackMixin, View):
         })
 
         if self.provider.method == "oidc":
-            return redirect_to_url(oidc_authorize_url(self.provider, f'{nonce}%{next_url}', redirect_uri))
+            return redirect_to_url(oidc_authorize_url(self.provider, f'{nonce}%{next_url}', redirect_uri, pkce_code_verifier))
         else:
             raise Http404("Unknown SSO method.")
 
@@ -718,6 +720,7 @@ class SSOLoginReturnView(RedirectBackMixin, View):
                 )
             return HttpResponseRedirect(redirect_to)
         r = super().dispatch(request, *args, **kwargs)
+        request.session.pop(f'pretix_customerauth_{self.provider.pk}_pkce_code_verifier', None)
         request.session.pop(f'pretix_customerauth_{self.provider.pk}_nonce', None)
         request.session.pop(f'pretix_customerauth_{self.provider.pk}_popup_origin', None)
         request.session.pop(f'pretix_customerauth_{self.provider.pk}_cross_domain_requested', None)
@@ -763,6 +766,7 @@ class SSOLoginReturnView(RedirectBackMixin, View):
                     self.provider,
                     request.GET.get('code'),
                     redirect_uri,
+                    request.session.get(f'pretix_customerauth_{self.provider.pk}_pkce_code_verifier'),
                 )
             except ValidationError as e:
                 for msg in e:

src/tests/base/test_customer_oidc_rp.py

@@ -250,8 +250,10 @@ def test_authorize_url(provider):
         "client_id=abc123&"
         "scope=openid+email+profile&"
         "state=state_val&"
-        "redirect_uri=https%3A%2F%2Fredirect%3Ffoo%3Dbar"
-    ) == oidc_authorize_url(provider, "state_val", "https://redirect?foo=bar")
+        "redirect_uri=https%3A%2F%2Fredirect%3Ffoo%3Dbar&"
+        "code_challenge=S1ZnvzwMZHrWOO62nENdJ6jhODhf7VfyZFBIXQyrTKo&"
+        "code_challenge_method=S256"
+    ) == oidc_authorize_url(provider, "state_val", "https://redirect?foo=bar", "pkce_value")
 
 
 @pytest.mark.django_db
@@ -264,7 +266,7 @@ def test_validate_authorization_invalid(provider):
         status=400,
     )
     with pytest.raises(ValidationError):
-        oidc_validate_authorization(provider, "code_received", "https://redirect?foo=bar")
+        oidc_validate_authorization(provider, "code_received", "https://redirect?foo=bar", "pkce_value")
 
 
 @pytest.mark.django_db
@@ -281,6 +283,7 @@ def test_validate_authorization_userinfo_invalid(provider):
                 "grant_type": "authorization_code",
                 "code": "code_received",
                 "redirect_uri": "https://redirect?foo=bar",
+                "code_verifier": "pkce_value",
             })
         ],
     )
@@ -296,7 +299,7 @@ def test_validate_authorization_userinfo_invalid(provider):
         ],
     )
     with pytest.raises(ValidationError) as e:
-        oidc_validate_authorization(provider, "code_received", "https://redirect?foo=bar")
+        oidc_validate_authorization(provider, "code_received", "https://redirect?foo=bar", "pkce_value")
     assert 'could not fetch' in str(e.value)
 
 
@@ -314,6 +317,7 @@ def test_validate_authorization_valid(provider):
                 "grant_type": "authorization_code",
                 "code": "code_received",
                 "redirect_uri": "https://redirect?foo=bar",
+                "code_verifier": "pkce_value",
             })
         ],
     )
@@ -328,4 +332,4 @@ def test_validate_authorization_valid(provider):
             matchers.header_matcher({"Authorization": "Bearer test_access_token"})
         ],
     )
-    oidc_validate_authorization(provider, "code_received", "https://redirect?foo=bar")
+    oidc_validate_authorization(provider, "code_received", "https://redirect?foo=bar", "pkce_value")