Merge pull request #428 from Zicchio/feat/jwt_trust_header_inclusion

trust header (including x5c) rework

Author: peppelinux

example/satosa/pyeudiw_backend.yaml

@@ -209,6 +209,7 @@ config:
         httpc_params: *httpc_params
         cache_ttl: 0
         entity_configuration_exp: 600
+        # include_issued_jwt_header_param: true # default false; if true, it will include trust_chain header parameters in the signed presentation request issued by this trust handler
         metadata_type: "openid_credential_verifier"
         metadata: *metadata
         authority_hints:
@@ -248,6 +249,7 @@ config:
       config:
         # client_id: *client_id
         client_id_scheme: x509_san_dns # this will be prepended in the client id scheme used in the request. 
+        include_issued_jwt_header_param: true # default false; if true, it will include x5c header parameters in the signed presentation request issued by this trust handler
         certificate_authorities:
           ca.example.com: |
             -----BEGIN CERTIFICATE-----

pyeudiw/jwk/parse.py

@@ -1,3 +1,4 @@
+import base64
 from cryptojwt.jwk.ec import import_ec_key, ECKey
 from cryptojwt.jwk.rsa import RSAKey, import_rsa_key
 from ssl import DER_cert_to_PEM_cert
@@ -63,6 +64,15 @@ def parse_certificate(cert: str | bytes) -> JWK:
 
     return parse_pem(cert)
 
