refactor: drop pyopenssl dependency, use cryptography to re-implement…

… OTA image verification logic (#437)

This PR drops the pyopenssl dependency, refactors the OTA image verification logic with cryptography instead. 
Note that we don't use the cryptography's high-level x509 verification APIs as that APIs are dedicated for TLS. Instead we directly use low-level APIs to perform certificate signature & valid period check.

This PR pins the cryptography version to >=42.0.8, <45.

Major changes:
1. ota_metadata.utils.cert_store: re-implement with cryptography.
2. ota_metadata.legacy.parser: integrate the new cert_store.

pyproject.toml

@@ -26,12 +26,11 @@ dynamic = [
 dependencies = [
   "aiofiles<25,>=24.1",
   "aiohttp>=3.10.11,<3.12",
-  "cryptography>=42.0.8,<43",
+  "cryptography>=42.0.8,<45",
   "grpcio>=1.53.2,<1.69",
   "protobuf>=4.21.12,<5.29",
   "pydantic<3,>=2.6",
   "pydantic-settings<3,>=2.3",
-  "pyopenssl==24.1.0",
   "pyyaml<7,>=6.0.1",
   "requests<2.33,>=2.32",
   "simple-sqlite3-orm<0.3,>=0.2",

requirements.txt

@@ -2,12 +2,11 @@
 # DO NOT EDIT! Only for reference use.
 aiofiles<25,>=24.1
 aiohttp>=3.10.11,<3.12
-cryptography>=42.0.8,<43
+cryptography>=42.0.8,<45
 grpcio>=1.53.2,<1.69
 protobuf>=4.21.12,<5.29
 pydantic<3,>=2.6
 pydantic-settings<3,>=2.3
-pyopenssl==24.1.0
 pyyaml<7,>=6.0.1
 requests<2.33,>=2.32
 simple-sqlite3-orm<0.3,>=0.2

src/ota_metadata/legacy/parser.py

@@ -67,10 +67,12 @@
 )
 from urllib.parse import quote
 
-from OpenSSL import crypto
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric.ec import ECDSA, EllipticCurvePublicKey
+from cryptography.x509 import load_pem_x509_certificate
 from typing_extensions import Self
 
-from ota_metadata.utils.cert_store import CACertChainStore, load_cert_in_pem
+from ota_metadata.utils.cert_store import CAChainStore
 from ota_proxy import OTAFileCacheControl
 from otaclient_common.common import urljoin_ensure_base
 from otaclient_common.downloader import Downloader
@@ -91,6 +93,7 @@
 logger = logging.getLogger(__name__)
 
 CACHE_CONTROL_HEADER = OTAFileCacheControl.HEADER_LOWERCASE
+ES256 = ECDSA(algorithm=hashes.SHA256())
 
 _shutdown = False
 
@@ -243,9 +246,7 @@ class _MetadataJWTParser:
         will be skipped! This SHOULD ONLY happen at non-production environment!
     """
 
-    HASH_ALG = "sha256"
-
-    def __init__(self, metadata_jwt: str, *, ca_chains_store: CACertChainStore):
+    def __init__(self, metadata_jwt: str, *, ca_chains_store: CAChainStore):
         self.ca_chains_store = ca_chains_store
 
         # pre_parse metadata_jwt
@@ -279,24 +280,15 @@ def verify_metadata_cert(self, metadata_cert: bytes) -> None:
             raise MetadataJWTVerificationFailed(_err_msg)
 
         try:
-            cert_to_verify = load_cert_in_pem(metadata_cert)
-        except crypto.Error as e:
+            cert_to_verify = load_pem_x509_certificate(metadata_cert)
+        except Exception as e:
             _err_msg = f"invalid certificate {metadata_cert}: {e!r}"
             logger.exception(_err_msg)
             raise MetadataJWTVerificationFailed(_err_msg) from e
 
-        # verify the OTA image cert against CA cert chain store
-        for ca_prefix, ca_chain in self.ca_chains_store.items():
-            try:
-                crypto.X509StoreContext(
-                    store=ca_chain,
-                    certificate=cert_to_verify,
-                ).verify_certificate()
-
-                logger.info(f"verfication succeeded against: {ca_prefix}")
-                return
-            except crypto.X509StoreContextError as e:
-                logger.info(f"verify against {ca_prefix} failed: {e}")
+        hit_cachain = self.ca_chains_store.verify(cert_to_verify)
+        if hit_cachain:
+            return
 
         _err_msg = f"metadata sign certificate {metadata_cert} could not be verified"
         logger.error(_err_msg)
@@ -309,15 +301,16 @@ def verify_metadata_signature(self, metadata_cert: bytes):
             Raise MetadataJWTVerificationFailed on validation failed.
         """
         try:
-            cert = crypto.load_certificate(crypto.FILETYPE_PEM, metadata_cert)
+            cert = load_pem_x509_certificate(metadata_cert)
             logger.debug(f"verify data: {self.metadata_bytes=}")
-            crypto.verify(
-                cert,
-                self.metadata_signature,
-                self.metadata_bytes,
-                self.HASH_ALG,
+            _pubkey: EllipticCurvePublicKey = cert.public_key()  # type: ignore[assignment]
+
+            _pubkey.verify(
+                signature=self.metadata_signature,
+                data=self.metadata_bytes,
+                signature_algorithm=ES256,
             )
-        except crypto.Error as e:
+        except Exception as e:
             msg = f"failed to verify metadata against sign cert: {e!r}"
             logger.error(msg)
             raise MetadataJWTVerificationFailed(msg) from e
@@ -590,7 +583,7 @@ def __init__(
         url_base: str,
         downloader: Downloader,
         run_dir: Path,
-        ca_chains_store: CACertChainStore,
+        ca_chains_store: CAChainStore,
         retry_interval: int = 1,
     ) -> None:
         if not ca_chains_store:

src/ota_metadata/utils/cert_store.py

@@ -18,11 +18,11 @@
 
 import logging
 import re
-from functools import partial
+from datetime import datetime, timezone
 from pathlib import Path
-from typing import Dict
+from typing import Dict, Iterable
 
-from OpenSSL import crypto
+from cryptography.x509 import Certificate, Name, load_pem_x509_certificate
 
 from otaclient_common.typing import StrOrPath
 
@@ -40,14 +40,105 @@
 class CACertStoreInvalid(Exception): ...
 
 
-class CACertChainStore(Dict[str, crypto.X509Store]):
-    """A dict-like type that stores CA chain name and CA store mapping."""
-
-
-load_cert_in_pem = partial(crypto.load_certificate, crypto.FILETYPE_PEM)
+class CAChain(Dict[Name, Certificate]):
+    """A dict that stores all the CA certs of a cert chains.
 
+    The key is the cert's subject, value is the cert itself.
+    """
 
-def load_ca_cert_chains(cert_dir: StrOrPath) -> CACertChainStore:
+    chain_prefix: str = ""
+
+    def set_chain_prefix(self, prefix: str) -> None:
+        self.chain_prefix = prefix
+
+    def add_cert(self, cert: Certificate) -> None:
+        self[cert.subject] = cert
+
+    def add_certs(self, certs: Iterable[Certificate]) -> None:
+        for cert in certs:
+            self[cert.subject] = cert
+
+    def internal_check(self) -> None:
+        """Do an internal check to see if this CACertChain is valid.
+
+        Currently one check will be performed:
+        1. at least one root cert should be presented in the store.
+
+        Raises:
+            ValueError on failed check.
+        """
+        for _, cert in self.items():
+            if cert.issuer == cert.subject:
+                return
+        raise ValueError("invalid chain: no root cert is presented")
+
+    def verify(self, cert: Certificate) -> None:
+        """Verify the input <cert> against this chain.
+
+        Raises:
+            ValueError on input cert is not signed by this chain.
+            Other exceptions that could be raised by verify_directly_issued_by API.
+
+        Returns:
+            Return None on successful verification, otherwise raises exception.
+        """
+        _now = datetime.now(tz=timezone.utc)
+        if not (cert.not_valid_after_utc >= _now >= cert.not_valid_before_utc):
+            _err_msg = (
+                "cert is not within valid period: "
+                f"{cert.not_valid_after_utc=}, {_now=}, {cert.not_valid_before_utc=}"
+            )
+            raise ValueError(_err_msg)
+
+        _start = cert
+        for _ in range(len(self) + 1):
+            if _start.issuer == _start.subject:
+                return
+
+            _issuer = self[_start.issuer]
+            _start.verify_directly_issued_by(_issuer)
+
+            _start = _issuer
+        raise ValueError(f"failed to verify {cert} against chain {self.chain_prefix}")
+
+
+class CAChainStore(Dict[str, CAChain]):
+    """A dict that stores CA chain name and CACertChains mapping."""
+
+    def add_chain(self, chain: CAChain) -> None:
+        self[chain.chain_prefix] = chain
+
+    def verify(self, cert: Certificate) -> CAChain | None:
+        """Verify the input <cert> against this CAChainStore.
+
+        This verification only performs the following check:
+        1. ensure the cert is in valid period.
+        2. check cert is signed by any one of the CA chain.
+
+        Returns:
+            Return the cachain that issues the <cert>, None if no cachain matches.
+        """
+        _now = datetime.now(tz=timezone.utc)
+        if not (cert.not_valid_after_utc >= _now >= cert.not_valid_before_utc):
+            logger.error(
+                "cert is not within valid period: "
+                f"{cert.not_valid_after_utc=}, {_now=}, {cert.not_valid_before_utc=}"
+            )
+            return
+
+        for _, chain in self.items():
+            try:
+                chain.verify(cert)
+                logger.info(f"verfication succeeded against: {chain.chain_prefix}")
+                return chain
+            except Exception as e:
+                logger.info(
+                    f"failed to verify against CA chain {chain.chain_prefix}: {e!r}"
+                )
+        logger.error(f"failed to verify {cert=} against all CA chains: {list(self)}")
+
+
+def load_ca_cert_chains(cert_dir: StrOrPath) -> CAChainStore:
     """Load CA cert chains from <cert_dir>.
 
     Raises:
@@ -73,13 +164,18 @@ def load_ca_cert_chains(cert_dir: StrOrPath) -> CACertChainStore:
 
     logger.info(f"found installed CA chains: {ca_set_prefix}")
 
-    ca_chains = CACertChainStore()
+    ca_chains = CAChainStore()
     for ca_prefix in sorted(ca_set_prefix):
         try:
-            ca_chain = crypto.X509Store()
+            ca_chain = CAChain()
+            ca_chain.set_chain_prefix(ca_prefix)
+
             for c in cert_dir.glob(f"{ca_prefix}.*.pem"):
-                ca_chain.add_cert(load_cert_in_pem(c.read_bytes()))
-            ca_chains[ca_prefix] = ca_chain
+                _loaded_ca_cert = load_pem_x509_certificate(c.read_bytes())
+                ca_chain.add_cert(_loaded_ca_cert)
+
+            ca_chain.internal_check()
+            ca_chains.add_chain(ca_chain)
         except Exception as e:
             _err_msg = f"failed to load CA chain {ca_prefix}: {e!r}"
             logger.warning(_err_msg)

src/otaclient/ota_core.py

@@ -36,8 +36,8 @@
 from ota_metadata.legacy import parser as ota_metadata_parser
 from ota_metadata.legacy import types as ota_metadata_types
 from ota_metadata.utils.cert_store import (
-    CACertChainStore,
     CACertStoreInvalid,
+    CAChainStore,
     load_ca_cert_chains,
 )
 from otaclient import errors as ota_errors
@@ -161,7 +161,7 @@ def __init__(
         version: str,
         raw_url_base: str,
         cookies_json: str,
-        ca_chains_store: CACertChainStore,
+        ca_chains_store: CAChainStore,
         upper_otaproxy: str | None = None,
         boot_controller: BootControllerProtocol,
         create_standby_cls: Type[StandbySlotCreatorProtocol],
@@ -682,7 +682,7 @@ def __init__(
             _err_msg = f"failed to import ca_chains_store: {e!r}, OTA will NOT occur on no CA chains installed!!!"
             logger.error(_err_msg)
 
-            self.ca_chains_store = CACertChainStore()
+            self.ca_chains_store = CAChainStore()
 
         self.started = True
         logger.info("otaclient started")

tests/test_ota_metadata/test_ca_store.py

@@ -20,13 +20,9 @@
 from pathlib import Path
 
 import pytest
-from OpenSSL import crypto
+from cryptography.x509 import load_pem_x509_certificate
 
-from ota_metadata.utils.cert_store import (
-    CACertStoreInvalid,
-    load_ca_cert_chains,
-    load_cert_in_pem,
-)
+from ota_metadata.utils.cert_store import CACertStoreInvalid, load_ca_cert_chains
 from tests.conftest import TEST_DIR
 from tests.conftest import TestConfiguration as cfg
 
@@ -60,22 +56,11 @@ def setup_ca_chain(tmp_path: Path) -> tuple[str, Path, Path]:
 
 
 def test_ca_store(setup_ca_chain: tuple[str, Path, Path]):
-    ca_chain, sign_pem, certs_dir = setup_ca_chain
+    _, sign_pem, certs_dir = setup_ca_chain
+    sign_cert = load_pem_x509_certificate(sign_pem.read_bytes())
 
     ca_store = load_ca_cert_chains(certs_dir)
-
-    # verification should fail with wrong cert signed by other chain.
-    with pytest.raises(crypto.X509StoreContextError):
-        crypto.X509StoreContext(
-            store=ca_store[ca_chain],
-            certificate=load_cert_in_pem(TEST_BASE_SIGN_PEM.read_bytes()),
-        ).verify_certificate()
-
-    # verification should succeed with proper chain and corresponding sign cert.
-    crypto.X509StoreContext(
-        store=ca_store[ca_chain],
-        certificate=load_cert_in_pem(sign_pem.read_bytes()),
-    ).verify_certificate()
+    assert ca_store.verify(sign_cert)
 
 
 def test_ca_store_empty(tmp_path: Path):

tools/offline_ota_image_builder/builder.py

@@ -24,7 +24,7 @@
 from typing import Mapping, Optional, Sequence
 
 from ota_metadata.legacy import parser as ota_metadata_parser
-from ota_metadata.utils.cert_store import CACertChainStore
+from ota_metadata.utils.cert_store import CAChainStore
 from otaclient_common.common import subprocess_call
 
 from .configs import cfg
@@ -75,7 +75,7 @@ def _process_ota_image(ota_image_dir: StrPath, *, data_dir: StrPath, meta_dir: S
     # NOTE: we don't need to do certificate verification here, sso we use an empty cert store.
     metadata_jwt = ota_metadata_parser._MetadataJWTParser(
         metadata_jwt_fpath.read_text(),
-        ca_chains_store=CACertChainStore(),
+        ca_chains_store=CAChainStore(),
     ).get_otametadata()
 
     rootfs_dir = ota_image_dir / metadata_jwt.rootfs_directory