Add cross signed root support

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog 
 
+## 1.5.0
+
+* New: discover cross-signed root variants from the issuing CA's AIA bundle and expose them on the chain.
+    * CLI: new `--include-cross-signs` flag appends discovered cross-signs to the bundle output. Cross-signs are also surfaced in `--info` regardless of the flag.
+    * API: `CertificateChain.cross_signs` lists the discovered variants; `Cert.public_key_fingerprint` exposes the SubjectPublicKeyInfo hash used for matching.
+    * Utils: new `load_ascii_to_x509_all`, `load_der_to_x509_all`, `load_bytes_to_x509_all` return every certificate from a PKCS7 bundle (existing `load_*_to_x509` keeps returning the first cert).
+
 ## 1.4.0
 
 * Extended support to python 3.13

README.md

@@ -20,6 +20,7 @@ The library provides an easy to use API to access each property of a certificate
 * PKCS7, PEM and DER formats
 * LetsEncrypt certificates
 * Including the root certificate using the system CA bundle or custom bundle
+* Discovering and including cross-signed variants of the root from the issuing CA's AIA bundle
 * Python2 (but not for much longer..)
 
 ## Installation
@@ -48,6 +49,20 @@ The bundle gets written to stdout and the chain information to stderr.
  3. <Cert common_name="DigiCert High Assurance EV Root CA" subject="CN=DigiCert High Assurance EV Root CA,OU=www.digicert.com,O=DigiCert Inc,C=US" issuer="CN=DigiCert High Assurance EV Root CA,OU=www.digicert.com,O=DigiCert Inc,C=US">
 ```
 
+### Including cross-signed root variants
+
+Some roots are cross-signed by older CAs so that clients which don't yet trust the modern root can still build a path to a trust anchor they do have. Pass `--include-cross-signs` to append every cross-signed variant discovered in the issuing CA's AIA bundle:
+
+```
+ $ cert_chain_resolver --include-cross-signs certificate.crt > compat-bundle.crt
+ 1. <Cert common_name="*.tinifycdn.com" subject="CN=*.tinifycdn.com" issuer="CN=Sectigo Public Server Authentication CA DV R36,...">
+ 2. <Cert common_name="Sectigo Public Server Authentication CA DV R36" subject="..." issuer="CN=Sectigo Public Server Authentication Root R46,...">
+ 3. <Cert common_name="Sectigo Public Server Authentication Root R46" subject="..." issuer="CN=USERTrust RSA Certification Authority,...">
+ 4. <Cert common_name="Sectigo Public Server Authentication Root R46" subject="..." issuer="CN=AAA Certificate Services,...">
+```
+
+The flag composes with `--include-root`: with both, the self-signed root is emitted first and the cross-signs follow. Cross-signs that share Subject + Subject Public Key Info with a cert already in the chain (but were issued by a different CA) are detected automatically; nothing happens when the AIA bundle contains no extras.
+
 ## Python API
 
 Make sure to read the [documentation](https://certificate-resolver.readthedocs.io/en/latest/api.html) for more examples and options.

cert_chain_resolver/__init__.py

@@ -1,4 +1,4 @@
 import sys
 
-__version__ = "1.4.1"
+__version__ = "1.5.0"
 __is_py3__ = sys.version_info >= (3, 0)

cert_chain_resolver/castore/base_store.py

@@ -1,7 +1,6 @@
 from cert_chain_resolver.exceptions import RootCertificateNotFound
 from cert_chain_resolver.models import Cert
 
-
 try:
     from typing import TYPE_CHECKING
 

cert_chain_resolver/castore/file_system.py

@@ -31,12 +31,9 @@
 class FileSystemStore(CAStore):
     """The :class:`SystemStore <SystemStore>` for finding the CA from file system bundle (PEM ONLY)."""
 
-    _cache = None  # type: dict[str, list[Cert]]
-    path = None  # type: str
-
     def __init__(self, path=None):
         # (None | str) -> None
-        self._cache = defaultdict(list)
+        self._cache = defaultdict(list)  # type: dict[str, list[Cert]]
         if not path:
             try:
                 path = next(p for p in eligible_paths if p and os.path.exists(p))

cert_chain_resolver/cli.py

@@ -13,45 +13,70 @@
     pass
 
 
-def _print_chain_details(chain, include_root):
+def _print_cert_details(cert, index, heading="Certificate"):
+    print("== {0} #{1} ==".format(heading, index))
+    print("Subject:".ljust(20) + cert.subject)
+    print("Issuer:".ljust(20) + cert.issuer)
+    print("NotBefore:".ljust(20) + cert.not_valid_before.isoformat())
+    print("NotAfter:".ljust(20) + cert.not_valid_after.isoformat())
+    print("Serial:".ljust(20) + str(cert.serial))
+    print("Sha256Fingeprint:".ljust(20) + str(cert.fingerprint))
+    print(
+        "CAIssuerLoc:".ljust(20) + cert.ca_issuer_access_location
+        if cert.ca_issuer_access_location
+        else ""
+    )
+    print("Is root:".ljust(20) + repr(cert.is_root))
+    print("Is CA:".ljust(20) + repr(cert.is_ca))
+    print("Domains:")
+    print("  Common name:".ljust(20) + cert.common_name)
+    if cert.subject_alternative_names:
+        print("  SANExtensions:")
+        for domain in cert.subject_alternative_names:
+            print("    " + domain)
+    print("")
+
+
+def _print_chain_details(chain, include_root, include_cross_signs):
     for index, cert in enumerate(chain, 1):
-        print("== Certificate #{0} ==".format(index))
-        print("Subject:".ljust(20) + cert.subject)
-        print("Issuer:".ljust(20) + cert.issuer)
-        print("NotBefore:".ljust(20) + cert.not_valid_before.isoformat())
-        print("NotAfter:".ljust(20) + cert.not_valid_after.isoformat())
-        print("Serial:".ljust(20) + str(cert.serial))
-        print("Sha256Fingeprint:".ljust(20) + str(cert.fingerprint))
-        print(
-            "CAIssuerLoc:".ljust(20) + cert.ca_issuer_access_location
-            if cert.ca_issuer_access_location
-            else ""
-        )
-        print("Is root:".ljust(20) + repr(cert.is_root))
-        print("Is CA:".ljust(20) + repr(cert.is_ca))
-        print("Domains:")
-        print("  Common name:".ljust(20) + cert.common_name)
-        if cert.subject_alternative_names:
-            print("  SANExtensions:")
-            for domain in cert.subject_alternative_names:
-                print("    " + domain)
-        print("")
+        _print_cert_details(cert, index)
 
     if include_root and not chain.root:
         sys.stderr.write("WARNING: Root certificate was requested, but not found!\n")
 
+    if chain.cross_signs:
+        for index, cert in enumerate(chain.cross_signs, 1):
+            _print_cert_details(cert, index, heading="Cross-sign")
+        if not include_cross_signs:
+            sys.stderr.write(
+                "Note: {0} cross-signed variant(s) discovered. "
+                "Pass --include-cross-signs to add them to the bundle.\n".format(
+                    len(chain.cross_signs)
+                )
+            )
+
 
-def cli(file_bytes, show_details=False, include_root=False, root_ca_store=None):
-    # type: (bytes, bool, bool, Optional[CAStore]) -> None
+def cli(
+    file_bytes,
+    show_details=False,
+    include_root=False,
+    include_cross_signs=False,
+    root_ca_store=None,
+):
+    # type: (bytes, bool, bool, bool, Optional[CAStore]) -> None
     chain = resolve(file_bytes, root_ca_store=root_ca_store)
     if show_details:
-        _print_chain_details(chain, include_root=include_root)
+        _print_chain_details(
+            chain, include_root=include_root, include_cross_signs=include_cross_signs
+        )
     else:
         root = [chain.root] if chain.root and include_root else []
-        for c in [chain.leaf] + list(chain.intermediates) + root:
+        cross_signs = list(chain.cross_signs) if include_cross_signs else []
+        bundle = [chain.leaf] + list(chain.intermediates) + root + cross_signs
+        for c in bundle:
             sys.stdout.write(c.export())
 
-        for i, c in enumerate([chain.leaf] + list(chain.intermediates) + root, 1):
+        for i, c in enumerate(bundle, 1):
             sys.stderr.write(str(i) + ". " + repr(c) + "\n")
 
         if not root and include_root:
@@ -63,6 +88,14 @@ def cli(file_bytes, show_details=False, include_root=False, root_ca_store=None):
                     "Consider running the CLI with --use-certifi-store to find the matching root CA\n"
                 )
 
+        if chain.cross_signs and not include_cross_signs:
+            sys.stderr.write(
+                "Note: {0} cross-signed variant(s) discovered. "
+                "Pass --include-cross-signs to add them to the bundle.\n".format(
+                    len(chain.cross_signs)
+                )
+            )
+
 
 def parse_args():
     parser = argparse.ArgumentParser(
@@ -96,6 +129,15 @@ def parse_args():
         action="store_true",
         help="Include root certificate in the chain if available",
     )
+    parser.add_argument(
+        "--include-cross-signs",
+        action="store_true",
+        help=(
+            "Append cross-signed variants of the chain (same Subject and public key, "
+            "different Issuer) to the bundle output. Useful for max compatibility with "
+            "older clients that don't yet trust the modern root."
+        ),
+    )
     parser.add_argument(
         "--ca-bundle-path",
         type=str,
@@ -115,6 +157,7 @@ def main():
         "file_bytes": None,
         "show_details": args.info,
         "include_root": args.include_root,
+        "include_cross_signs": args.include_cross_signs,
     }
 
     cli_args["root_ca_store"] = FileSystemStore(args.ca_bundle_path)

cert_chain_resolver/models.py

@@ -2,7 +2,7 @@
 from cert_chain_resolver.utils import load_ascii_to_x509, load_bytes_to_x509
 from cryptography import x509
 from cryptography.x509.oid import ExtensionOID, AuthorityInformationAccessOID, NameOID
-from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.serialization import Encoding
 from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
 from cryptography.hazmat.primitives.asymmetric.ec import ECDSA, EllipticCurvePublicKey
@@ -12,7 +12,6 @@
 
 import binascii
 
-
 try:
     from typing import List, Union, Optional, Type, Iterator, TYPE_CHECKING
 
@@ -136,7 +135,7 @@ def signature_hash_algorithm(self):
     def not_valid_before(self):
         # type: () -> datetime.datetime
         """Date from the underlying :py:class:`cryptography.x509.Certificate` object. returns the UTC version if cryptography version is 42.0 or higher"""
-        if hasattr(self._x509, 'not_valid_before_utc'):
+        if hasattr(self._x509, "not_valid_before_utc"):
             return self._x509.not_valid_before_utc
         else:
             return self._x509.not_valid_before
@@ -145,7 +144,7 @@ def not_valid_before(self):
     def not_valid_after(self):
         # type: () -> datetime.datetime
         """Date from the underlying :py:class:`cryptography.x509.Certificate` object. returns the UTC version if cryptography version is 42.0 or higher"""
-        if hasattr(self._x509, 'not_valid_after_utc'):
+        if hasattr(self._x509, "not_valid_after_utc"):
             return self._x509.not_valid_after_utc
         else:
             return self._x509.not_valid_after
@@ -156,6 +155,22 @@ def fingerprint(self):
         """ascii encoded sha256 fingerprint by calling :py:func:`get_fingerprint`"""
         return self.get_fingerprint(hashes.SHA256)
 
+    @property
+    def public_key_fingerprint(self):
+        # type: () -> str
+        """SHA-256 hex digest of the certificate's SubjectPublicKeyInfo.
+
+        Two certificates with the same Subject and same ``public_key_fingerprint``
+        represent the same identity - this is how cross-signed variants of a CA
+        certificate are detected, since they share both fields but differ in Issuer.
+        """
+        spki = self._x509.public_key().public_bytes(
+            Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo
+        )
+        digest = hashes.Hash(hashes.SHA256())
+        digest.update(spki)
+        return binascii.hexlify(digest.finalize()).decode("ascii")
+
     @property
     def ca_issuer_access_location(self):
         # type: () -> Union[str, None]
@@ -188,6 +203,21 @@ def get_fingerprint(self, _hash=hashes.SHA256):
         txt = binascii.hexlify(binary).decode("ascii")
         return txt
 
+    def is_cross_sign_of(self, other):
+        # type: (Cert) -> bool
+        """Whether this certificate is a cross-signed variant of ``other``.
+
+        Two certificates are cross-signs of each other when they share Subject and
+        Subject Public Key Info but were issued by different CAs - i.e. they represent
+        the same identity but provide alternative paths to different trust anchors.
+        A certificate is never its own cross-sign.
+        """
+        return (
+            self != other
+            and self.subject == other.subject
+            and self.public_key_fingerprint == other.public_key_fingerprint
+        )
+
     def is_issued_by(self, other):
         # type: (Cert) -> bool
         """Verify if certificate is issued by the passed CA cert"""
@@ -258,6 +288,8 @@ def __init__(self, chain=None):
         self._fingerprints = (
             set() if not chain else {x509_obj.fingerprint for x509_obj in chain}
         )
+        self._cross_signs = []  # type: List[Cert]
+        self._cross_sign_fingerprints = set()  # type: set
 
     def __iter__(self):
         # type: () -> Iterator[Cert]
@@ -299,6 +331,26 @@ def root(self):
             return self._chain[-1]
         return None
 
+    @property
+    def cross_signs(self):
+        # type: () -> List[Cert]
+        """Cross-signed variants discovered alongside the chain.
+
+        A cross-sign is a certificate that shares Subject and public key with a cert
+        already in the chain but is issued by a different CA. Modern clients ignore
+        them; older clients can use them to build a path to a different trust anchor.
+        Returns a shallow copy.
+        """
+        return list(self._cross_signs)
+
+    def add_cross_sign(self, cert):
+        # type: (Cert) -> None
+        """Add a cross-signed variant to the chain, deduplicating by fingerprint."""
+        if cert.fingerprint in self._cross_sign_fingerprints:
+            return
+        self._cross_signs.append(cert)
+        self._cross_sign_fingerprints.add(cert.fingerprint)
+
     @classmethod
     def load_from_pem(cls, input_bytes):
         # type: (bytes) -> CertificateChain

cert_chain_resolver/resolver.py

@@ -1,6 +1,7 @@
 from contextlib import closing
 
 from cert_chain_resolver.models import CertificateChain, Cert
+from cert_chain_resolver.utils import load_bytes_to_x509_all
 
 try:
     from urllib.request import urlopen, Request
@@ -51,12 +52,19 @@ def resolve(bytes_cert, _chain=None, root_ca_store=None):
 
     _chain += cert
 
-    parent_cert = None
+    parent_bytes = None
     if cert.ca_issuer_access_location:
-        parent_cert = _download(cert.ca_issuer_access_location)
-
-    if parent_cert:
-        return resolve(parent_cert, _chain=_chain, root_ca_store=root_ca_store)
+        bundle_bytes = _download(cert.ca_issuer_access_location)
+        parents = load_bytes_to_x509_all(bundle_bytes)
+        primary = Cert(parents[0])
+        for extra_x509 in parents[1:]:
+            extra = Cert(extra_x509)
+            if extra.is_cross_sign_of(primary):
+                _chain.add_cross_sign(extra)
+        parent_bytes = bundle_bytes
+
+    if parent_bytes:
+        return resolve(parent_bytes, _chain=_chain, root_ca_store=root_ca_store)
     elif not _chain.root and root_ca_store:
         _chain += root_ca_store.find_issuer(cert)