+
+def parse_b64der(b64der: str) -> JWK:
+    """
+    Parse a (public) key from a Base64 encoded DER certificate.
+    """
+    der = base64.b64decode(b64der)
+    return parse_certificate(der)
+
+
 def parse_x5c_keys(x5c: list[str]) -> list[JWK]:
     """
     Parse a the keys from a x5c chain.

pyeudiw/satosa/default/request_handler.py

@@ -5,8 +5,7 @@
 from pyeudiw.satosa.interfaces.request_handler import RequestHandlerInterface
 from pyeudiw.satosa.utils.response import Response
 from pyeudiw.tools.base_logger import BaseLogger
-from pyeudiw.jwt.exceptions import JWSSigningError
-from pyeudiw.jwk.parse import parse_certificate
+from pyeudiw.jwk.parse import parse_b64der
 from pyeudiw.jwk import JWK
 
 
@@ -75,7 +74,8 @@ def request_endpoint(self, context: Context, *args) -> Response:
         metadata_key = None
 
         if "x5c" in _protected_jwt_headers:
-            jwk = parse_certificate(_protected_jwt_headers["x5c"][0])
+            # TODO: move this logic in the JWS signer...
+            jwk = parse_b64der(_protected_jwt_headers["x5c"][0])
 
             for key in self.config["metadata_jwks"]:
                 if JWK(key).thumbprint == jwk.thumbprint:
@@ -98,6 +98,7 @@ def request_endpoint(self, context: Context, *args) -> Response:
                 data,
                 protected=_protected_jwt_headers,
             )
+            self._log_debug(context, f"created request object {request_object_jwt}")
             return Response(
                 message=request_object_jwt,
                 status="200",

pyeudiw/tests/federation/base.py

@@ -29,17 +29,17 @@
 leaf_cred = {
     "exp": EXP,
     "iat": NOW,
-    "iss": "https://credential_issuer.example.org",
-    "sub": "https://credential_issuer.example.org",
+    "iss": "https://credential-issuer.example.org",
+    "sub": "https://credential-issuer.example.org",
     "jwks": {"keys": []},
     "metadata": {
         "openid_credential_issuer": {"jwks": {"keys": []}},
         "federation_entity": {
             "organization_name": "OpenID Credential Issuer example",
-            "homepage_uri": "https://credential_issuer.example.org/home",
-            "policy_uri": "https://credential_issuer.example.org/policy",
-            "logo_uri": "https://credential_issuer.example.org/static/logo.svg",
-            "contacts": ["tech@credential_issuer.example.org"],
+            "homepage_uri": "https://credential-issuer.example.org/home",
+            "policy_uri": "https://credential-issuer.example.org/policy",
+            "logo_uri": "https://credential-issuer.example.org/static/logo.svg",
+            "contacts": ["tech@credential-issuer.example.org"],
         },
     },
     "authority_hints": ["https://intermediate.eidas.example.org"],
@@ -55,7 +55,7 @@
     "exp": EXP,
     "iat": NOW,
     "iss": "https://intermediate.eidas.example.org",
-    "sub": "https://credential_issuer.example.org",
+    "sub": "https://credential-issuer.example.org",
     "jwks": {"keys": []},
 }
 intermediate_es_cred["jwks"]["keys"] = [leaf_cred_jwk.serialize()]

pyeudiw/tests/satosa/test_backend.py

@@ -467,7 +467,6 @@ def test_response_endpoint(self, context):
 
         # case (4): good aud, nonce and state
         good_response = self._generate_payload(self.issuer_jwk, self.holder_jwk, nonce, state, self.backend.client_id)
-
         encrypted_response = JWEHelper(
             CONFIG["metadata_jwks"][1]).encrypt(good_response)
         context.request = {
@@ -897,7 +896,7 @@ def test_request_endpoint(self, context):
         # msg = json.loads(state_endpoint_response.message)
         # assert msg["response"] == "Authentication successful"
 
-    def test_trust_patameters_in_response(self, context):
+    def test_trust_parameters_in_response(self, context):
         internal_data = InternalData()
         context.http_headers = dict(
             HTTP_USER_AGENT="Mozilla/5.0 (Linux; Android 10; SM-G960F) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.92 Mobile Safari/537.36"
@@ -913,12 +912,13 @@ def test_trust_patameters_in_response(self, context):
 
         tsd = TrustSourceData.empty(CREDENTIAL_ISSUER_ENTITY_ID)
         tsd.add_trust_param(
-            "trust_chain",
+            "federation",
             TrustEvaluationType(
                 attribute_name="trust_chain",
                 jwks=[JWK(key=ta_jwk).as_dict()],
                 expiration_date=datetime.datetime.now(),
                 trust_chain=trust_chain_wallet,
+                trust_handler_name="FederationHandler",
             )
         )
 

pyeudiw/tests/settings.py

@@ -268,6 +268,7 @@ def base64url_to_int(val):
                         ta_jwk.serialize(private=False),
                     ]
                 },
+                "include_issued_jwt_header_param": True,
                 "default_sig_alg": "RS256",
                 "federation_jwks": [
                     jwk,
@@ -310,6 +311,7 @@ def base64url_to_int(val):
             "class": "X509Handler",
             "config": {
                 "client_id": f"{BASE_URL}/OpenID4VP",
+                "include_issued_jwt_header_param": True,
                 "relying_party_certificate_chains_by_ca": {
                     "ca.example.com": DEFAULT_X509_CHAIN,
                 },

pyeudiw/tests/trust/handler/test_direct_trust.py

@@ -11,7 +11,7 @@
     issuer,
 )
 from pyeudiw.tests.trust.handler import issuer_jwk as expected_jwk
-from pyeudiw.trust.handler._direct_trust_jwk import build_jwk_issuer_endpoint
+from pyeudiw.trust.handler._direct_trust_jwk import build_jwk_issuer_endpoint, is_url
 from pyeudiw.trust.handler.direct_trust_sd_jwt_vc import (
     DirectTrustSdJwtVc,
     build_metadata_issuer_endpoint,
@@ -24,7 +24,7 @@
 def fake_get_http_url(
     urls: list[str] | str, httpc_params: dict, http_async: bool = True
 ) -> list[requests.Response]:
-    issuer = f"https://example_url.issuer.it/vct"
+    issuer = f"https://example-url.issuer.it/vct"
 
     if urls[0].endswith("vct"):
         response = Response()
@@ -112,7 +112,7 @@ def test_direct_trust_extract_jwks_from_jwk_metadata_invalid():
 def test_direct_trust_jwk():
     trust_handler = DirectTrustSdJwtVc()
 
-    random_issuer = f"{uuid.uuid4()}.issuer.it"
+    random_issuer = f"https://{uuid.uuid4()}.issuer.it"
 
     mocked_issuer_jwt_vc_issuer_endpoint = unittest.mock.patch(
         "pyeudiw.trust.handler._direct_trust_jwk.get_http_url",
@@ -143,7 +143,7 @@ def test_direct_trust_jwk():
 def test_direct_trust_jwk_not_conformat_url():
     trust_handler = DirectTrustSdJwtVc()
 
-    issuer = f"https://example_url.issuer.it/vct"
+    issuer = f"https://example-url.issuer.it/vct"
 
     mocked_issuer_jwt_vc_issuer_endpoint = unittest.mock.patch(
         "pyeudiw.trust.handler._direct_trust_jwk.get_http_url",
@@ -163,3 +163,12 @@ def test_direct_trust_jwk_not_conformat_url():
 
     assert len(obtained_jwks) == 1, f"expected 1 jwk, obtained {len(obtained_jwks)}"
     assert expected_jwk == obtained_jwks[0]
+
+
+def test_is_url():
+    assert is_url("missing-scheme.net") == False
+    assert is_url("http//malformed-scheme.net") == False
+    assert is_url("https://malformed_domain.org") == False
+    assert is_url("https://domain.example") == True
+    assert is_url("https://domain.example/path") == True
+    assert is_url("https://domain.example/path/trailing/") == True

pyeudiw/tests/trust/mock_trust_handler.py

@@ -32,6 +32,7 @@ class MockTrustHandler(TrustHandlerInterface):
     def __init__(self, *args, **kwargs):
         self.client_id = kwargs.get("default_client_id", None)
         self.exp = kwargs.get("exp", 10)
+        self.include_issued_jwt_header_param = kwargs.get("include_issued_jwt_header_param", False)
 
     def get_metadata(self, issuer: str, trust_source: TrustSourceData) -> dict:
         if issuer == self.client_id:
@@ -75,6 +76,9 @@ def extract_and_update_trust_materials(
         trust_source.add_trust_param("test_trust_param", trust_param)
 
         return trust_source
+    
+    def extract_jwt_header_trust_parameters(self, trust_source: TrustSourceData) -> dict:
+        return {'trust_param_name': trust_source.test_trust_param.trust_param_name}
 
 class UpdateTrustHandler(MockTrustHandler):
     """
