Add pkce per client

Author: nsklikas

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,12 @@ 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"]["essential"]
+    if essential and "code_challenge" not in request:
         return AuthorizationErrorResponse(
             error="invalid_request", error_description="Missing required code_challenge",
         )

tests/test_33_oauth2_pkce.py

@@ -4,7 +4,6 @@
 import secrets
 import string
 
-from oidcop.configure import ASConfiguration
 import pytest
 import yaml
 from oidcmsg.message import Message
@@ -15,6 +14,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
@@ -285,6 +285,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"