Merge pull request #436 from italia/fix/invalid_x5c_store

peppelinux · web-flow · commit 035224ad2dfd · 2025-04-29T16:31:20.000+02:00
Invalid x5c store and status list fix
diff --git a/pyeudiw/jwk/parse.py b/pyeudiw/jwk/parse.py
@@ -59,11 +59,13 @@ def parse_certificate(cert: str | bytes) -> JWK:
     :rtype: JWK
     """
 
-    if type(cert) == bytes or type(cert) == str and not cert.startswith("-----BEGIN CERTIFICATE-----"):
-        cert = DER_cert_to_PEM_cert(cert)
-
-    return parse_pem(cert)
-
+    try:
+        return parse_pem(cert)
+    except Exception:
+        try:
+            return parse_pem(DER_cert_to_PEM_cert(cert))
+        except Exception:        
+            raise InvalidJwk(f"unable to parse key from pem: {cert}")
 
 def parse_b64der(b64der: str) -> JWK:
     """
@@ -88,4 +90,10 @@ def parse_x5c_keys(x5c: list[str]) -> list[JWK]:
     :rtype: JWK
     """
 
-    return [parse_pem(pem) for pem in x5c]
+    try:
+        return [parse_certificate(cert) for cert in x5c]
+    except Exception:
+        try:
+            return [parse_certificate(DER_cert_to_PEM_cert(cert)) for cert in x5c]
+        except Exception:
+            raise InvalidJwk(f"unable to parse key from pem chain: {x5c}")
diff --git a/pyeudiw/openid4vp/vp_sd_jwt_vc.py b/pyeudiw/openid4vp/vp_sd_jwt_vc.py
@@ -89,7 +89,7 @@ def validate(
         
         payload = decode_jwt_payload(token)
 
-        if "status" in payload:
+        if "status" in payload and "status_list" in payload["status"]:
             status_list = StatusListTokenHelper.from_status(payload["status"])
             if status_list.is_expired() or \
                status_list.get_status(payload["status"]["status_list"]["idx"]) > 0:
diff --git a/pyeudiw/tests/jwk/test_parse.py b/pyeudiw/tests/jwk/test_parse.py
@@ -54,4 +54,4 @@ def test_parse_x5c_keys():
         assert False
     except Exception as e:
         assert isinstance(e, InvalidJwk)
-        assert str(e) == "unable to parse key from pem: invalid_x5c"
+        assert str(e) == "unable to parse key from pem chain: ['invalid_x5c']"
diff --git a/pyeudiw/tests/satosa/test_backend.py b/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 der_list_to_pem_list
+from pyeudiw.x509.verify import to_pem_list
 from pyeudiw.jwk.parse import parse_pem
 
 PKEY = {
@@ -113,7 +113,7 @@ class TestOpenID4VPBackend:
     def create_backend(self):
         db_engine_inst = DBEngine(CONFIG["storage"])
         
-        self.chain = der_list_to_pem_list(DEFAULT_X509_CHAIN)
+        self.chain = to_pem_list(DEFAULT_X509_CHAIN)
         issuer_pem = self.chain[-1]
         self.x509_leaf_private_key = DEFAULT_X509_LEAF_JWK
 
diff --git a/pyeudiw/trust/handler/x509.py b/pyeudiw/trust/handler/x509.py
@@ -4,15 +4,15 @@
 from pyeudiw.trust.handler.interface import TrustHandlerInterface
 from pyeudiw.trust.model.trust_source import TrustSourceData, TrustEvaluationType
 from pyeudiw.trust.handler.exceptions import InvalidTrustHandlerConfiguration
-from pyeudiw.jwk.parse import parse_pem, parse_x5c_keys, parse_certificate
+from pyeudiw.jwk.parse import parse_x5c_keys, parse_certificate
 from cryptojwt.jwk.jwk import key_from_jwk_dict
 from pyeudiw.x509.verify import (
     PEM_cert_to_B64DER_cert,
     to_DER_cert,
     verify_x509_attestation_chain, 
     get_expiry_date_from_x5c, 
-    der_list_to_pem_list, 
-    pem_list_to_der_list, 
+    to_pem_list, 
+    to_der_list, 
     get_x509_info,
     get_trust_anchor_from_x5c,
     get_certificate_type
@@ -94,7 +94,7 @@ def __init__(
                 logger.error(f"Invalid x509 leaf certificate using CA {k}. Unmatching private key, the chain will be removed")
                 continue
 
-            chain = pem_list_to_der_list(v) if type(v[0]) == str and v[0].startswith("-----BEGIN CERTIFICATE-----") else v
+            chain = to_der_list(v)
 
             if verify_x509_attestation_chain(chain):
                 self.relying_party_certificate_chains_by_ca[k] = chain
@@ -104,79 +104,87 @@ def __init__(
 
         self.private_keys = private_keys
 
-    def extract_and_update_trust_materials(
-        self, issuer: str, trust_source: TrustSourceData
-    ) -> TrustSourceData:
-        # Return the first valid chain
-        for ca, chain in self.relying_party_certificate_chains_by_ca.items():
-            if not verify_x509_attestation_chain(chain):
-                logger.error(f"Invalid x509 certificate chain using CA {ca}. Chain validation failed, the chain will be removed")
-                del self.relying_party_certificate_chains_by_ca[ca]
-                continue
-            
-            exp = get_expiry_date_from_x5c(chain)
-
-            trust_source.add_trust_param(
-                X509Handler._TRUST_TYPE,
-                TrustEvaluationType(
-                    attribute_name="x5c",
-                    x5c=der_list_to_pem_list(chain),
-                    expiration_date=exp,
-                    jwks=self.private_keys,
-                    trust_handler_name=self.name,
-                )
-            )
-
-            return trust_source
-        
-        return trust_source
-    
-    def validate_trust_material(
-        self,
-        x5c: list[str],
-        trust_source: TrustSourceData,
-    ) -> tuple[bool, TrustSourceData]:
-        # TODO: qui c'è del lavoro veramente sporco da fare.
-        #  Bisogna
-        #  (1) normalizzare la rappresentazione della chain a DER; per fare questo bisogna fare inferenza se PEM o Base64+DER
-        #  (2) normalizzare il salvatagggio della chain a PEM
-        #  (3) incrociare le dita che MDOC non si sfasci...
+    def _verify_chain(self, x5c: list[str]) -> bool:
+        """
+        Verify the x5c chain.
+        :param x5c: The x5c chain to verify.
+        :return: True if the chain is valid, False otherwise.
+        """
         der_chain = [to_DER_cert(cert) for cert in x5c]
-        pem_chain = der_list_to_pem_list(der_chain)
 
         if len(der_chain) > 1 and not verify_x509_attestation_chain(der_chain):
             logger.error(f"Invalid x509 certificate chain. Chain validation failed")
-            return False, trust_source
+            return False
 
         issuer = get_trust_anchor_from_x5c(der_chain)
 
         if not issuer:
             logger.error("Invalid x509 certificate chain. Issuer not found")
-            return False, trust_source
+            return False
         
         if not issuer in self.certificate_authorities:
             logger.error("Invalid x509 certificate chain. Issuer not found in the list of trusted CAs")
-            return False, trust_source
+            return False
         
-        issuer_pem = self.certificate_authorities[issuer]
+        issuer_cert = self.certificate_authorities[issuer]
 
         try:
-            issuer_jwk = parse_pem(issuer_pem)
-            chain_jwks = parse_x5c_keys(x5c)
+            issuer_jwk = parse_certificate(issuer_cert)
+            chain_jwks = parse_x5c_keys(der_chain)
         except Exception as e:
-            logger.error("Invalid x509 certificate chain. Parsing failed: {e}")
-            return False, trust_source
+            logger.error(f"Invalid x509 certificate chain. Parsing failed: {e}")
+            return False
 
         if not issuer_jwk.thumbprint == chain_jwks[-1].thumbprint:
             logger.error("Invalid x509 certificate chain. Issuer thumbprint does not match")
+            return False
+        
+        return True
+
+    def extract_and_update_trust_materials(
+        self, issuer: str, trust_source: TrustSourceData
+    ) -> TrustSourceData:
+        # Return the first valid chain
+        if issuer == self.client_id:
+            for ca, chain in self.relying_party_certificate_chains_by_ca.items():
+                if not self._verify_chain(chain):
+                    logger.error(f"Invalid x509 certificate chain using CA {ca}. Chain will be ignored")
+                    continue
+                
+                exp = get_expiry_date_from_x5c(chain)
+
+                trust_source.add_trust_param(
+                    X509Handler._TRUST_TYPE,
+                    TrustEvaluationType(
+                        attribute_name="x5c",
+                        x5c=to_pem_list(chain),
+                        expiration_date=exp,
+                        jwks=self.private_keys,
+                        trust_handler_name=self.name,
+                    )
+                )
+
+                return trust_source
+            
+        return trust_source
+    
+    def validate_trust_material(
+        self,
+        x5c: list[str],
+        trust_source: TrustSourceData,
+    ) -> tuple[bool, TrustSourceData]:
+        chain_jwks = parse_x5c_keys(x5c)
+        valid = self._verify_chain(x5c)
+
+        if not valid:
             return False, trust_source
         
         trust_source.add_trust_param(
             "x509",
             TrustEvaluationType(
                 attribute_name=self.get_handled_trust_material_name(),
-                x5c=pem_chain,
-                expiration_date=get_expiry_date_from_x5c(der_chain),
+                x5c=to_pem_list(x5c),
+                expiration_date=get_expiry_date_from_x5c(x5c),
                 jwks=chain_jwks,
                 trust_handler_name=self.name,
             )
diff --git a/pyeudiw/x509/verify.py b/pyeudiw/x509/verify.py
@@ -90,7 +90,7 @@ def verify_x509_attestation_chain(x5c: list[bytes]) -> bool:
     if not _check_datetime(exp):
         return False
     
-    pems = [DER_cert_to_PEM_cert(cert) for cert in x5c]
+    pems = [to_PEM_cert(cert) for cert in x5c]
 
     return _verify_x509_certificate_chain(pems)
 
@@ -149,8 +149,22 @@ def to_PEM_cert(cer: str | bytes) -> str:
     use it unless you do NOT hany prior way to know the actual representation
     format of a certificate
     """
