diff --git a/tools/lora_ota/README.md b/tools/lora_ota/README.md index 653f3daf3..a40cb7a33 100644 --- a/tools/lora_ota/README.md +++ b/tools/lora_ota/README.md @@ -12,12 +12,20 @@ wire format. Some deployed adaptive RAK3401, RAK4631 and MeshTower receiver images omit the bootloader ABI and codec mask from `ota self`. For application updates, `lora_ota.py` can verify these known diagnostic formats with an additional -read-only `ota bootloader status` query. Hardware, selected storage and the +`ota bootloader status` query. Hardware, selected storage and the installed bootloader's board/name/target identity, CRC status and storage profile must agree. That status contract guarantees codec subset `0x5` (full image and in-place delta); storage `caps` flags are not interpreted as codecs. Unsupported, incomplete or conflicting replies still stop preflight. +On an idle RAK receiver that reports `storage unsafe: not probed`, the runner +first uses one bounded bootloader-status query to prepare optional staging +media. It then rechecks the running image, hardware, geometry and positive +storage result before applying the usual capability checks. Preparation may +initialize optional staging media; it does not download firmware or change +preferences. Ordinary firmware status getters remain independent of that +media preparation. + This compatibility path applies only to application updates. Bootloader packages require directly reported ABI and codec-mask fields. New adaptive firmware reports those actual fields early in `ota self`, before storage diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index 590db43c8..f9d1587f5 100755 --- a/tools/lora_ota/lora_ota.py +++ b/tools/lora_ota/lora_ota.py @@ -2862,9 +2862,104 @@ def query_bootloader_target( ) +def validate_adaptive_bootloader_status(live: BootloaderStatus, hw_id: str) -> None: + """Require an installed, canonical bootloader for this adaptive family.""" + if hw_id in ("RAK_3401", "RAK4631"): + name = "3401" if hw_id == "RAK_3401" else "4631" + expected_board = 0x239A0029 + identities = {f"{name}_DFU": 0x0A, f"{name}_AUTO_DFU": 0x1E} + elif hw_id == "Heltec_tower_v2": + expected_board = 0x239A0071 + identities = {"TOWER_V2_OTA": 0x09} + else: + raise OtaError("adaptive destination hardware family is unverified") + if ( + live.board_id != expected_board + or identities.get(live.device_name) != live.storage_flags + or not 3 <= live.apply_abi < 0xFFFF + or live.crc32 in (0, 0xFFFFFFFF) + ): + raise OtaError("adaptive destination bootloader identity, ABI or storage profile is unverified") + + +def prepare_cold_rak_application_storage( + controller: Controller, target_name: str, status: str, self_status: str, +) -> tuple[str, str, BootloaderStatus | None]: + """Prepare only the known idle, deferred RAK application-store selector. + + Ordinary status/self getters intentionally do not probe optional storage. + The explicit bootloader status query enters the existing OTA media/staging + preparation boundary without downloading, changing preferences, or writing + firmware. Optional staging media may be initialized by that preparation. + Bind both sides of that one query to the same running application, then + require a complete positive storage diagnosis; preparation is never a + substitute for the normal package/capability checks. + """ + identity_pattern = ( + r"self body=([1-9][0-9]*) image=([1-9][0-9]*) " + r"base_hash=([0-9A-Fa-f]{16}) \| maxblk:(1024|2048)" + r"(?: abi=([0-9]+) codecs=0x([0-9A-Fa-f]+))? \| " + ) + before = re.fullmatch( + identity_pattern + r"storage unsafe: not probed", self_status.strip(), + ) + if before is None: + return status, self_status, None + + def status_identity(reply: str) -> tuple[str, str, int]: + hardware = re.findall(r"\bhw=([^ |]+)", reply) + targets = re.findall(r"\btarget:([0-9A-Fa-f]{8})\b", reply) + limits = re.findall(r"\bmaxblk:(\d+)\b", reply) + backends = re.findall(r"\bbl:([^ |]+)", reply) + if ( + not reply.startswith("OTA | ") + or len(hardware) != 1 or hardware[0] not in ("RAK_3401", "RAK4631") + or len(targets) != 1 or len(limits) != 1 or len(backends) != 1 + or reply.split(" | ").count("no download") != 1 + or "download:" in reply + ): + raise OtaError("cold RAK storage preparation requires a known, idle destination") + return hardware[0], targets[0].upper(), parse_target_max_block_size(reply, self_status) + + bound_status = status_identity(status) + if re.findall(r"\bbl:([^ |]+)", status)[0] not in ("NONE", "storage-ERR"): + raise OtaError("cold RAK destination reports conflicting unprobed storage diagnostics") + boot_status = controller.remote_command(target_name, "ota bootloader status", retry=False) + live = parse_bootloader_status(boot_status) + validate_adaptive_bootloader_status(live, bound_status[0]) + if not boot_status.strip().endswith("| staged:none mid=- hash=-"): + raise OtaError("cold RAK storage preparation requires no staged bootloader update") + prepared_status = controller.remote_command(target_name, "ota status", retry=False) + prepared_self = controller.remote_command(target_name, "ota self", retry=False) + after = re.fullmatch(identity_pattern + r"(.+)", prepared_self.strip()) + if ( + after is None + or before.groups() != after.groups()[:6] + or bound_status != status_identity(prepared_status) + ): + raise OtaError("destination firmware identity or geometry changed during cold RAK storage preparation") + if before.group(5) is not None and ( + int(before.group(5)) != live.apply_abi or int(before.group(6), 16) & 0x5 != 0x5 + ): + raise OtaError("cold RAK destination reports inconsistent or unsupported application capabilities") + tail = after.group(7) + if re.fullmatch(r"internal store:[1-9][0-9]*K \| bootloader: delta apply OK", tail): + backend = "internal" + elif re.fullmatch(r"QSPI store:[1-9][0-9]*K \| bootloader: QSPI apply OK", tail): + backend = "QSPI" + elif bound_status[0] == "RAK4631" and tail == "SD store | bootloader: authenticated SD apply OK": + backend = "SD" + else: + raise OtaError("cold RAK application storage remains unprepared, unsafe or unsupported") + if re.findall(r"\bbl:([^ |]+)", prepared_status) != [backend]: + raise OtaError("cold RAK destination reports conflicting prepared storage diagnostics") + print("[storage] prepared idle RAK application storage; running firmware identity unchanged") + return prepared_status, prepared_self, live + + def query_adaptive_bootloader_caps( controller: Controller, target_name: str, status: str, self_status: str, - hw_id: str | None, + hw_id: str | None, *, verified_bootloader: BootloaderStatus | None = None, ) -> tuple[int, int] | None: """Recover only capabilities guaranteed by known adaptive firmware. @@ -2896,9 +2991,6 @@ def query_adaptive_bootloader_caps( self_status.strip(), ): storage = "SD" - name = "3401" if hw_id == "RAK_3401" else "4631" - expected_board = 0x239A0029 - identities = {f"{name}_DFU": 0x0A, f"{name}_AUTO_DFU": 0x1E} elif hw_id == "Heltec_tower_v2": shape = re.fullmatch( prefix + r"store:(SD|internal) \| bootloader:apply OK", @@ -2906,8 +2998,6 @@ def query_adaptive_bootloader_caps( ) if shape: storage = shape.group(1) - expected_board = 0x239A0071 - identities = {"TOWER_V2_OTA": 0x09} else: return None if storage is None: @@ -2918,16 +3008,10 @@ def query_adaptive_bootloader_caps( or re.findall(r"\bbl:([^ |]+)", status) != [storage] ): raise OtaError("adaptive destination reports conflicting hardware/storage diagnostics") - live = parse_bootloader_status( + live = verified_bootloader or parse_bootloader_status( controller.remote_command(target_name, "ota bootloader status", retry=False) ) - if ( - live.board_id != expected_board - or identities.get(live.device_name) != live.storage_flags - or not 3 <= live.apply_abi < 0xFFFF - or live.crc32 in (0, 0xFFFFFFFF) - ): - raise OtaError("adaptive destination bootloader identity, ABI or storage profile is unverified") + validate_adaptive_bootloader_status(live, hw_id) print( "[bootloader] adaptive diagnostics verified through installed " f"bootloader status: ABI {live.apply_abi}, guaranteed codec subset 0x5 " @@ -2962,6 +3046,11 @@ def query_target( except OtaError as exc: print(f"[warn] optional `get bootloader.ver` probe unavailable: {exc}") self_status = controller.remote_command(args.target, "ota self") + verified_bootloader = None + if getattr(args, "package_kind", "application") == "application": + status, self_status, verified_bootloader = prepare_cold_rak_application_storage( + controller, args.target, status, self_status, + ) hash_match = re.search(r"base_hash=([0-9A-Fa-f]{16})", self_status) if not hash_match: raise OtaError("could not read destination base hash from `ota self`") @@ -3034,6 +3123,7 @@ def query_target( try: adaptive_caps = query_adaptive_bootloader_caps( controller, args.target, status, self_status, hw_id, + verified_bootloader=verified_bootloader, ) except OtaError as exc: raise OtaError( @@ -3048,6 +3138,12 @@ def query_target( "to expose complete bootloader diagnostics before retrying" ) bootloader_abi, bootloader_codecs = adaptive_caps + if ( + platform == "nrf52" + and getattr(args, "package_kind", "application") == "application" + and "storage unsafe:" in self_status + ): + raise OtaError("destination reports unsafe application OTA storage") current_version = None current_version_source = None try: diff --git a/tools/lora_ota/test_lora_ota_cold_storage.py b/tools/lora_ota/test_lora_ota_cold_storage.py new file mode 100644 index 000000000..a99b383a1 --- /dev/null +++ b/tools/lora_ota/test_lora_ota_cold_storage.py @@ -0,0 +1,253 @@ +"""Deferred RAK application storage is prepared once, with identity binding.""" + +import argparse +import contextlib +import io +import unittest +from unittest import mock + +import lora_ota as ota + + +class ColdRakStorageTests(unittest.TestCase): + def fixture(self, *, hw="RAK4631", storage="QSPI", direct=False, + auto=True, initial_bl="storage-ERR", changes=None): + prefix = "self body=100 image=156 base_hash=0011223344556677 | maxblk:2048" + if direct: + prefix += " abi=3 codecs=0x5" + prefix += " | " + tail = { + "internal": "internal store:252K | bootloader: delta apply OK", + "QSPI": "QSPI store:2048K | bootloader: QSPI apply OK", + "SD": "SD store | bootloader: authenticated SD apply OK", + }[storage] + header = ( + f"OTA | this fw 00112233 (0K) hw={hw} | target:1234ABCD | " + "maxblk:2048 | bl:" + ) + name = ("3401" if hw == "RAK_3401" else "4631") + name += "_AUTO_DFU" if auto else "_DFU" + values = { + "before_status": header + initial_bl + " blrc:00 | no download", + "before_self": prefix + "storage unsafe: not probed", + "boot_status": self.boot_reply(name=name, caps=0x1E if auto else 0x0A), + "after_status": header + storage + " blrc:00 | no download", + "after_self": prefix + tail, + } + for key, value in (changes or {}).items(): + values[key] = value(values[key]) if callable(value) else value + status_replies = iter((values["before_status"], values["after_status"])) + self_replies = iter((values["before_self"], values["after_self"])) + controller = mock.Mock() + + def remote(_target, command, **kwargs): + if command == "ota status": + value = next(status_replies) + elif command == "ota self": + value = next(self_replies) + elif command == "ota bootloader status": + self.assertIs(kwargs.get("retry"), False) + value = values["boot_status"] + elif command == "get bootloader.ver": + value = "> OTAFIX2.4.11-preview.126" + elif command == "ota stats": + value = "OTA | fw v1.18.0.1" + else: + raise AssertionError(f"unexpected or mutating command: {command}") + if isinstance(value, Exception): + raise value + return value + + controller.remote_command.side_effect = remote + return controller + + @staticmethod + def boot_reply(board=0x239A0029, name="4631_AUTO_DFU", caps=0x1E, + abi=3, crc=0x12345678): + target = ota.bootloader_library().bootloader_target_id(board, name) + return ( + f"BL board={board:08X} target={target:08X} name={name} " + f"crc={crc:08X} abi={abi} caps={caps:02X} | staged:none mid=- hash=-" + ) + + def query(self, controller, kind="application"): + with contextlib.redirect_stdout(io.StringIO()) as output: + result = ota.query_target( + controller, argparse.Namespace(target="remote", package_kind=kind), + ) + return result, output.getvalue() + + @staticmethod + def commands(controller): + return [call.args[1] for call in controller.remote_command.call_args_list] + + def test_old_and_complete_formats_prepare_all_known_rak_backends_once(self): + for direct in (False, True): + for hw in ("RAK_3401", "RAK4631"): + for storage in ("internal", "QSPI", "SD"): + if hw == "RAK_3401" and storage == "SD": + continue + for auto in (False, True): + for initial_bl in ("NONE", "storage-ERR"): + with self.subTest(direct=direct, hw=hw, storage=storage, + auto=auto, initial_bl=initial_bl): + controller = self.fixture(hw=hw, storage=storage, direct=direct, + auto=auto, initial_bl=initial_bl) + target, log = self.query(controller) + self.assertEqual(target.base_hash, bytes.fromhex("0011223344556677")) + self.assertEqual((target.hw_id, target.target_id), (hw, 0x1234ABCD)) + self.assertEqual(target.max_block_size, 2048) + self.assertEqual((target.bootloader_abi, target.bootloader_codecs), (3, 5)) + self.assertEqual(target.nrf_sd, storage == "SD") + self.assertEqual(target.nrf_qspi, storage == "QSPI") + self.assertIsNone(target.boot_target_id) + self.assertIn("running firmware identity unchanged", log) + self.assertEqual(self.commands(controller), [ + "ota status", "get bootloader.ver", "ota self", + "ota bootloader status", "ota status", "ota self", "ota stats", + ]) + for call in controller.remote_command.call_args_list[3:6]: + self.assertIs(call.kwargs.get("retry"), False) + + def test_running_identity_and_geometry_are_bound_across_preparation(self): + mutations = ( + ("after_self", "body=100", "body=101"), + ("after_self", "image=156", "image=157"), + ("after_self", "base_hash=0011223344556677", "base_hash=1011223344556677"), + ("after_self", "maxblk:2048", "maxblk:1024"), + ("after_status", "target:1234ABCD", "target:1234ABCE"), + ("after_status", "hw=RAK4631", "hw=RAK_3401"), + ("after_status", "maxblk:2048", "maxblk:1024"), + ) + for direct in (False, True): + for field, old, new in mutations: + with self.subTest(direct=direct, field=field, new=new): + controller = self.fixture(direct=direct, changes={ + field: lambda value, old=old, new=new: value.replace(old, new), + }) + with self.assertRaises(ota.OtaError): + self.query(controller) + self.assertEqual(self.commands(controller).count("ota bootloader status"), 1) + self.assertNotIn("ota stats", self.commands(controller)) + + def test_application_capabilities_cannot_change_or_conflict_with_live_abi(self): + for field, old, new in ( + ("after_self", "abi=3", "abi=4"), + ("after_self", "codecs=0x5", "codecs=0xD"), + ("before_self", "abi=3", "abi=2"), + ("before_self", "codecs=0x5", "codecs=0x1"), + ): + with self.subTest(field=field, new=new): + changes = {field: lambda value, old=old, new=new: value.replace(old, new)} + if field == "before_self": + changes["after_self"] = changes[field] + with self.assertRaises(ota.OtaError): + self.query(self.fixture(direct=True, changes=changes)) + + def test_initial_active_or_ambiguous_status_never_prepares(self): + mutations = ( + lambda value: value.replace("no download", "download: downloading 1/9 id=12345678"), + lambda value: value.replace("no download", "download: ready to install 9/9 id=12345678"), + lambda value: value.replace("no download", "download: failed (hash) 1/9 id=12345678"), + lambda value: value.replace("no download", "no download extra"), + lambda value: value + " | no download", + lambda value: value.replace("hw=RAK4631", "hw=RAK4631 hw=RAK4631"), + lambda value: value + " | target:1234ABCD", + lambda value: value + " | maxblk:2048", + lambda value: value + " | bl:storage-ERR", + lambda value: value.replace("bl:storage-ERR", "bl:QSPI"), + lambda value: value.replace("hw=RAK4631", "hw=OtherBoard"), + lambda value: value.replace("maxblk:2048", "maxblk:1024"), + ) + for mutation in mutations: + with self.subTest(mutation=mutation): + controller = self.fixture(changes={"before_status": mutation}) + with self.assertRaises(ota.OtaError): + self.query(controller) + self.assertNotIn("ota bootloader status", self.commands(controller)) + + def test_post_preparation_must_still_be_idle_and_storage_views_agree(self): + for mutation in ( + lambda value: value.replace("no download", "download: downloading 1/9 id=12345678"), + lambda value: value.replace("bl:QSPI", "bl:internal"), + lambda value: value.replace("bl:QSPI", "bl:NONE"), + ): + with self.subTest(mutation=mutation): + with self.assertRaises(ota.OtaError): + self.query(self.fixture(changes={"after_status": mutation})) + + def test_unsafe_unsupported_unprepared_and_zero_capacity_results_are_refused(self): + tails = ( + "storage unsafe: not probed", "storage unsafe: external NOR and OTAFIX bootloader do not match", + "QSPI store:0K | bootloader: QSPI apply OK", + "QSPI store:ERR 0K | bootloader: QSPI apply OK", + "QSPI store:2048K | bootloader: NO QSPI apply", + "internal store:0K | bootloader: delta apply OK", + "internal store:252K | bootloader: NO delta apply", + "SD store | bootloader: NO SD apply", + "QSPI store:2048K | bootloader: QSPI apply OK extra", + ) + for direct in (False, True): + for tail in tails: + with self.subTest(direct=direct, tail=tail): + with self.assertRaises(ota.OtaError): + self.query(self.fixture(direct=direct, changes={ + "after_self": lambda value, tail=tail: value.rsplit(" | ", 2)[0] + " | " + tail, + })) + with self.assertRaises(ota.OtaError): + self.query(self.fixture(hw="RAK_3401", storage="SD")) + + def test_only_exact_installed_family_profile_abi_crc_and_idle_staging_qualify(self): + valid = self.boot_reply() + invalid = ( + self.boot_reply(board=0x239A0071, name="TOWER_V2_OTA", caps=9), + self.boot_reply(name="3401_AUTO_DFU"), + self.boot_reply(name="OTHER_DFU"), + self.boot_reply(abi=2), self.boot_reply(abi=65535), + self.boot_reply(caps=0x0A), self.boot_reply(caps=0xFF), + self.boot_reply(crc=0), self.boot_reply(crc=0xFFFFFFFF), + valid.replace("target=", "target=0"), + valid.replace("staged:none mid=- hash=-", "staged:ready mid=12345678 hash=0011223344556677"), + valid + " extra", "Unknown command", + ota.TransmissionError("no reply"), + ) + for reply in invalid: + with self.subTest(reply=reply): + controller = self.fixture(changes={"boot_status": reply}) + with self.assertRaises(ota.OtaError): + self.query(controller) + self.assertEqual(self.commands(controller), [ + "ota status", "get bootloader.ver", "ota self", "ota bootloader status", + ]) + + def test_requery_failure_is_bounded_without_fallback_retry(self): + for field in ("after_status", "after_self"): + with self.subTest(field=field): + controller = self.fixture(changes={field: ota.TransmissionError("missing reply")}) + with self.assertRaises(ota.OtaError): + self.query(controller) + self.assertEqual(self.commands(controller).count("ota bootloader status"), 1) + self.assertNotIn("ota stats", self.commands(controller)) + + def test_bootloader_packages_never_enter_application_preparation(self): + for direct in (False, True): + with self.subTest(direct=direct): + controller = self.fixture(direct=direct, initial_bl="NONE") + with self.assertRaises(ota.OtaError): + self.query(controller, kind="bootloader") + self.assertNotIn("ota bootloader status", self.commands(controller)) + + def test_unrecognized_unsafe_shape_does_not_trigger_preparation_or_bypass_safety(self): + for direct in (False, True): + for tail in ("not probed extra", "unavailable"): + with self.subTest(direct=direct, tail=tail): + controller = self.fixture(direct=direct, changes={ + "before_self": lambda value, tail=tail: value.replace("not probed", tail), + }) + with self.assertRaises(ota.OtaError): + self.query(controller) + self.assertNotIn("ota bootloader status", self.commands(controller)) + + +if __name__ == "__main__": + unittest.main()