Different endpoint accept different audience in a JWT used for client authentication. New specifications like the pushed authorization adds complexity to this. So a refactoring was needed.

Author: rohe

src/oidcendpoint/client_authn.py

@@ -150,7 +150,8 @@ def verify(self, request, **kwargs):
             else:
                 raise NotForMe("Not for me!")
         else:
-            if self.endpoint_context.endpoint[_endpoint].full_path in ca_jwt["aud"]:
+            if set(ca_jwt["aud"]).intersection(
+                    self.endpoint_context.endpoint[_endpoint].allowed_target_uris()):
                 pass
             else:
                 raise NotForMe("Not for me!")

src/oidcendpoint/endpoint.py

@@ -52,8 +52,8 @@
 "response" MUST be present
 "http_headers" MAY be present
 "cookie": MAY be present 
-"response_placement": If absent defaults the endpoints response_placement parameter value
-    or if that is also missing 'url'
+"response_placement": If absent defaults to the endpoints response_placement 
+    parameter value or if that is also missing 'url'
 """
 
 
@@ -187,6 +187,9 @@ def __init__(self, endpoint_context, **kwargs):
         self.endpoint_info = construct_endpoint_info(
             self.default_capabilities, **kwargs
         )
+        # This is for matching against aud in JWTs
+        # By default the endpoint's endpoint URL is an allowed target
+        self.allowed_targets = [self.name]
 
     def parse_request(self, request, auth=None, **kwargs):
         """
@@ -209,7 +212,7 @@ def parse_request(self, request, auth=None, **kwargs):
                         request,
                         "jwt",
                         keyjar=self.endpoint_context.keyjar,
-                        verify=self.endpoint_context.verify_ssl,
+                        verify=self.endpoint_context.httpc_params["verify"],
                         **kwargs
                     )
                 elif self.request_format == "url":
@@ -425,3 +428,12 @@ def do_response(self, response_args=None, request=None, error="", **kwargs):
             pass
 
         return _resp
+
+    def allowed_target_uris(self):
+        res = []
+        for t in self.allowed_targets:
+            if t == "":
+                res.append(self.endpoint_context.issuer)
+            else:
+                res.append(self.endpoint_context.endpoint[t].full_path)
+        return set(res)

src/oidcendpoint/oidc/userinfo.py

@@ -36,6 +36,8 @@ def __init__(self, endpoint_context, **kwargs):
         self.scope_to_claims = None
         if "client_authn_method" not in kwargs:
             self.client_authn_method = self.default_capabilities["client_authn_method"]
+        # Add the issuer ID as an allowed JWT target
+        self.allowed_targets.append("")
 
     def get_client_id_from_token(self, endpoint_context, token, request=None):
         sinfo = self.endpoint_context.sdb[token]

tests/test_02_client_authn.py

@@ -22,6 +22,7 @@
 from oidcendpoint.exception import NotForMe
 from oidcendpoint.oidc.authorization import Authorization
 from oidcendpoint.oidc.token import AccessToken
+from oidcendpoint.oidc.userinfo import UserInfo
 
 KEYDEFS = [
     {"type": "RSA", "key": "", "use": ["sig"]},
@@ -39,7 +40,8 @@
     "verify_ssl": False,
     "endpoint": {
         "token": {"path": "token", "class": AccessToken, "kwargs": {}},
-        "authorization": {"path": "auth", "class": Authorization, "kwargs": {}}
+        "authorization": {"path": "auth", "class": Authorization, "kwargs": {}},
+        "userinfo": {"path": "user", "class": UserInfo, "kwargs": {}}
     },
     "template_dir": "template",
     "jwks": {
@@ -138,7 +140,7 @@ def test_private_key_jwt_reusage_other_endpoint():
     request = {"client_assertion": _assertion, "client_assertion_type": JWT_BEARER}
 
     # This should be OK
-    authn_info = PrivateKeyJWT(endpoint_context).verify(request, endpoint="token")
+    PrivateKeyJWT(endpoint_context).verify(request, endpoint="token")
 
     # This should NOT be OK
     with pytest.raises(NotForMe):
@@ -375,3 +377,19 @@ def test_verify_client_bearer_header():
     res = verify_client(endpoint_context, request, token, get_client_id_from_token)
     assert set(res.keys()) == {"token", "method", "client_id"}
     assert res["method"] == "bearer_header"
+
+
+def test_jws_authn_method_aud_userinfo_endpoint():
+    client_keyjar = KeyJar()
+    client_keyjar[conf["issuer"]] = KEYJAR.issuer_keys[""]
+    # The only own key the client has a this point
+    client_keyjar.add_symmetric("", client_secret, ["sig"])
+
+    _jwt = JWT(client_keyjar, iss=client_id, sign_alg="HS256")
+
+    # audience is the OP - not specifically the user info endpoint
+    _assertion = _jwt.pack({"aud": [conf["issuer"]]})
+
+    request = {"client_assertion": _assertion, "client_assertion_type": JWT_BEARER}
+
+    assert JWSAuthnMethod(endpoint_context).verify(request, endpoint="userinfo")