-    raise NotImplementedError("TODO")    
+    cert_s = b""
 
+    if isinstance(cer, str):
+        if is_pem_format(cer):
+            return cer
+        cert_s = cer.encode()
+    else:
+        cert_s = cer
+
+    if cert_s.startswith(b"-----BEGIN CERTIFICATE-----"):
+        return str(cert_s)
+
+    if _BASE64_RE.fullmatch(str(cert_s)):
+        return B64DER_cert_to_PEM_cert(cert_s)
+    else:
+        return DER_cert_to_PEM_cert(cert_s)
 
 def pem_to_pems_list(cert: str) -> list[str]:
     """
@@ -164,29 +178,31 @@ def pem_to_pems_list(cert: str) -> list[str]:
     """
     return [str(cert) for cert in pem.parse(cert)]
 
-def der_list_to_pem_list(der_list: list[bytes]) -> list[str]:
+def to_pem_list(der_list: list[bytes] | list[str]) -> list[str]:
     """
-    Convert the x509 certificate chain from DER to PEM.
+    If the input is a list of DER certificates, it will be converted to a list of PEM certificates.
+    If the input is a list of PEM certificates, it will be returned as is.
 
     :param der: The x509 certificate chain in DER format
     :type der: list[bytes]
 
     :returns: The x509 certificate chain in PEM format
     :rtype: list[str]
     """
