Merge pull request #437 from italia/feat/sign_header_x5c

feat: refactor of jwt key selection for x5c

Author: peppelinux

pyeudiw/jwk/parse.py

@@ -1,4 +1,3 @@
-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
@@ -7,6 +6,8 @@
 from pyeudiw.jwk.exceptions import InvalidJwk
 from typing import Optional
 
+from pyeudiw.x509.verify import B64DER_cert_to_DER_cert
+
 def _parse_rsa_key(pem: str) -> Optional[JWK]:
     try:
         public_key = import_rsa_key(pem)
@@ -71,7 +72,7 @@ def parse_b64der(b64der: str) -> JWK:
     """
     Parse a (public) key from a Base64 encoded DER certificate.
     """
-    der = base64.b64decode(b64der)
+    der = B64DER_cert_to_DER_cert(b64der)
     return parse_certificate(der)
 
 

pyeudiw/jwt/helper.py

@@ -87,11 +87,11 @@ def find_self_contained_key(header: dict) -> tuple[set[str], JWK] | None:
         candidate_key: JWK | None = None
         try:
             candidate_key = parse_x5c_keys(header["x5c"])[0]
+            return set(["5xc"]), candidate_key
         except Exception as e:
             logger.debug(
                 f"failed to parse key from x5c chain {header['x5c']}", exc_info=e
             )
-        return set(["5xc"]), candidate_key
     if "jwk" in header:
         candidate_key = JWK(header["jwk"])
         return set(["jwk"]), candidate_key

pyeudiw/jwt/jws_helper.py

@@ -10,6 +10,7 @@
 from pyeudiw.jwk import JWK
 from pyeudiw.jwk.exceptions import KidError
 from pyeudiw.jwk.jwks import find_jwk_by_kid, find_jwk_by_thumbprint
