OIDC: Implement PKCE as server

src/pretix/base/migrations/0277_customerssoclient_require_pkce_and_more.py

@@ -0,0 +1,28 @@
+# Generated by Django 4.2.17 on 2025-02-07 16:42
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("pretixbase", "0276_item_hidden_if_item_available_mode"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="customerssoclient",
+            name="require_pkce",
+            field=models.BooleanField(default=False),
+        ),
+        migrations.AddField(
+            model_name="customerssogrant",
+            name="code_challenge",
+            field=models.TextField(null=True),
+        ),
+        migrations.AddField(
+            model_name="customerssogrant",
+            name="code_challenge_method",
+            field=models.CharField(max_length=255, null=True),
+        ),
+    ]

src/pretix/base/models/customers.py

@@ -416,6 +416,10 @@ class CustomerSSOClient(LoggedModel):
     authorization_grant_type = models.CharField(
         max_length=32, choices=GRANT_TYPES, verbose_name=_("Grant type"), default=GRANT_AUTHORIZATION_CODE,
     )
+    require_pkce = models.BooleanField(
+        verbose_name=_("Require PKCE extension"),
+        default=False,
+    )
     redirect_uris = models.TextField(
         blank=False,
         verbose_name=_("Redirection URIs"),
@@ -481,6 +485,8 @@ class CustomerSSOGrant(models.Model):
     expires = models.DateTimeField()
     redirect_uri = models.TextField()
     scope = models.TextField(blank=True)
+    code_challenge = models.TextField(blank=True, null=True)
+    code_challenge_method = models.CharField(max_length=255, blank=True, null=True)
 
 
 class CustomerSSOAccessToken(models.Model):

src/pretix/control/forms/organizer.py

@@ -1113,7 +1113,7 @@ class SSOClientForm(I18nModelForm):
     class Meta:
         model = CustomerSSOClient
         fields = ['is_active', 'name', 'client_id', 'client_type', 'authorization_grant_type', 'redirect_uris',
-                  'allowed_scopes']
+                  'allowed_scopes', 'require_pkce']
         widgets = {
             'authorization_grant_type': forms.RadioSelect,
             'client_type': forms.RadioSelect,

src/pretix/presale/views/oidc_op.py

@@ -72,6 +72,9 @@ We currently do not implement the following optional parts of the spec:
 We also implement the Discovery extension (without issuer discovery)
 as per https://openid.net/specs/openid-connect-discovery-1_0.html
 
+We also implement the PKCE extension for OAuth:
+https://www.rfc-editor.org/rfc/rfc7636
+
 The implementation passed the certification tests against the following profiles, but we did not
 acquire formal certification:
 
@@ -136,19 +139,22 @@ class AuthorizeView(View):
             self._construct_redirect_uri(redirect_uri, response_mode, qs)
         )
 
-    def _require_login(self, request, client, scope, redirect_uri, response_type, response_mode, state, nonce):
+    def _require_login(self, request, client, scope, redirect_uri, response_type, response_mode, state, nonce,
+                       code_challenge, code_challenge_method):
         form = AuthenticationForm(data=request.POST if "login-email" in request.POST else None, request=request,
                                   prefix="login")
         if "login-email" in request.POST and form.is_valid():
             customer_login(request, form.get_customer())
-            return self._success(client, scope, redirect_uri, response_type, response_mode, state, nonce, form.get_customer())
+            return self._success(client, scope, redirect_uri, response_type, response_mode, state, nonce,
+                                 code_challenge, code_challenge_method, form.get_customer())
         else:
             return render(request, 'pretixpresale/organizers/customer_login.html', {
                 'providers': [],
                 'form': form,
             })
 
-    def _success(self, client, scope, redirect_uri, response_type, response_mode, state, nonce, customer):
+    def _success(self, client, scope, redirect_uri, response_type, response_mode, state, nonce, code_challenge,
+                 code_challenge_method, customer):
         response_type = response_type.split(' ')
         qs = {}
         id_token_kwargs = {}
@@ -162,6 +168,8 @@ class AuthorizeView(View):
                 expires=now() + timedelta(minutes=10),
                 auth_time=get_customer_auth_time(self.request),
                 nonce=nonce,