-    return [DER_cert_to_PEM_cert(cert) for cert in der_list]
+    return [to_PEM_cert(cert) for cert in der_list]
 
-def pem_list_to_der_list(pem_list: list[str]) -> list[bytes]:
+def to_der_list(pem_list: list[str] | list[bytes]) -> list[bytes]:
     """
-    Convert the x509 certificate chain from PEM to DER.
+    If the input is a list of PEM certificates, it will be converted to a list of DER certificates.
+    If the input is a list of DER certificates, it will be returned as is.
 
     :param pem_list: The x509 certificate chain in PEM format
     :type pem_list: list[str]
 
     :returns: The x509 certificate chain in DER format
     :rtype: list[bytes]
     """
-    return [PEM_cert_to_DER_cert(cert) for cert in pem_list]
+    return [to_DER_cert(cert) for cert in pem_list]
 
 def get_expiry_date_from_x5c(x5c: list[bytes]) -> datetime:
     """
@@ -211,7 +227,7 @@ def verify_x509_anchor(pem_str: str) -> bool:
     :returns: True if the x509 anchor certificate is valid else False
     :rtype: bool
     """
-    cert_data = load_der_x509_certificate(PEM_cert_to_DER_cert(pem_str))
+    cert_data = load_der_x509_certificate(to_DER_cert(pem_str))
 
     if not _check_datetime(cert_data.not_valid_after):
         logging.error(LOG_ERROR.format("check datetime failed"))
