feat!: pre-load CA certification chains on otaclient starts up, rejec…

…t OTA if no CA certs are installed (#408)

This PR resolves an issue that comes from very old version of otaclient(since v1.0.0), which results in OTA image sign cert verification being skipped when no valid CA cert chains can be imported by otaclient. Also the logic of loading and managing the CA cert chains are split away from ota_metadata.legacy package and becomes a standalone module in ota_metadata.utils.cert_store.

BREAKING CHANGE: Now otaclient will explicitly reject OTA when otaclient is not installed with valid CA cert chains. Also otaclient now only loads the CA cert chains once at startup. Re-configuring CA chains now requires otaclient restart.

docker/test_base/Dockerfile

@@ -4,18 +4,17 @@ ARG UBUNTU_BASE
 # ------ stage 1: prepare base image ------ #
 #
 
-FROM ${UBUNTU_BASE} AS builder
+FROM ${UBUNTU_BASE} AS image_build
 
 SHELL ["/bin/bash", "-c"]
 ENV DEBIAN_FRONTEND=noninteractive
 ENV SPECIAL_FILE="path;adf.ae?qu.er\y=str#fragファイルement"
 
 # special treatment to the ota-image: create file that needs url escaping
 # NOTE: include special identifiers #?; into the pathname
-RUN echo -n "${SPECIAL_FILE}" > "${OTA_IMAGE_DIR}/${SPECIAL_FILE}"
-
-# install required packages
 RUN set -eux; \
+    echo -n "${SPECIAL_FILE}" > "/${SPECIAL_FILE}"; \
+    # install required packages
     apt-get update -qq; \
     apt-get install -y linux-image-generic; \
     apt-get clean; \
@@ -44,26 +43,24 @@ ENV SPECIAL_FILE="path;adf.ae?qu.er\y=str#fragファイルement"
 RUN set -eux; \
     apt-get update -qq; \
     apt-get install -y -qq --no-install-recommends \
+	    gcc \
+        git \
+        libcurl4-openssl-dev \
+        libssl-dev \
+        python3-dev \
         python3-minimal \
         python3-pip \
         python3-venv \
-        python3-dev \
-	    libcurl4-openssl-dev \
-	    libssl-dev \
-	    gcc \
-        wget \
-        git; \
+        wget; \
     apt-get install -y -qq linux-image-generic; \
-    apt-get clean
-
-# install hatch
-RUN set -eux; \
+    apt-get clean; \
+    # install hatch
     python3 -m pip install --no-cache-dir -q -U pip; \
     python3 -m pip install --no-cache-dir -U hatch
 
 WORKDIR ${OTA_IMAGE_SERVER_ROOT}
 
-COPY --from=builder / /${OTA_IMAGE_DIR}
+COPY --from=image_build / /${OTA_IMAGE_DIR}
 
 # generate test certs and sign key
 COPY --chmod=755 ./tests/keys/gen_certs.sh /tmp/certs/
@@ -72,10 +69,8 @@ RUN set -eu; \
     pushd /tmp/certs; \
     ./gen_certs.sh; \
     cp ./* "${CERTS_DIR}"; \
-    popd
-
-# build the test OTA image
-RUN set -eu; \
+    popd; \
+    # build the test OTA image
     cp "${CERTS_DIR}"/sign.key sign.key; \
     cp "${CERTS_DIR}"/sign.pem sign.pem; \
     git clone ${OTA_METADATA_REPO}; \
@@ -96,10 +91,8 @@ RUN set -eu; \
         --persistent-file ota-metadata/metadata/persistents.txt \
         --rootfs-directory data \
         --compressed-rootfs-directory data.zst; \
-    cp ota-metadata/metadata/persistents.txt .
-
-# cleanup
-RUN set -eux; \
+    cp ota-metadata/metadata/persistents.txt .; \
+    # cleanup
     apt-get clean; \
     rm -rf \
         /tmp/* \

docker/test_base/entry_point.sh

@@ -4,18 +4,10 @@ set -eu
 TEST_ROOT=/test_root
 OTACLIENT_SRC=/otaclient_src
 OUTPUT_DIR="${OUTPUT_DIR:-/test_result}"
-CERTS_DIR="${CERTS_DIR:-/certs}"
 
 # copy the source code as source is read-only
 cp -R "${OTACLIENT_SRC}" "${TEST_ROOT}"
 
-# copy the certs generated in the docker image to otaclient dir
-echo "setup certificates for testing..."
-cd ${TEST_ROOT}
-mkdir -p ./certs
-cp -av ${CERTS_DIR}/root.pem ./certs/1.root.pem
-cp -av ${CERTS_DIR}/interm.pem ./certs/1.interm.pem
-
 # exec the input params
 echo "execute test with coverage"
 cd ${TEST_ROOT}

src/ota_metadata/legacy/parser.py

@@ -47,7 +47,6 @@
 import shutil
 import time
 from dataclasses import dataclass, fields
-from functools import partial
 from os import PathLike
 from pathlib import Path
 from tempfile import NamedTemporaryFile, TemporaryDirectory
@@ -71,6 +70,7 @@
 from OpenSSL import crypto
 from typing_extensions import Self
 
+from ota_metadata.utils.cert_store import CACertChainStore, load_cert_in_pem
 from ota_proxy import OTAFileCacheControl
 from otaclient_common.common import urljoin_ensure_base
 from otaclient_common.downloader import Downloader
@@ -245,8 +245,8 @@ class _MetadataJWTParser:
 
     HASH_ALG = "sha256"
 
-    def __init__(self, metadata_jwt: str, *, certs_dir: Union[str, Path]):
-        self.cert_dir = Path(certs_dir)
+    def __init__(self, metadata_jwt: str, *, ca_chains_store: CACertChainStore):
+        self.ca_chains_store = ca_chains_store
 
         # pre_parse metadata_jwt
         jwt_list = metadata_jwt.split(".")
@@ -267,46 +267,32 @@ def __init__(self, metadata_jwt: str, *, certs_dir: Union[str, Path]):
         self.ota_metadata = _MetadataJWTClaimsLayout.parse_payload(self.payload_bytes)
         logger.info(f"metadata={self.ota_metadata!r}")
 
-    def _verify_metadata_cert(self, metadata_cert: bytes) -> None:
+    def verify_metadata_cert(self, metadata_cert: bytes) -> None:
         """Verify the metadata's sign certificate against local pinned CA.
 
         Raises:
             Raise MetadataJWTVerificationFailed on verification failed.
         """
-        ca_set_prefix = set()
-        # e.g. under _certs_dir: A.1.pem, A.2.pem, B.1.pem, B.2.pem
-        for cert in self.cert_dir.glob("*.*.pem"):
-            if m := re.match(r"(.*)\..*.pem", cert.name):
-                ca_set_prefix.add(m.group(1))
-            else:
-                raise MetadataJWTVerificationFailed("no pem file is found")
-        if len(ca_set_prefix) == 0:
-            logger.warning("there is no root or intermediate certificate")
-            return
-        logger.info(f"certs prefixes {ca_set_prefix}")
-
-        load_pem = partial(crypto.load_certificate, crypto.FILETYPE_PEM)
+        if not self.ca_chains_store:
+            _err_msg = "CA chains store is empty!!! immediately fail the verification"
+            logger.error(_err_msg)
+            raise MetadataJWTVerificationFailed(_err_msg)
 
         try:
-            cert_to_verify = load_pem(metadata_cert)
+            cert_to_verify = load_cert_in_pem(metadata_cert)
         except crypto.Error as e:
             _err_msg = f"invalid certificate {metadata_cert}: {e!r}"
             logger.exception(_err_msg)
             raise MetadataJWTVerificationFailed(_err_msg) from e
 
-        for ca_prefix in sorted(ca_set_prefix):
-            certs_list = [
-                self.cert_dir / c.name for c in self.cert_dir.glob(f"{ca_prefix}.*.pem")
-            ]
-
-            store = crypto.X509Store()
-            for c in certs_list:
-                logger.info(f"cert {c}")
-                store.add_cert(load_pem(c.read_bytes()))
-
+        # verify the OTA image cert against CA cert chain store
+        for ca_prefix, ca_chain in self.ca_chains_store.items():
             try:
-                store_ctx = crypto.X509StoreContext(store, cert_to_verify)
-                store_ctx.verify_certificate()
+                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:
@@ -316,7 +302,7 @@ def _verify_metadata_cert(self, metadata_cert: bytes) -> None:
         logger.error(_err_msg)
         raise MetadataJWTVerificationFailed(_err_msg)
 
-    def _verify_metadata(self, metadata_cert: bytes):
+    def verify_metadata_signature(self, metadata_cert: bytes):
         """Verify metadata against sign certificate.
 
         Raises:
@@ -344,17 +330,6 @@ def get_otametadata(self) -> "_MetadataJWTClaimsLayout":
         """
         return self.ota_metadata
 
-    def verify_metadata(self, metadata_cert: bytes):
-        """Verify metadata_jwt against metadata cert and local pinned CA certs.
-
-        Raises:
-            Raise MetadataJWTVerificationFailed on verification failed.
-        """
-        # step1: verify the cert itself against local pinned CA cert
-        self._verify_metadata_cert(metadata_cert)
-        # step2: verify the metadata against input metadata_cert
-        self._verify_metadata(metadata_cert)
-
 
 # place holder for unset must field in _MetadataJWTClaimsLayout
 _MUST_SET_CLAIM = object()
@@ -615,13 +590,18 @@ def __init__(
         url_base: str,
         downloader: Downloader,
         run_dir: Path,
-        certs_dir: Path,
+        ca_chains_store: CACertChainStore,
         retry_interval: int = 1,
     ) -> None:
+        if not ca_chains_store:
+            _err_msg = "CA chains store is empty!!! immediately fail the verification"
+            logger.error(_err_msg)
+            raise MetadataJWTVerificationFailed(_err_msg)
+
         self.url_base = url_base
         self._downloader = downloader
         self.run_dir = run_dir
-        self.certs_dir = certs_dir
+        self.ca_chains_store = ca_chains_store
         self.retry_interval = retry_interval
         self._tmp_dir = TemporaryDirectory(prefix="ota_metadata", dir=run_dir)
         self._tmp_dir_path = Path(self._tmp_dir.name)
@@ -673,7 +653,8 @@ def _process_metadata_jwt(self) -> _MetadataJWTClaimsLayout:
                     time.sleep(self.retry_interval)
 
             _parser = _MetadataJWTParser(
-                _downloaded_meta_f.read_text(), certs_dir=self.certs_dir
+                _downloaded_meta_f.read_text(),
+                ca_chains_store=self.ca_chains_store,
             )
         # get not yet verified parsed ota_metadata
         _ota_metadata = _parser.get_otametadata()
@@ -700,7 +681,11 @@ def _process_metadata_jwt(self) -> _MetadataJWTClaimsLayout:
                 except Exception as e:
                     logger.warning(f"failed to download {cert_info}, retrying: {e!r}")
                     time.sleep(self.retry_interval)
-            _parser.verify_metadata(cert_file.read_bytes())
+
+            cert_bytes = cert_file.read_bytes()
+
+            _parser.verify_metadata_cert(cert_bytes)
+            _parser.verify_metadata_signature(cert_bytes)
 
         # return verified ota metadata
         return _ota_metadata

src/ota_metadata/utils/__init__.py

@@ -0,0 +1,13 @@
+# Copyright 2022 TIER IV, INC. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

src/ota_metadata/utils/cert_store.py

@@ -0,0 +1,93 @@
+# Copyright 2022 TIER IV, INC. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Implementation of otaclient CA cert chain store."""
+
+
+from __future__ import annotations
+
+import logging
+import re
+from functools import partial
+from pathlib import Path
+from typing import Dict
+
+from OpenSSL import crypto
+
+from otaclient_common.typing import StrOrPath
+
+logger = logging.getLogger(__name__)
+
+# we search the CA chains with the following naming schema:
+#   <env_name>.<intermediate|root>.pem,
+#   in which <env_name> should match regex pattern `[\w\-]+`
+# e.g:
+#   dev chain: dev.intermediate.pem, dev.root.pem
+#   prd chain: prd.intermediate.pem, prd.intermediate.pem
+CERT_NAME_PA = re.compile(r"(?P<chain>[\w\-]+)\.[\w\-]+\.pem")
+
+
+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)
+
+
+def load_ca_cert_chains(cert_dir: StrOrPath) -> CACertChainStore:
+    """Load CA cert chains from <cert_dir>.
+
+    Raises:
+        CACertStoreInvalid on failed import.
+
+    Returns:
+        A dict
+    """
+    cert_dir = Path(cert_dir)
+
+    ca_set_prefix = set()
+    for cert in cert_dir.glob("*.pem"):
+        if m := CERT_NAME_PA.match(cert.name):
+            ca_set_prefix.add(m.group("chain"))
+        else:
+            _err_msg = f"detect invalid named certs: {cert.name}"
+            logger.warning(_err_msg)
+
+    if not ca_set_prefix:
+        _err_msg = f"no CA cert chains found to be installed under the {cert_dir}!!!"
+        logger.error(_err_msg)
+        raise CACertStoreInvalid(_err_msg)
+
+    logger.info(f"found installed CA chains: {ca_set_prefix}")
+
+    ca_chains = CACertChainStore()
+    for ca_prefix in sorted(ca_set_prefix):
+        try:
+            ca_chain = crypto.X509Store()
+            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
+        except Exception as e:
+            _err_msg = f"failed to load CA chain {ca_prefix}: {e!r}"
+            logger.warning(_err_msg)
+
+    if not ca_chains:
+        _err_msg = "all found CA chains are invalid, no CA chain is imported!!!"
+        logger.error(_err_msg)
+        raise CACertStoreInvalid(_err_msg)
+
+    logger.info(f"loaded CA cert chains: {list(ca_chains)}")
+    return ca_chains