+                code_challenge=code_challenge,
+                code_challenge_method=code_challenge_method,
             )
             qs['code'] = grant.code
             id_token_kwargs['with_code'] = grant.code
@@ -209,6 +217,8 @@ class AuthorizeView(View):
         prompt = request_data.get("prompt")
         response_type = request_data.get("response_type")
         scope = request_data.get("scope", "").split(" ")
+        code_challenge = request_data.get("code_challenge")
+        code_challenge_method = request_data.get("code_challenge_method")
 
         if not client_id:
             return self._final_error("invalid_request", "client_id missing")
@@ -247,6 +257,16 @@ class AuthorizeView(View):
             self._redirect_error("invalid_request", "id_token_hint currently not supported by this server",
                                  redirect_uri, response_mode, state)
 
+        if code_challenge and code_challenge_method != "S256":
+            # "Clients re permitted to use "plain" only if they cannot support "S256" for some technical reason and
+            # know via out-of-band configuration that the S256 MUST be implemented, plain is not mandatory."
+            return self._redirect_error("invalid_request", "code_challenge transform algorithm not supported",
+                                        redirect_uri, response_mode, state)
+
+        if client.require_pkce and not code_challenge:
+            return self._redirect_error("invalid_request", "code_challenge (PKCE) required",
+                                        redirect_uri, response_mode, state)
+
         has_valid_session = bool(request.customer)
         if has_valid_session and max_age:
             try:
@@ -261,9 +281,11 @@ class AuthorizeView(View):
             has_valid_session = False
 
         if has_valid_session:
-            return self._success(client, scope, redirect_uri, response_type, response_mode, state, nonce, request.customer)
+            return self._success(client, scope, redirect_uri, response_type, response_mode, state, nonce, code_challenge,
+                                 code_challenge_method, request.customer)
         else:
-            return self._require_login(request, client, scope, redirect_uri, response_type, response_mode, state, nonce)
+            return self._require_login(request, client, scope, redirect_uri, response_type, response_mode, state, nonce,
+                                       code_challenge, code_challenge_method)
 
 
 class TokenView(View):
@@ -361,6 +383,24 @@ class TokenView(View):
                 "error_description": "Mismatch of redirect_uri"
             }, status=400)
 
