fix: x509 chain representation

Author: Zicchio

pyeudiw/trust/handler/interface.py

@@ -134,6 +134,12 @@ def validate_trust_material(
         """
         raise NotImplementedError
 
+    def is_it_me(self, client_id: str) -> bool:
+        """
+        TODO
+        """
+        return client_id == self.default_client_id
+
     @property
     def name(self) -> str:
         """
@@ -152,4 +158,7 @@ def default_client_id(self) -> str:
         :returns: The default client id of the trust handler
         :rtype: str
         """
+        # TODO: investiga dove questo viene veramente chiamat
+        # TODO: proposal to rename configured_client_id
+        # breakpoint()
         return self.client_id

pyeudiw/trust/handler/x509.py

@@ -1,6 +1,4 @@
-import base64
 import logging
-from ssl import PEM_cert_to_DER_cert
 from typing import Union
 
 from pyeudiw.trust.handler.interface import TrustHandlerInterface
@@ -9,6 +7,8 @@
 from pyeudiw.jwk.parse import parse_pem, 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, 
@@ -36,11 +36,13 @@ def __init__(
         private_keys: list[dict[str, str]],
         client_id_scheme: str = "x509_san_uri",
         certificate_authorities: dict[str, str] = [],
+        include_issued_jwt_header_param: bool = False,
         **kwargs
     ) -> None:        
         self.client_id = client_id
         self.client_id_scheme = client_id_scheme
         self.certificate_authorities = certificate_authorities
+        self.include_issued_jwt_header_param = include_issued_jwt_header_param
 
         if not relying_party_certificate_chains_by_ca:
             raise InvalidTrustHandlerConfiguration("No x509 certificate chains provided in the configuration")
@@ -70,7 +72,7 @@ def __init__(
                     break
 
             if not found_client_id:
-                logger.error(f"Invalid x509 leaf certificate using CA {k}. Unmatching client id ({client_id}); searched among {search_set}, the chain will be removed")
+                logger.error(f"Invalid x509 leaf certificate using CA {k}. Unmatching client id ({client_id}); the chain will be removed")
                 continue
 
             pem_type = get_certificate_type(v[0])
@@ -130,24 +132,30 @@ def extract_and_update_trust_materials(
         return trust_source
 
     def validate_trust_material(
-        self, 
-        x5c: list[str], 
+        self,
+        x5c: list[str],
         trust_source: TrustSourceData,
     ) -> tuple[bool, TrustSourceData]:
-        chain = [base64.b64decode(b64der) for b64der in x5c]
-
-        if len(chain) > 1 and not verify_x509_attestation_chain(chain):
+        # 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...
+        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
 
-        issuer = get_trust_anchor_from_x5c(chain)
+        issuer = get_trust_anchor_from_x5c(der_chain)
 
         if not issuer:
-            logger.error(f"Invalid x509 certificate chain. Issuer not found")
+            logger.error("Invalid x509 certificate chain. Issuer not found")
             return False, trust_source
 
         if not issuer in self.certificate_authorities:
-            logger.error(f"Invalid x509 certificate chain. Issuer not found in the list of trusted CAs")
+            logger.error("Invalid x509 certificate chain. Issuer not found in the list of trusted CAs")
             return False, trust_source
 
         issuer_pem = self.certificate_authorities[issuer]
@@ -156,19 +164,19 @@ def validate_trust_material(
             issuer_jwk = parse_pem(issuer_pem)
             chain_jwks = parse_x5c_keys(x5c)
         except Exception as e:
-            logger.error(f"Invalid x509 certificate chain. Parsing failed: {e}")
+            logger.error("Invalid x509 certificate chain. Parsing failed: {e}")
             return False, trust_source
 
         if not issuer_jwk.thumbprint == chain_jwks[-1].thumbprint:
-            logger.error(f"Invalid x509 certificate chain. Issuer thumbprint does not match")
+            logger.error("Invalid x509 certificate chain. Issuer thumbprint does not match")
             return False, trust_source
 
         trust_source.add_trust_param(
             "x509",
             TrustEvaluationType(
                 attribute_name=self.get_handled_trust_material_name(),
-                x5c=x5c,
-                expiration_date=get_expiry_date_from_x5c(chain),
+                x5c=pem_chain,
+                expiration_date=get_expiry_date_from_x5c(der_chain),
                 jwks=chain_jwks,
                 trust_handler_name=self.name,
             )
@@ -179,7 +187,7 @@ def validate_trust_material(
     def extract_jwt_header_trust_parameters(self, trust_source: TrustSourceData) -> dict:
         tp: dict = trust_source.serialize().get(X509Handler._TRUST_TYPE, {})
         if (x5c_pem := tp.get(X509Handler._TRUST_PARAMETER_NAME, None)):
-            x5c = [base64.b64encode(PEM_cert_to_DER_cert(pem)).decode() for pem in x5c_pem]
+            x5c = [PEM_cert_to_B64DER_cert(pem) for pem in x5c_pem]
             return {"x5c": x5c}
         return {}
 

pyeudiw/x509/verify.py

@@ -1,5 +1,7 @@
+import base64
 import logging
 from datetime import datetime
+import re
 from ssl import DER_cert_to_PEM_cert, PEM_cert_to_DER_cert
 
 import pem
@@ -16,6 +18,8 @@
 
 logger = logging.getLogger(__name__)
 
+_BASE64_RE = re.compile("^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$")
+
 
 def _verify_x509_certificate_chain(pems: list[str]):
     """
@@ -90,6 +94,64 @@ def verify_x509_attestation_chain(x5c: list[bytes]) -> bool:
 
     return _verify_x509_certificate_chain(pems)
 
+
+def PEM_cert_to_B64DER_cert(cert: str) -> str:
+    """
+    Takes a certificate in ANSII PEM format and returns the base64
+    encoding of the corresponding DER certificate.
+    """
+    return base64.b64encode(PEM_cert_to_DER_cert(cert)).decode()
+
+
+def B64DER_cert_to_PEM_cert(cert: str) -> str:
+    """
+    Takes a certificate Base64 encoded DER and returns the
+    certificate in ANSII PEM format.
+    """
+    return DER_cert_to_PEM_cert(base64.b64decode(cert))
+
+
+def to_DER_cert(cert: str | bytes) -> bytes:
+    """
+    This function takes in a certificate with unknown representation
+    (allegedly, PEM, DER or Base64 encoded DER) and applies some
+    heuristics to convert it to a DER certificate.
+
+    This function should be treated as UNSAFE and inefficient. Do NOT
+    use it unless you do NOT hany prior way to know the actual representation
+    format of a certificate
+    """
+    cert_s = ""
+    if isinstance(cert, bytes):
+        if is_der_format(cert):
+            return cert
+        cert_s = cert.decode()
+    else:
+        cert_s = cert
+
+    if cert_s.startswith("-----BEGIN CERTIFICATE-----"):
+        return PEM_cert_to_DER_cert(cert_s)
+
+    cert_s = cert_s.replace('\n\r', '')
+    if _BASE64_RE.fullmatch(cert_s):
+        return B64DER_cert_to_PEM_cert(cert_s)
+
+    raise ValueError("unable to recognize input [cert] as a ccertifficate")
+
+
+def to_PEM_cert(cer: str | bytes) -> str:
+    """
+    This function takes in a certificate with unknown representation
+    (allegedly, PEM, DER or Base64 encoded DER) and applies some
+    heuristics to convert it to a PEM certificate.
+
+    This function should be treated as UNSAFE and inefficient. Do NOT
+    use it unless you do NOT hany prior way to know the actual representation
+    format of a certificate
+    """
+    raise NotImplementedError("TODO")    
+
+
 def pem_to_pems_list(cert: str) -> list[str]:
     """
     Convert the x509 certificate chain from PEM to multiple PEMs.