feat!: reject any OTA update/rollback request on ecu_info.yaml not properly loaded

src/otaclient/_status_monitor.py

@@ -318,7 +318,15 @@ def _status_collector_thread(self) -> None:
                     break
 
                 # ------ push status on load_report ------ #
-                if self.load_report(report) and self._status and _now > _next_shm_push:
+                # NOTE: always push OTAStatus change report
+                if (
+                    self.load_report(report)
+                    and self._status
+                    and (
+                        isinstance(report.payload, OTAStatusChangeReport)
+                        or _now > _next_shm_push
+                    )
+                ):
                     try:
                         self._shm_status.write_msg(self._status)
                         _next_shm_push = _now + self.shm_push_interval

src/otaclient/grpc/api_v2/servicer.py

@@ -21,6 +21,7 @@
 import multiprocessing.queues as mp_queue
 from concurrent.futures import ThreadPoolExecutor
 
+import otaclient.configs.cfg as otaclient_cfg
 from otaclient._types import (
     IPCRequest,
     IPCResEnum,
@@ -111,6 +112,22 @@ async def update(
         update_acked_ecus = set()
         response = api_types.UpdateResponse()
 
+        # NOTE(20241220): due to the fact that OTA Service API doesn't have field
+        #                 in UpdateResponseEcu msg, the only way to pass the failure_msg
+        #                 to upper is by status API.
+        if not otaclient_cfg.ECU_INFO_LOADED_SUCCESSFULLY:
+            logger.error(
+                "ecu_info.yaml is not loaded properly, reject any update request"
+            )
+            for _update_req in request.iter_ecu():
+                response.add_ecu(
+                    api_types.UpdateResponseEcu(
+                        ecu_id=_update_req.ecu_id,
+                        result=api_types.FailureType.UNRECOVERABLE,
+                    )
+                )
+            return response
+
         # first: dispatch update request to all directly connected subECUs
         tasks: dict[asyncio.Task, ECUContact] = {}
         for ecu_contact in self.sub_ecus:
@@ -225,6 +242,22 @@ async def rollback(
         logger.info(f"receive rollback request: {request}")
         response = api_types.RollbackResponse()
 
+        # NOTE(20241220): due to the fact that OTA Service API doesn't have field
+        #                 in UpdateResponseEcu msg, the only way to pass the failure_msg
+        #                 to upper is by status API.
+        if not otaclient_cfg.ECU_INFO_LOADED_SUCCESSFULLY:
+            logger.error(
+                "ecu_info.yaml is not loaded properly, reject any rollback request"
+            )
+            for _rollback_req in request.iter_ecu():
+                response.add_ecu(
+                    api_types.RollbackResponseEcu(
+                        ecu_id=_rollback_req.ecu_id,
+                        result=api_types.FailureType.UNRECOVERABLE,
+                    )
+                )
+            return response
+
         # first: dispatch rollback request to all directly connected subECUs
         tasks: dict[asyncio.Task, ECUContact] = {}
         for ecu_contact in self.sub_ecus:

src/otaclient/ota_core.py

@@ -36,6 +36,7 @@
 import requests.exceptions as requests_exc
 from requests import Response
 
+import otaclient.configs.cfg as otaclient_cfg
 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 (
@@ -672,17 +673,7 @@ def __init__(
             )
             return
 
-        # load and report booted OTA status
-        _boot_ctrl_loaded_ota_status = self.boot_controller.get_booted_ota_status()
-        self._live_ota_status = _boot_ctrl_loaded_ota_status
-        status_report_queue.put_nowait(
-            StatusReport(
-                payload=OTAStatusChangeReport(
-                    new_ota_status=_boot_ctrl_loaded_ota_status,
-                ),
-            )
-        )
-
+        # ------ load firmware version ------ #
         self.current_version = self.boot_controller.load_version()
         status_report_queue.put_nowait(
             StatusReport(
@@ -691,7 +682,9 @@ def __init__(
                 ),
             )
         )
+        logger.info(f"firmware_version: {self.current_version}")
 
+        # ------ load CA store ------ #
         self.ca_chains_store = None
         try:
             self.ca_chains_store = load_ca_cert_chains(cfg.CERT_DPATH)
@@ -701,9 +694,36 @@ def __init__(
 
             self.ca_chains_store = CAChainStore()
 
-        self.started = True
-        logger.info("otaclient started")
-        logger.info(f"firmware_version: {self.current_version}")
+        # load and report booted OTA status
+        _boot_ctrl_loaded_ota_status = self.boot_controller.get_booted_ota_status()
+        if not otaclient_cfg.ECU_INFO_LOADED_SUCCESSFULLY:
+            logger.error(
+                "ecu_info.yaml file is not loaded properly, will reject any OTA request."
+            )
+            logger.error(f"set live_ota_status to {OTAStatus.FAILURE!r}")
+            self._live_ota_status = OTAStatus.FAILURE
+            status_report_queue.put_nowait(
+                StatusReport(
+                    payload=OTAStatusChangeReport(
+                        new_ota_status=OTAStatus.FAILURE,
+                        failure_type=FailureType.UNRECOVERABLE,
+                        failure_reason=f"ecu_info.yaml file is broken or missing, please check {cfg.ECU_INFO_FPATH}. "
+                        "reject any OTA request.",
+                    ),
+                )
+            )
+        else:
+            self._live_ota_status = _boot_ctrl_loaded_ota_status
+            status_report_queue.put_nowait(
+                StatusReport(
+                    payload=OTAStatusChangeReport(
+                        new_ota_status=_boot_ctrl_loaded_ota_status,
+                    ),
+                )
+            )
+
+            self.started = True
+            logger.info("otaclient started")
 
     def _on_failure(
         self,
@@ -843,6 +863,19 @@ def main(
                     )
                 )
 
+            elif not self.started:
+                _err_msg = (
+                    "otaclient is not started, might be due to broken ecu_info.yaml"
+                )
+                logger.error(_err_msg)
+                resp_queue.put_nowait(
+                    IPCResponse(
+                        res=IPCResEnum.REJECT_OTHER,
+                        msg=_err_msg,
+                        session_id=request.session_id,
+                    )
+                )
+
             elif isinstance(request, UpdateRequestV2):
 
                 _update_thread = threading.Thread(