+        if grant.code_challenge:
+            if not request.POST.get("code_verifier"):
+                return JsonResponse({
+                    "error": "invalid_grant",
+                    "error_description": "Missing of code_verifier"
+                }, status=400)
+
+            if grant.code_challenge_method == "S256":
+                expected_challenge = base64.urlsafe_b64encode(hashlib.sha256(request.POST["code_verifier"].encode()).digest()).decode().rstrip("=")
+                print(grant.code_challenge, expected_challenge)
+                if expected_challenge != grant.code_challenge:
+                    return JsonResponse({
+                        "error": "invalid_grant",
+                        "error_description": "Mismatch of code_verifier with code_challenge"
+                    }, status=400)
+            else:
+                raise ValueError("Unsupported code_challenge_method in database")
+
         with transaction.atomic():
             token = self.client.access_tokens.create(
                 customer=grant.customer,

src/tests/presale/test_oidc_op.py

@@ -208,6 +208,37 @@ def test_authorize_with_prompt_none(env, client, ssoclient):
     assert re.match(r'https://example.net\?code=([a-z0-9A-Z]{64})&state=STATE', r.headers['Location'])
 
 
+@pytest.mark.django_db
+def test_authorize_with_invalid_pkce_method(env, client, ssoclient):
+    url = f'/bigevents/oauth2/v1/authorize?' \
+          f'client_id={ssoclient[0].client_id}&' \
+          f'redirect_uri=https://example.net&' \
+          f'response_type=code&state=STATE&scope=openid+profile&' \
+          f'code_challenge=pkce_value&code_challenge_method=plain'
+    r = client.get(url)
+    assert r.status_code == 302
+    assert r.headers['Location'] == 'https://example.net?' \
+                                    'error=invalid_request&' \
+                                    'error_description=code_challenge+transform+algorithm+not+supported&' \
+                                    'state=STATE'
+
+
+@pytest.mark.django_db
+def test_authorize_with_missing_pkce_if_required(env, client, ssoclient):
+    ssoclient[0].require_pkce = True
+    ssoclient[0].save()
+    url = f'/bigevents/oauth2/v1/authorize?' \
+          f'client_id={ssoclient[0].client_id}&' \
+          f'redirect_uri=https://example.net&' \
+          f'response_type=code&state=STATE&scope=openid+profile'
+    r = client.get(url)
+    assert r.status_code == 302
+    assert r.headers['Location'] == 'https://example.net?' \
+                                    'error=invalid_request&' \
+                                    'error_description=code_challenge+%28PKCE%29+required&' \
+                                    'state=STATE'
+
+
 @pytest.mark.django_db
 def test_authorize_require_login_if_prompt_requires_it_or_is_expired(env, client, ssoclient):
     with freeze_time("2021-04-10T11:00:00+02:00"):
@@ -286,7 +317,7 @@ def test_token_require_client_id(env, client, ssoclient):
     assert b'unsupported_grant_type' in r.content
 
 
-def _authorization_step(client, ssoclient):
+def _authorization_step(client, ssoclient, code_challenge=None):
     r = client.post('/bigevents/account/login', {
         'email': 'john@example.org',
         'password': 'foo',
@@ -299,6 +330,8 @@ def _authorization_step(client, ssoclient):
           f'client_id={ssoclient[0].client_id}&' \
           f'redirect_uri=https://example.net&' \
           f'response_type=code&state=STATE&scope=openid+profile+email+phone'
+    if code_challenge:
+        url += f'&code_challenge={code_challenge}&code_challenge_method=S256'
     r = client.get(url)
     assert r.status_code == 302
     m = re.match(r'https://example.net\?code=([a-z0-9A-Z]{64})&state=STATE', r.headers['Location'])
@@ -373,6 +406,55 @@ def test_token_success(env, client, ssoclient):
         CustomerSSOAccessToken.objects.get(token=d['access_token']).expires < now()
 
 
+@pytest.mark.django_db
+def test_token_pkce_required_if_used_in_authorization(env, client, ssoclient):
+    code = _authorization_step(client, ssoclient, "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM")
+
+    r = client.post('/bigevents/oauth2/v1/token', {
+        'grant_type': 'authorization_code',
+        'code': code,
+        'redirect_uri': 'https://example.net',
+    }, HTTP_AUTHORIZATION='Basic ' + base64.b64encode(f'{ssoclient[0].client_id}:{ssoclient[1]}'.encode()).decode())
+    assert r.status_code == 400
+    d = json.loads(r.content)
+    assert d['error'] == 'invalid_grant'
+    assert d['error_description'] == 'Missing of code_verifier'
+
+
+@pytest.mark.django_db
+def test_token_pkce_incorrect(env, client, ssoclient):
+    code = _authorization_step(client, ssoclient, "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM")
+
+    r = client.post('/bigevents/oauth2/v1/token', {
+        'grant_type': 'authorization_code',
+        'code': code,
+        'redirect_uri': 'https://example.net',
+        'code_verifier': "WRONG",
+    }, HTTP_AUTHORIZATION='Basic ' + base64.b64encode(f'{ssoclient[0].client_id}:{ssoclient[1]}'.encode()).decode())
+    assert r.status_code == 400
+    d = json.loads(r.content)
+    assert d['error'] == 'invalid_grant'
+    assert d['error_description'] == 'Mismatch of code_verifier with code_challenge'
+
+
+@pytest.mark.django_db
+def test_token_success_pkce(env, client, ssoclient):
+    # this is the sample from the actual RFC
+    code_verifier = "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"
+    code = _authorization_step(client, ssoclient, "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM")
+
+    r = client.post('/bigevents/oauth2/v1/token', {
+        'grant_type': 'authorization_code',
+        'code': code,
+        'redirect_uri': 'https://example.net',
+        'code_verifier': code_verifier,
+    }, HTTP_AUTHORIZATION='Basic ' + base64.b64encode(f'{ssoclient[0].client_id}:{ssoclient[1]}'.encode()).decode())
+    print(r.content)
+    assert r.status_code == 200
+    d = json.loads(r.content)
+    assert d['access_token']
+
+
 @pytest.mark.django_db
 def test_scope_enforcement(env, client, ssoclient):
     ssoclient[0].allowed_scopes = ['openid', 'profile']