Add pkce essential per client

docs/source/contents/conf.rst

@@ -84,6 +84,27 @@ An example::
         }
       }
 
+The provided add-ons can be seen in the following sections.
+
+pkce
+####
+
+The pkce add on is activated using the ``oidcop.oidc.add_on.pkce.add_pkce_support``
+function. The possible configuration options can be found below.
+
+essential
+---------
+
+Whether pkce is mandatory, authentication requests without a ``code_challenge``
+will fail if this is True. This option can be overridden per client by defining
+``pkce_essential`` in the client metadata.
+
+code_challenge_method
+---------------------
+
+The allowed code_challenge methods. The supported code challenge methods are:
+``plain, S256, S384, S512``
+
 --------------
 authentication
 --------------

src/oidcop/oidc/add_on/pkce.py

@@ -3,11 +3,9 @@
 from typing import Dict
 
 from cryptojwt.utils import b64e
-from oidcmsg.oauth2 import (
-    AuthorizationErrorResponse,
-    RefreshAccessTokenRequest,
-    TokenExchangeRequest,
-)
+from oidcmsg.oauth2 import AuthorizationErrorResponse
+from oidcmsg.oauth2 import RefreshAccessTokenRequest
+from oidcmsg.oauth2 import TokenExchangeRequest
 from oidcmsg.oidc import TokenErrorResponse
 
 from oidcop.endpoint import Endpoint
@@ -41,7 +39,14 @@ def post_authn_parse(request, client_id, endpoint_context, **kwargs):
     :param kwargs:
     :return:
     """
-    if endpoint_context.args["pkce"]["essential"] and "code_challenge" not in request:
+    client = endpoint_context.cdb[client_id]
+    if "pkce_essential" in client:
+        essential = client["pkce_essential"]
+    else:
+        essential = endpoint_context.args["pkce"].get(
+            "essential", False
+        )
+    if essential and "code_challenge" not in request:
         return AuthorizationErrorResponse(
             error="invalid_request", error_description="Missing required code_challenge",
         )
@@ -131,9 +136,6 @@ def add_pkce_support(endpoint: Dict[str, Endpoint], **kwargs):
     authn_endpoint.post_parse_request.append(post_authn_parse)
     token_endpoint.post_parse_request.append(post_token_parse)
 
-    if "essential" not in kwargs:
-        kwargs["essential"] = False
-
     code_challenge_methods = kwargs.get("code_challenge_methods", CC_METHOD.keys())
 
     kwargs["code_challenge_methods"] = {}

tests/test_33_oauth2_pkce.py

@@ -6,6 +6,7 @@
 
 from . import full_path
 from oidcop.configure import ASConfiguration
+
 import pytest
 import yaml
 from oidcmsg.message import Message
@@ -16,6 +17,7 @@
 from oidcmsg.oidc import TokenErrorResponse
 
 import oidcop.oauth2.introspection
+from oidcop.configure import ASConfiguration
 from oidcop.configure import OPConfiguration
 from oidcop.cookie_handler import CookieHandler
 from oidcop.endpoint import Endpoint
@@ -290,6 +292,41 @@ def test_not_essential(self, conf):
 
         assert isinstance(_req, Message)
 
+    def test_essential_per_client(self, conf):
+        conf["add_on"]["pkce"]["kwargs"]["essential"] = False
+        server = create_server(conf)
+        authn_endpoint = server.server_get("endpoint", "authorization")
+        token_endpoint = server.server_get("endpoint", "token")
+        _authn_req = AUTH_REQ.copy()
+        endpoint_context = server.server_get("endpoint_context")
+        endpoint_context.cdb[AUTH_REQ["client_id"]]["pkce_essential"] = True
+
+        _pr_resp = authn_endpoint.parse_request(_authn_req.to_dict())
+
+        assert isinstance(_pr_resp, AuthorizationErrorResponse)
+        assert _pr_resp["error"] == "invalid_request"
+        assert _pr_resp["error_description"] == "Missing required code_challenge"
+
+    def test_not_essential_per_client(self, conf):
+        conf["add_on"]["pkce"]["kwargs"]["essential"] = True
+        server = create_server(conf)
+        authn_endpoint = server.server_get("endpoint", "authorization")
+        token_endpoint = server.server_get("endpoint", "token")
+        _authn_req = AUTH_REQ.copy()
+        endpoint_context = server.server_get("endpoint_context")
+        endpoint_context.cdb[AUTH_REQ["client_id"]]["pkce_essential"] = False
+
+        _pr_resp = authn_endpoint.parse_request(_authn_req.to_dict())
+        resp = authn_endpoint.process_request(_pr_resp)
+
+        assert isinstance(resp["response_args"], AuthorizationResponse)
+
+        _token_request = TOKEN_REQ.copy()
+        _token_request["code"] = resp["response_args"]["code"]
+        _req = token_endpoint.parse_request(_token_request)
+
+        assert isinstance(_req, Message)
+
     def test_unknown_code_challenge_method(self):
         _authn_req = AUTH_REQ.copy()
         _authn_req["code_challenge"] = "aba"