@@ -258,7 +274,7 @@ def get_issuer_from_x5c(x5c: list[bytes] | list[str]) -> Optional[str]:
     :returns: The issuer
     :rtype: str
     """
-    der = x5c[0] if isinstance(x5c[0], bytes) else PEM_cert_to_DER_cert(x5c[0])
+    der = to_DER_cert(x5c[0])
     return get_get_subject_name(der)
     
 
@@ -272,7 +288,7 @@ def get_trust_anchor_from_x5c(x5c: list[bytes] | list[str]) -> Optional[str]:
     :returns: The issuer
     :rtype: str
     """
-    der = x5c[-1] if isinstance(x5c[-1], bytes) else PEM_cert_to_DER_cert(x5c[-1])
+    der = to_DER_cert(x5c[-1])
     return get_get_subject_name(der)
 
 def get_expiry_date_from_x5c(x5c: list[bytes] | list[str]) -> datetime:
@@ -285,7 +301,7 @@ def get_expiry_date_from_x5c(x5c: list[bytes] | list[str]) -> datetime:
     :returns: The expiry date
     :rtype: datetime
     """
-    der = x5c[0] if isinstance(x5c[0], bytes) else PEM_cert_to_DER_cert(x5c[0])
+    der = to_DER_cert(x5c[0])
     cert = load_der_x509_certificate(der)
     return cert.not_valid_after
 
@@ -303,7 +319,7 @@ def get_x509_info(cert: bytes | str, info_type: str = "x509_san_dns") -> str:
     """
     get_common_name = lambda cert: cert.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)[0].value
 
-    der = cert if isinstance(cert, bytes) else PEM_cert_to_DER_cert(cert)
+    der = to_DER_cert(cert)
     cert: x509.Certificate = load_der_x509_certificate(der, default_backend())
 
     try:
@@ -334,13 +350,29 @@ def is_der_format(cert: bytes) -> str:
     except crypto.Error as e:
         logging.error(LOG_ERROR.format(e))
         return False
+    
+def is_pem_format(cert: str) -> str:
+    """
+    Check if the certificate is in PEM format.
 
+    :param cert: The certificate
+    :type cert: bytes
+
+    :returns: True if the certificate is in PEM format else False
+    :rtype: bool
+    """
+    try:
+        crypto.load_certificate(crypto.FILETYPE_PEM, cert)
+        return True
+    except crypto.Error as e:
+        logging.error(LOG_ERROR.format(e))
+        return False
 
 def get_public_key_from_x509_chain(x5c: list[bytes]) -> ECKey | RSAKey | dict:
     raise NotImplementedError("TODO")
 
 def get_certificate_type(cert: str | bytes) -> str:
-    pem = cert if isinstance(cert, str) and cert.startswith("-----BEGIN CERTIFICATE-----") else DER_cert_to_PEM_cert(cert)
+    pem = to_PEM_cert(cert)
 
     cert = x509.load_pem_x509_certificate(pem.encode(), default_backend())
     public_key = cert.public_key()