+from pyeudiw.jwk.parse import parse_b64der
 from pyeudiw.jwt.exceptions import (
     JWEEncryptionError,
     JWSSigningError,
@@ -76,9 +77,12 @@ def sign(
         of available keys.
 
         If the header already contains indication of a key, such as 'kid',
-        'trust_chain' and 'x5c', there is no guarantee that the signing
-        key to be used will be aligned with those header. We assume that is
-        it responsibility of the class initiator to make those checks.
+        'trust_chain' and 'x5c', the method will attempt to match the signing
+        key among the available keys based on such claims, but there is no
+        guarantee that the correct key will be selected. We assume that is
+        it responsibility of the class initiator to make those checks. To
+        avoid any possible ambiguity, it is suggested to initilize the class
+        with one (signing) key only.
 
         :param plain_dict: The payload to be signed.
         :param protected: Protected header for the JWS.
@@ -106,13 +110,10 @@ def sign(
         if signing_key["kty"] == "oct":
             raise JWSSigningError(f"Key {signing_key['kid']} is a symmetric key")
 
-        # Ensure the key ID in the header matches the signing key
-        header_kid = protected.get("kid")
-        signer_kid = signing_key.get("kid")
-        if header_kid and signer_kid and (header_kid != signer_kid):
-            raise JWSSigningError(
-                f"token header contains a kid {header_kid} that does not match the signing key kid {signer_kid}"
-            )
+        try:
+            _validate_key_with_jws_header(signing_key, protected, unprotected)
+        except Exception as e:
+            raise JWSSigningError(f"failed to validate signing key: it's content it not valid for current header claims: {e}", e)
 
         payload = serialize_payload(plain_dict)
 
@@ -125,6 +126,8 @@ def sign(
             protected["typ"] = "sd-jwt" if self.is_sd_jwt(plain_dict) else "JWT"
 
         # Include the signing key's kid in the header if required
+        header_kid = protected.get("kid")
+        signer_kid = signing_key.get("kid")
         if kid_in_header and signer_kid:
             # note that is actually redundant as the underlying library auto-update the header with the kid
             protected["kid"] = signer_kid
@@ -177,9 +180,12 @@ def _select_signing_key(
         # Case 2: only one *singing* key
         if signing_key := self._select_key_by_use(use="sig"):
             return signing_key
-        # Case 3: match key by kid: this goes beyond what promised on the method definition
+        # Case 3: match key by kid
         if signing_key := self._select_key_by_kid(headers):
             return signing_key
+        # Case 4: match key by x5c
+        if signing_key := self._select_key_by_x5c(headers):
+            return signing_key
         raise JWSSigningError(
             "signing error: not possible to uniquely determine the signing key"
         )
@@ -199,7 +205,7 @@ def _select_key_by_use(self, use: str) -> dict | None:
             return candidate_signing_keys[0]
         return None
 
-    def _select_key_by_kid(self, headers: tuple[dict, dict]) -> dict | None:
+    def _select_key_by_kid(self, headers: tuple[dict[str, Any], dict[str, Any]]) -> dict | None:
         if not headers:
             return None
         if "kid" in headers[0]:
@@ -210,6 +216,19 @@ def _select_key_by_kid(self, headers: tuple[dict, dict]) -> dict | None:
             return None
         return find_jwk_by_kid([key.to_dict() for key in self.jwks], kid)
 
+    def _select_key_by_x5c(self, headers: tuple[dict[str, Any], dict[str, Any]]) -> dict | None:
+        if not headers:
+            return None
+        x5c: list[str] | None = headers[0].get("x5c") or headers[1].get("x5c")
+        if not x5c:
+            return None
+        header_jwk = parse_b64der(x5c[0])
+        for key in self.jwks:
+            key_d = key.to_dict()
+            if JWK(key_d).thumbprint == header_jwk.thumbprint:
+                return key_d
+        return None
+
     def verify(
         self, jwt: str, tolerance_s: int = DEFAULT_TOKEN_TIME_TOLERANCE
     ) -> str | Any | bytes:
@@ -320,3 +339,60 @@ def is_sd_jwt(self, token: str) -> bool:
             # Log or handle errors (optional)
             logger.warning(f"Unable to determine if token is SD-JWT: {e}")
             return False
+
+
+def _validate_key_with_header_kid(key: dict, header: dict) -> None:
+    """
+    :raises Exception: if the key is not compatible with the header content kid (if any)
+    """
+    if (key_kid := key.get("kid")) and (header_kid := header.get("kid")) and (key_kid != header_kid):
+        raise Exception(
+            f"token header contains a kid {header_kid} that does not match the signing key kid {key_kid}"
+        )
+    return
+
+
+def _validate_key_with_header_x5c(key: dict, header: dict) -> None:
+    """
+    Validate that a key has a public component that matches what defined in
+    the x5c leaf certificate in the header (if any).
+    Note that this method DOES NOT validate the chain. Instead, it actually
+    checks that the leaf of the chain has the same cryptographic material
+    of the argument key.
+
+    :raises Exception: if the key is not compatible with the header content x5c (if any)
+    """
+    x5c: list[str] | None = header.get("x5c")
+    if not x5c:
+        return
+    leaf_cert: str = x5c[0]
+
+    # if the key has a certificate, check the cert, otherwise check the public material    
+    key_x5c: list[str] | None = key.get("x5c")
+    if key_x5c:
+        if leaf_cert != (leaf_x5c_cert := key_x5c[0]):
+            raise Exception(
+                f"token header containes a chain whose leaf certificate {leaf_cert} does not match the signing key leaf certificate {leaf_x5c_cert}"\
+            )
+        return
+    header_key = parse_b64der(leaf_cert)
+    if header_key.thumbprint != JWK(key).thumbprint:
+        raise Exception(
+            f"public material of the key does not matches the key in the leaf certificate {leaf_cert}"
+        )
+    return
+
+
+def _validate_key_with_jws_header(key: dict, protected_jws_header: dict, unprotected_jws_header: dict) -> None:
+    """
+    Validate that a key used for some operations (sign, verify) on a token
+    is compatible with the token header itself.
+
+    :raises Exception: if the key is not compatible with the token header
+    """
+    header = deepcopy(protected_jws_header)
+    header.update(unprotected_jws_header)
+    # NOTE: consistency with usage claims such as 'alg', 'kty' and 'use'
+    # are done by the signer library and are not required here
+    _validate_key_with_header_kid(key, header)
+    _validate_key_with_header_x5c(key, header)

pyeudiw/openid4vp/presentation_submission/__init__.py

@@ -137,7 +137,6 @@ def validate(
 
         for descriptor in validated_submission.descriptor_map:
             handler = self.handlers.get(descriptor.format)
-
             if not handler:
                 raise MissingHandler(f"Handler for format '{descriptor.format}' not found.")
 

pyeudiw/satosa/default/request_handler.py

@@ -75,27 +75,11 @@ def request_endpoint(self, context: Context, *args) -> Response:
         trust_params = self.trust_evaluator.get_jwt_header_trust_parameters(issuer=self.client_id)
         _protected_jwt_headers.update(trust_params)
 
-        metadata_key = None
-
-        if "x5c" in _protected_jwt_headers:
-            # 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:
-                    metadata_key = key
-                    break
-            
-            if not metadata_key:
-                return self._handle_500(
-                    context,
-                    "internal error: unable to find the key in the metadata",
-                    ValueError("unable to find the key in the metadata"),
-                )
+        if ("x5c" in _protected_jwt_headers) or ("kid" in _protected_jwt_headers):
+            # let helper decide which key best fit the given header, otherise use default hich is the first confgiured key
+            helper = JWSHelper(self.config["metadata_jwks"])
         else:
-            metadata_key = self.default_metadata_private_jwk
-
-        helper = JWSHelper(metadata_key)
+            helper = JWSHelper(self.default_metadata_private_jwk)
 
         try:
             request_object_jwt = helper.sign(

pyeudiw/tests/jwt/test_helper.py

@@ -1,5 +1,12 @@
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptojwt.jwk.ec import ECKey
+
+from pyeudiw.jwk import JWK
 from pyeudiw.jwt.helper import validate_jwt_timestamps_claims
+from pyeudiw.jwt.jws_helper import _validate_key_with_jws_header
 from pyeudiw.tools.utils import iat_now
+import pyeudiw.tests.x509.test_x509 as test_x509
+from pyeudiw.x509.verify import DER_cert_to_B64DER_cert
 
 
 def test_validate_jwt_timestamps_claims_ok():
@@ -71,3 +78,63 @@ def test_test_validate_jwt_timestamps_claims_tolerance_window():
         assert (
             False
         ), f"encountered unexpeted error when validating the lifetime of a token payload with a tolerance window (for exp): {e}"
+
+
+def test_validate_key_with_jws_header_x5c_ok():
+    private_ec_key = ec.generate_private_key(ec.SECP256R1())
+    x509_der_chain = test_x509.gen_chain(leaf_private_key=private_ec_key)
+    x5c = [DER_cert_to_B64DER_cert(der) for der in x509_der_chain]
+    
+    ec_jwk = ECKey()
+    ec_jwk.load_key(private_ec_key)
+    key = ec_jwk.serialize(private=True)
+
+    try:
+        _validate_key_with_jws_header(key, {"x5c": x5c}, {})
+        assert True
+    except Exception as e:
+        assert False, f"unexpected exception when validating header for correct key: {e}"
+
+
+def test_validate_key_with_jws_header_kid_ok():
+    key = JWK().as_dict()
+    kid = "1234567890"
+    key["kid"] = kid
+
+    try:
+        _validate_key_with_jws_header(key, {"kid": kid}, {})
+        assert True
+    except Exception as e:
+        assert False, f"unexpected exception when validating header for correct key: {e}"
+
+
+def test_validate_key_with_jws_header_expect_x5c_fail():
+    private_ec_key = ec.generate_private_key(ec.SECP256R1())
+    x509_der_chain = test_x509.gen_chain(leaf_private_key=private_ec_key)
+    x5c = [DER_cert_to_B64DER_cert(der) for der in x509_der_chain]
+    
+    wrong_ec_key = ec.generate_private_key(ec.SECP256R1())
+    wrong_ec_jwk = ECKey()
+    wrong_ec_jwk.load_key(wrong_ec_key)
+    wrong_key = wrong_ec_jwk.serialize(private=True)
+
+    try:
+        _validate_key_with_jws_header(wrong_key, {"x5c": x5c}, {})
+        assert False, f"should have encountered exception when validating header 'x5c' for wrong key"
+    except Exception as _:
+        assert True
+
+def test_validate_key_with_jws_header_expect_kid_fail():
+    wrong_key = JWK().as_dict()
+    wrong_kid = "1234567890"
+    wrong_key["kid"] = wrong_kid
+    
+    key = JWK().as_dict()
+    kid = "qwertyuiop"
+    key["kid"] = kid
+
+    try:
+        _validate_key_with_jws_header(key, {"kid": "1234567890"}, {})
+        assert False, f"should have encountered exception when validating header 'kid' for wrong key"
+    except Exception as _:
+        assert True

pyeudiw/tests/jwt/test_sign_verify.py

@@ -1,8 +1,12 @@
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptojwt.jwk.ec import ECKey
 import pytest
 
 from pyeudiw.jwt.jws_helper import DEFAULT_TOKEN_TIME_TOLERANCE, JWSHelper
 from pyeudiw.jwt.utils import decode_jwt_header
+import pyeudiw.tests.x509.test_x509 as test_x509
 from pyeudiw.tools.utils import iat_now
+from pyeudiw.x509.verify import DER_cert_to_B64DER_cert
 
 
 class TestJWSHeperSelectSigningKey:
@@ -49,6 +53,19 @@ def test_JWSHelper_select_signing_key_infer_kid(self, sign_jwks):
         k = signer._select_signing_key(({"kid": exp_k["kid"]}, {}))
         assert k == exp_k
 
+    def test_JWSHelper_select_signing_key_infer_kid(self, sign_jwks: list[dict]):
+        new_private_ec_key = ec.generate_private_key(ec.SECP256R1())
+        x509_der_chain = test_x509.gen_chain(leaf_private_key=new_private_ec_key)
+        x5c = [DER_cert_to_B64DER_cert(der) for der in x509_der_chain]
+        new_ec_jwk = ECKey()
+        new_ec_jwk.load_key(new_private_ec_key)
+        exp_key: dict = new_ec_jwk.serialize(private=True)
+        sign_jwks.append(exp_key)
+
+        signer = JWSHelper(sign_jwks)
+        obt_key = signer._select_signing_key(({"x5c": x5c}, {}))
+        assert exp_key == obt_key
+
     def test_JWSHelper_select_signing_key_unique(self, sign_jwks):
         signer = JWSHelper(sign_jwks[0])
         exp_k = sign_jwks[0]

pyeudiw/tests/satosa/test_backend.py

@@ -54,7 +54,7 @@
 from pyeudiw.jwt.jwe_helper import JWEHelper
 from pyeudiw.satosa.utils.response import JsonResponse
 from pyeudiw.tests.x509.test_x509 import gen_chain
-from pyeudiw.x509.verify import to_pem_list
+from pyeudiw.x509.verify import PEM_cert_to_B64DER_cert, to_pem_list
 from pyeudiw.jwk.parse import parse_pem
 
 PKEY = {
@@ -192,7 +192,7 @@ def _generate_payload(self, issuer_jwk, holder_jwk, nonce, state, aud, x509=Fals
 
         if x509:
             additional_headers = {
-                "x5c": self.chain
+                "x5c": [PEM_cert_to_B64DER_cert(pem) for pem in self.chain]
             }
         else:
             additional_headers = {
@@ -622,7 +622,6 @@ def test_response_endpoint_x5c_chain(self, context):
         }
         context.request_method = "POST"
         context.http_headers = {"HTTP_CONTENT_TYPE": "application/x-www-form-urlencoded"}
-        
         response_endpoint = self.backend.response_endpoint(context)
         assert response_endpoint.status == "200"
         assert "redirect_uri" in response_endpoint.message