Prepare idle adaptive RAK storage before OTA preflight

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

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:

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()