@@ -108,6 +112,9 @@ def extract_and_update_trust_materials(
 
         return trust_source
 
+    def extract_jwt_header_trust_parameters(self, trust_source: TrustSourceData) -> dict:
+        return {'trust_param_name': trust_source.test_trust_param.trust_param_name}
+
 class NonConformatTrustHandler:
     def get_metadata(self, issuer: str, trust_source: TrustSourceData) -> dict:
         return trust_source

pyeudiw/tests/trust/test_dynamic.py

@@ -52,14 +52,13 @@ def test_public_key_and_metadata_retrive():
             "mock": {
                 "module": "pyeudiw.tests.trust.mock_trust_handler",
                 "class": "MockTrustHandler",
-                "config": {},
+                "config": {"include_issued_jwt_header_param": True},
             },
 
         }, db_engine, default_client_id="default-client-id", mode="update_first"
     )
 
     uuid_url = f"http://{uuid4()}.issuer.it"
-
     assert trust_ev.get_jwt_header_trust_parameters(uuid_url) == {'trust_param_name': {'trust_param_key': 'trust_param_value'}}
     metadata = trust_ev.get_metadata()
 
@@ -82,7 +81,7 @@ def test_update_first_strategy():
             "mock": {
                 "module": "pyeudiw.tests.trust.mock_trust_handler",
                 "class": "UpdateTrustHandler",
-                "config": {},
+                "config":  {"include_issued_jwt_header_param": True},
             },
 
         }, db_engine, default_client_id="default-client-id"
@@ -102,7 +101,7 @@ def test_cache_first_strategy():
             "mock": {
                 "module": "pyeudiw.tests.trust.mock_trust_handler",
                 "class": "UpdateTrustHandler",
-                "config": {},
+                "config":  {"include_issued_jwt_header_param": True},
             },
 
         }, db_engine, default_client_id="default-client-id", mode="cache_first"
@@ -122,7 +121,8 @@ def test_cache_first_strategy_expired():
                 "module": "pyeudiw.tests.trust.mock_trust_handler",
                 "class": "UpdateTrustHandler",
                 "config": {
-                    "exp": 0
+                    "exp": 0,
+                    "include_issued_jwt_header_param": True
                 },
             },
 
@@ -143,7 +143,7 @@ def test_cache_first_strategy_expired_revoked():
             "mock": {
                 "module": "pyeudiw.tests.trust.mock_trust_handler",
                 "class": "UpdateTrustHandler",
-                "config": {},
+                "config": {"include_issued_jwt_header_param": True},
             },
 
         }, db_engine, default_client_id="default-client-id", mode="cache_first"
@@ -166,7 +166,7 @@ def test_cache_first_strategy_expired_force_update():
             "mock": {
                 "module": "pyeudiw.tests.trust.mock_trust_handler",
                 "class": "UpdateTrustHandler",
-                "config": {},
+                "config":  {"include_issued_jwt_header_param": True},
             },
 
         }, db_engine, default_client_id="default-client-id", mode="cache_first"

pyeudiw/trust/dynamic.py

@@ -19,6 +19,8 @@
 
 UpsertMode = Union[Literal["update_first"], Literal["cache_first"]]
 
+INCLUDE_JWT_HEADER_CONFIG_NAME = "include_issued_jwt_header_param"
+
 
 class CombinedTrustEvaluator(BaseLogger):
     """
@@ -194,10 +196,7 @@ def get_public_keys(
                         else:
                             used_handlers.append(handler.__class__.__name__)
 
-            raise NoCriptographicMaterial(
-                f"no trust evaluator can provide cyptographic material "
-                f"for {issuer}: searched among: {self.handlers_names}"
-            )
+            self._log_warning("static trust evaluation", f"no configured trust handler can successfully process static trust material {static_trust_materials} of issuer {issuer}")
 
         # try with handlers that don't use static trust materials like DirectTrustJar
         filetered_handlers = [handler for handler in self.handlers if handler.__class__.__name__ not in used_handlers]
@@ -303,14 +302,11 @@ def get_jwt_header_trust_parameters(self, issuer: Optional[str] = None, force_up
         """
         trust_source = self._get_trust_source(issuer, force_update)
 
-        excluded_fields = ["entity_id", "policies", "metadata", "revoked"]
-
         headers_params = {}
-
-        for param_name, param_value in trust_source.serialize().items():
-            if param_name not in excluded_fields:
-                headers_params[param_value["attribute_name"]] = param_value[param_value["attribute_name"]]
-
+        for handler in self.handlers:
+            if getattr(handler, INCLUDE_JWT_HEADER_CONFIG_NAME, None):
+                if header := handler.extract_jwt_header_trust_parameters(trust_source):
+                    headers_params.update(header)
         return headers_params
 
     def build_metadata_endpoints(