From bb7ebaa1d9114cb024ad30faf09010c196d540aa Mon Sep 17 00:00:00 2001 From: mikecarper Date: Mon, 26 Oct 2026 21:36:06 -0700 Subject: [PATCH] Finalize interrupted OTA downloads before restoring automatic updates --- tools/lora_ota/ota_wizard_session.py | 226 +++++++++++++++-- tools/lora_ota/test_ota_wizard_session.py | 280 +++++++++++++++++++++- 2 files changed, 475 insertions(+), 31 deletions(-) diff --git a/tools/lora_ota/ota_wizard_session.py b/tools/lora_ota/ota_wizard_session.py index 95f570e67..65bf37001 100644 --- a/tools/lora_ota/ota_wizard_session.py +++ b/tools/lora_ota/ota_wizard_session.py @@ -5,6 +5,7 @@ import asyncio from dataclasses import dataclass import hashlib import json +import math from pathlib import Path import re import time @@ -143,7 +144,7 @@ class DirectSession: """One target, zero-hop routing, bounded operations, and durable recovery state.""" def __init__(self, transport, public_key: str, password: str, journal: Path, - *, progress: Callable[[str], None] = print, poll_seconds: float = 5, + *, progress: Callable[[str], None] = print, poll_seconds: float = 10, transfer_timeout: float = 4 * 60 * 60, reboot_timeout: float = 300): if not re.fullmatch(r"[0-9a-fA-F]{64}", public_key): raise ota.OtaError("Choose a destination with a complete public key.") @@ -151,6 +152,14 @@ class DirectSession: self.journal, self.progress = Path(journal), progress self.poll_seconds, self.transfer_timeout = poll_seconds, transfer_timeout self.reboot_timeout = reboot_timeout + # A run can install a bootloader and then an application. Keep one + # source lease valid for both complete transfer/reboot budgets, with + # setup and recovery slack, rather than granting each package four + # hours against a source window that started only once. + self.window_minutes = max(15, math.ceil( + (2 * transfer_timeout + 2 * reboot_timeout + 15 * 60) / 60)) + if self.window_minutes > 10080: + raise ota.OtaError("The combined update exceeds the supported temporary radio window.") self.saved: dict[str, dict[str, str]] = {} self.active = False self._serve_stop = None @@ -236,6 +245,8 @@ class DirectSession: self._state["source_key"] = source_key.lower() for label, command in (("source", self.local), ("target", self.remote)): await self._check_temporary_windows(command, label) + if label == "target": + await self._require_idle_download() config = await command("ota config") radio = await command("get radio") try: @@ -286,6 +297,7 @@ class DirectSession: # the first mutation rather than replacing a newly queued lease. for label, command in (("source", self.local), ("target", self.remote)): await self._check_temporary_windows(command, label) + await self._require_idle_download() self.active = True self.save("opening OTA windows") for label, command in (("target", self.remote), ("source", self.local)): @@ -296,17 +308,31 @@ class DirectSession: commands.append("set radio.rxps off") for text in commands: await command(text) + await self._verify_paused_policy(command, label) if self.saved[label]["rxps"] != "unsupported": current = await self._read_rxps(command, label) if current.enabled: raise ota.OtaError(f"The {label} did not disable RX power saving.") # Keep the existing RF tuple so both admin and file traffic stay # reachable even if the process is interrupted before cleanup. - values = radio_tuple(self.saved["target"]["radio"]) - await command("tempradio " + ",".join(map(str, values)) + ",240") - await self._wait_window(command) + await self._open_window(label) self.save("OTA windows active") + async def _open_window(self, label: str) -> None: + command = self.remote if label == "target" else self.local + values = radio_tuple(self.saved[label]["radio"]) + text = "tempradio " + ",".join(map(str, values)) + f",{self.window_minutes}" + try: + if label == "target": + await self.remote(text, retries=0) + else: + await self.local(text) + except (TimeoutError, ConnectionError, ota.TransmissionError): + # Relative leases must not be replayed after a lost acknowledgement. + # Both entry points have already proved the old window inactive. + pass + await self._wait_window(command) + async def _stop_serving(self) -> None: if self._serve_task: self._serve_stop.set() @@ -344,7 +370,19 @@ class DirectSession: raise ota.OtaError("The transfer timed out. The recovery record retains the package ID.") async def _download_state(self, info: ota.MotaInfo) -> tuple[str, bool]: + state, mid, status = await self._read_download() + if mid is None: + raise ota.OtaError("The destination did not confirm a live download and manifest ID.") + if mid != info.manifest_id.lower(): + raise ota.OtaError("The destination's download is not the selected package.") + if state == "failed": + raise ota.OtaError("The destination reported a failed download: " + status) + return status, state == "complete" + + async def _read_download(self) -> tuple[str, str | None, str]: status = await self.remote("ota status") + if re.search(r"\bno download\b", status, re.I): + return "idle", None, status download = re.search(r"\bdownload:\s*([^|]+)", status, re.I) section = download[1] if download else "" mid = re.search(r"\bid=([0-9A-Fa-f]{8})\b", section) @@ -354,16 +392,22 @@ class DirectSession: stats = await self.remote("ota stats") fetch = re.search(r"\bfetch\s+([^|]+)", stats, re.I) section = fetch[1] if fetch else "" + if re.match(r"idle\b", section, re.I): + return "idle", None, status + " | " + stats mid = re.search(r"\bid=([0-9A-Fa-f]{8})\b", section) status += " | " + stats if mid is None: raise ota.OtaError("The destination did not confirm a live download and manifest ID.") - if mid[1].lower() != info.manifest_id.lower(): - raise ota.OtaError("The destination's download is not the selected package.") - if re.match(r"failed\b", section, re.I): - raise ota.OtaError("The destination reported a failed download: " + status) - complete = bool(re.match(r"(?:ready to install|done)\b", section, re.I)) - return status, complete + state = ("failed" if re.match(r"failed\b", section, re.I) else + "complete" if re.match(r"(?:ready to install|done)\b", section, re.I) else "active") + return state, mid[1].lower(), status + + async def _require_idle_download(self) -> None: + state, mid, _ = await self._read_download() + if state != "idle": + raise ota.OtaError( + f"The destination already has OTA download {mid or 'unknown'}. " + "Finish or cancel that update before starting another one.") async def _installed(self, package: Package, before: Destination) -> Destination: info, deadline = package.info, time.monotonic() + self.reboot_timeout @@ -399,9 +443,17 @@ class DirectSession: await self.begin() # A bootloader reboot clears the destination's RAM-only lease. if temp_state(await self.remote("get tempradio")) == "off": - values = radio_tuple(self.saved["target"]["radio"]) - await self.remote("tempradio " + ",".join(map(str, values)) + ",240") - await self._wait_window(self.remote) + await self._open_window("target") + await self._require_idle_download() + self.save("preparing package", package_mid=info.manifest_id, package_hash=info.image_hash.hex(), + package_kind=info.kind, package_bootloader=info.is_bootloader, + package_version=info.fw_version, package_target=info.target_id, package_hw=info.hw_id, + package_storage=info.bootloader_storage, + expected_body=package.body_hash.hex() if package.body_hash else None, + previous_app=dict(target=before.target_id, hw=before.hw_id, + version=before.version, body=before.base_hash.hex()), + pull_attempted=False, install_attempted=False, + package_finalized=False, cancel_sent=False, cancel_confirmed=False) if package.signer: await self.remote("ota key add " + package.signer) if not info.is_bootloader and not info.is_full: @@ -409,8 +461,7 @@ class DirectSession: if live.base_hash != info.base_hash: raise ota.OtaError("The destination's firmware changed after delta preparation.") await self._start_serving(package) - self.save("downloading", package_mid=info.manifest_id, package_hash=info.image_hash.hex(), - package_kind=info.kind, expected_body=package.body_hash.hex() if package.body_hash else None) + self.save("serving package") # Discovery may need more than one advertisement. It is read-only and # always bound to the exact selected MID before a pull is issued. for _ in range(12): @@ -419,10 +470,17 @@ class DirectSession: await asyncio.sleep(self.poll_seconds) else: raise ota.OtaError("The selected package was not discovered over the direct link.") + await self._require_idle_download() + self.save("downloading", pull_attempted=True) try: await self.remote(f"ota pull {info.manifest_id} flash", retries=0) except ota.TransmissionError: await self._download_state(info) + except ota.OtaError: + # An explicit rejection did not grant us ownership of somebody + # else's slot; cleanup must not cancel that download. + self.save("pull rejected", pull_attempted=False) + raise await self._ready(info) if info.is_bootloader: staged = await self.remote("ota bootloader status") @@ -432,22 +490,144 @@ class DirectSession: command = f"ota bootloader install {info.manifest_id} {info.image_hash[:8].hex().upper()}" else: command = "ota install" - self.save("installing") + self.save("installing", install_attempted=True) try: await self.remote(command, retries=0) except ota.TransmissionError: self.progress("The install reply was lost; checking the running image before taking further action.") current = await self._installed(package, before) await self._stop_serving() - self.save("installed") + self.save("installed", package_finalized=True, install_verified=True) return current + async def _verify_paused_policy(self, command, label: str) -> None: + policy = await command("ota config") + fields = ("hops",) if "mode=seeder-only" in policy else ("hops", "autofetch", "autoinstall") + for field in fields: + match = re.search(rf"\b{field}=([A-Za-z0-9]+)", policy) + expected = "0" if field == "hops" else "off" + if match is None or match[1] != expected: + raise ota.OtaError(f"The {label} did not confirm isolated OTA policy {field}={expected}.") + + def _expected_install_matches(self, current: Destination) -> bool: + state = self._state + if state.get("package_bootloader"): + previous = state.get("previous_app", {}) + boot = current.bootloader + return (current.boot_version == state.get("package_version") and boot is not None + and boot.target_id == state.get("package_target") + and boot.hw_id == state.get("package_hw") + and boot.storage_flags == state.get("package_storage") + and current.target_id == previous.get("target") + and current.hw_id == previous.get("hw") + and current.version == previous.get("version") + and current.base_hash.hex() == previous.get("body")) + return (current.target_id == state.get("package_target") + and current.hw_id == state.get("package_hw") + and current.version == state.get("package_version") + and current.base_hash.hex() == state.get("expected_body")) + + async def _finalize_download(self) -> bool: + """Free only this run's receive slot, before restoring automatic policy.""" + state = self._state + if state.get("package_finalized") is True: + return True + if type(state.get("pull_attempted")) is not bool: + # Old journals recorded only a phase. Normalize it durably before + # cleanup can replace that phase with "needs cleanup"; otherwise + # the next recovery could forget an unresolved pull/install. + phase = state.get("phase") + owned = phase in ("downloading", "installing", "installed") or ( + bool(state.get("package_mid")) and phase not in ( + "inspecting", "prepared", "opening OTA windows", "OTA windows active", + "preparing package", "serving package", "pull rejected")) + self.save(phase or "needs cleanup", pull_attempted=owned, + install_attempted=owned and phase != "downloading", + cancel_sent=(state.get("cancel_sent") is True or + "cancel_sent" in state and type(state["cancel_sent"]) is not bool), + cancel_confirmed=state.get("cancel_confirmed") is True, + package_finalized=False) + attempted = state.get("pull_attempted") is True + if not attempted: + return True + if type(state.get("install_attempted")) is not bool: + # A partial or malformed journal must not turn a possibly armed + # install into a cancelable download when cleanup changes phase. + self.save(state.get("phase", "needs cleanup"), install_attempted=True) + if type(state.get("cancel_sent")) is not bool: + self.save(state.get("phase", "needs cleanup"), cancel_sent=True) + if type(state.get("cancel_confirmed")) is not bool: + self.save(state.get("phase", "needs cleanup"), cancel_confirmed=False) + mid = state.get("package_mid") + reset_session = getattr(self.link, "reset_remote_session", None) + if reset_session: + await reset_session(self.key) + # A canceled COMPLETE signed fetch must never become eligible merely + # because we restored its saved autoinstall=trusted setting. + await self.remote("ota config autoinstall off") + await self.remote("ota config autofetch off") + policy = await self.remote("ota config") + if not all(re.search(rf"\b{field}=off\b", policy) for field in ("autoinstall", "autofetch")): + raise ota.OtaError("The destination did not confirm automatic OTA policies are off.") + if not isinstance(mid, str) or not re.fullmatch(r"[0-9a-fA-F]{8}", mid): + raise ota.OtaError("The unfinished update has no valid package ID; automatic OTA policies remain off.") + if state.get("install_attempted") is True or state.get("phase") == "installing": + # An install may already be armed even if its acknowledgement was + # lost. Never erase that slot; first prove the resulting image. + current = await self.inspect() + if self._expected_install_matches(current): + self.save("installed", package_finalized=True, install_verified=True) + return True + raise ota.OtaError("Installation outcome is unresolved; automatic OTA policies remain off.") + live_state, live_mid, _ = await self._read_download() + if live_state == "idle": + if state.get("cancel_confirmed") is True: + return True + # IDLE alone says nothing about persistent media invalidation. + # Do not replay cancel against an unowned/unknown staged slot. + raise ota.OtaError("The live download is idle, but persistent slot clearance is unconfirmed; automatic OTA policies remain off.") + if live_mid != mid.lower(): + raise ota.OtaError("The destination now has a different download; it was not canceled and automatic OTA policies remain off.") + if state.get("cancel_sent") is True: + raise ota.OtaError("The previous cancellation is unconfirmed; it was not replayed and automatic OTA policies remain off.") + self.save(state.get("phase", "downloading"), cancel_sent=True) + reply = None + try: + reply = await self.remote("ota cancel", retries=0) + except (ota.OtaError, TimeoutError, ConnectionError): + pass # Read back once; never blindly repeat a destructive command. + live_state, _, _ = await self._read_download() + confirmed = (reply is not None and live_state == "idle" + and re.match(r"(?i)^OK dropped session\b", reply) + and re.search(rf"(?i)\bmid={re.escape(mid)}\b", reply) + and "OTA receive slot confirmed clear" in reply) + if not confirmed: + raise ota.OtaError("Cancellation or persistent slot clearance is unconfirmed; automatic OTA policies remain off.") + self.save(state.get("phase", "downloading"), cancel_confirmed=True, package_finalized=True) + return True + async def cleanup(self) -> list[str]: errors = [] + server_stopped = True try: await self._stop_serving() except Exception: + server_stopped = False errors.append("Stop the USB file server before restoring settings.") + target_policy_safe = True + if self.active: + try: + target_policy_safe = await self._finalize_download() + except Exception as exc: + target_policy_safe = False + errors.append(f"Resolve the recorded target update before restoring automatic OTA policy: {exc}") + if not server_stopped: + target_policy_safe = False + try: + await self.remote("ota config autoinstall off") + await self.remote("ota config autofetch off") + except Exception: + errors.append("Confirm both target automatic OTA policies are off while its source remains unconfirmed.") if self.active: for label, command in (("target", self.remote), ("source", self.local)): if label not in self.saved: @@ -465,6 +645,8 @@ class DirectSession: except Exception: errors.append(f"Restore the recorded {label} RXPS settings from {self.journal}.") for key in ("hops", "autofetch", "autoinstall"): + if label == "target" and key != "hops" and not target_policy_safe: + continue if key != "hops" and "mode=seeder-only" in settings["ota"]: continue match = re.search(rf"\b{key}=([A-Za-z0-9]+)", settings["ota"]) @@ -476,7 +658,8 @@ class DirectSession: except Exception: errors.append(f"{label}: {text}") try: - await self._verify_restored(command, label, settings) + await self._verify_restored(command, label, settings, + restore_auto=label != "target" or target_policy_safe) except Exception: errors.append(f"Verify and restore the recorded {label} radio/RXPS/OTA settings from {self.journal}.") restore_path = getattr(self.link, "restore_contact_path", None) @@ -498,7 +681,8 @@ class DirectSession: reply = await command("get radio.rxps") return ota.parse_rxps_settings(reply, label) - async def _verify_restored(self, command, label: str, settings: dict) -> None: + async def _verify_restored(self, command, label: str, settings: dict, + *, restore_auto: bool = True) -> None: deadline = time.monotonic() + 45 while temp_state(await command("get tempradio")) != "off": if time.monotonic() >= deadline: @@ -512,6 +696,10 @@ class DirectSession: raise ota.OtaError(f"The {label} RXPS settings did not restore exactly.") actual = await command("ota config") for key in ("hops", "autofetch", "autoinstall"): + if not restore_auto and key != "hops": + if not re.search(rf"\b{key}=off\b", actual): + raise ota.OtaError(f"The {label} unsafe automatic OTA policy was not kept off.") + continue previous = re.search(rf"\b{key}=([A-Za-z0-9]+)", settings["ota"]) current = re.search(rf"\b{key}=([A-Za-z0-9]+)", actual) if previous is None or current is None or previous[1] != current[1]: diff --git a/tools/lora_ota/test_ota_wizard_session.py b/tools/lora_ota/test_ota_wizard_session.py index 08fae3e7a..80bc1f1b3 100644 --- a/tools/lora_ota/test_ota_wizard_session.py +++ b/tools/lora_ota/test_ota_wizard_session.py @@ -48,12 +48,23 @@ class FakeTransport: self.rxps = {"source": "off,level=8,preamble=16,rx=10000,sleep=10000", "target": "on,level=8,preamble=16,rx=10000,sleep=10000"} self.policies = {"source": dict(hops="3", autofetch="off", autoinstall="off"), - "target": dict(hops="2", autofetch="on", autoinstall="on")} + "target": dict(hops="2", autofetch="signed", autoinstall="trusted")} self.seeder = True self.unsupported_source = False self.serving = None self.serves = [] self.mid = None + self.slot_mid = None + self.trusted_keys = set() + self.signer = None + self.auto_installed = False + self.apply_pending = False + self.defer_install = False + self.lost_cancel = self.discard_fails = False + self.lease_minutes = {} + self.lease_until = {} + self.now = 0 + self.lost_window_ack = set() self.lost_pull = self.lost_install = False self.bad_mid = self.bad_hash = self.bad_body = False self.truncate_status = False @@ -80,6 +91,8 @@ class FakeTransport: async def serve(self, blob, stop): self.serving = ota.parse_mota(blob) + manifest = ota.bootloader_library().parse_container(blob).manifest + self.signer = manifest.signer_pubkey.hex().upper() if manifest.is_signed else None self.serves.append(self.serving) await stop.wait() @@ -123,6 +136,8 @@ class FakeTransport: return "ERR seeder-only policy must stay off" if not self.ignore_restore or value == "0": self.policies[label][field] = value + if label == "target": + self.auto_install_tick() return "OK" if text == "get radio": return "910.525024,62.5,7,5" @@ -137,6 +152,8 @@ class FakeTransport: self.rxps[label] = "on,level=8,preamble=16,rx=10000,sleep=10000" return "OK on" if text == "get tempradio": + if self.now >= self.lease_until.get(label, float("inf")): + self.windows[label] = False if label == "source": return "TempRadio active: 910.525024,62.5,7,5 14000s left,preamble=16" if self.windows[label] else "TempRadio inactive" return "910.525024,62.5,7,5,240" if self.windows[label] else "off" @@ -144,11 +161,17 @@ class FakeTransport: return self.temp_schedules[label] if text.startswith("tempradio "): self.windows[label] = True + self.lease_minutes[label] = int(text.rsplit(",", 1)[1]) + self.lease_until[label] = self.now + self.lease_minutes[label] * 60 + if label in self.lost_window_ack: + self.lost_window_ack.remove(label) + raise TimeoutError("lost lease acknowledgment") return "OK applying in 2s" if text == "normalradio": self.windows[label] = False return "OK" if text.startswith("ota key add "): + self.trusted_keys.add(text.split()[-1]) return "OK trusted key added" if text == "ota ls": return f"OTA | {self.serving.manifest_id}" @@ -156,25 +179,57 @@ class FakeTransport: if not all(self.windows.values()): raise AssertionError("pull without overlapping leases") self.mid = text.split()[2] + self.slot_mid = self.mid self.download_started.set() if self.lost_pull: raise TimeoutError("lost pull acknowledgment") return "OK pulling" - if text.startswith("ota bootloader install ") or text == "ota install": - if self.serving.is_bootloader: - self.boot_version = self.serving.fw_version - else: - self.image = ota.parse_endf(self.serving.payload) - if self.bad_body: - self.image = application(self.image.fw_version, b"different body" * 600) - self.windows["target"] = False # reboot clears the RAM lease - self.login_stale = True + if text == "ota cancel": + # Firmware guards application apply_pending here, but not a + # separately armed bootloader installation. + if self.apply_pending and not self.serving.is_bootloader: + return "ERR update is armed; reboot is pending" + previous = self.mid self.mid = None + if not self.discard_fails: + self.slot_mid = None + if self.lost_cancel: + raise TimeoutError("lost cancel acknowledgment") + if self.discard_fails: + return f"ERR dropped live session (was C mid={previous}), but persistent OTA slot invalidation failed" + return f"OK dropped session (was C mid={previous}); OTA receive slot confirmed clear" + if text.startswith("ota bootloader install ") or text == "ota install": + self.apply_pending = True + if not self.defer_install: + self.finish_install() if self.lost_install: raise TimeoutError("lost install acknowledgment") return "OK rebooting" raise AssertionError((label, text)) + def finish_install(self): + if self.serving.is_bootloader: + self.boot_version = self.serving.fw_version + else: + self.image = ota.parse_endf(self.serving.payload) + if self.bad_body: + self.image = application(self.image.fw_version, b"different body" * 600) + self.windows["target"] = False + self.login_stale = True + self.mid = None + self.apply_pending = False + + def auto_install_tick(self): + # Match Mesh.cpp's COMPLETE/signed/non-bootloader/trusted-auto gate, + # including the apply path's trusted-key/newer-version requirement. + if (self.mid and not self.hold_download and not self.apply_pending + and self.policies["target"]["autoinstall"] == "trusted" + and self.serving is not None and self.serving.flags & 2 + and not self.serving.is_bootloader and self.signer in self.trusted_keys + and self.serving.fw_version > self.image.fw_version): + self.auto_installed = True + self.finish_install() + class SessionTests(unittest.IsolatedAsyncioTestCase): def setUp(self): @@ -214,7 +269,7 @@ class SessionTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(await self.session.cleanup(), []) self.assertEqual(self.link.rxps["source"].split(",")[0], "off") self.assertEqual(self.link.rxps["target"].split(",")[0], "on") - self.assertEqual(self.link.policies["target"], dict(hops="2", autofetch="on", autoinstall="on")) + self.assertEqual(self.link.policies["target"], dict(hops="2", autofetch="signed", autoinstall="trusted")) self.assertFalse(any(self.link.windows.values())) self.assertTrue(self.link.restore_path_called) journal = json.loads(self.path.read_text()) @@ -242,7 +297,9 @@ class SessionTests(unittest.IsolatedAsyncioTestCase): with self.assertRaisesRegex(ota.OtaError, "not the selected package"): await self.session.install(wizard.full_package(application(), 1024), before) finally: - self.assertEqual(await self.session.cleanup(), []) + self.assertTrue(await self.session.cleanup()) + self.assertNotIn(("target", "ota cancel"), self.link.calls) + self.assertEqual(self.link.policies["target"]["autoinstall"], "off") self.assertNotIn(("target", "ota install"), self.link.calls) async def test_staged_bootloader_hash_must_match_exactly(self): @@ -303,6 +360,205 @@ class SessionTests(unittest.IsolatedAsyncioTestCase): self.assertFalse(any(self.link.windows.values())) self.assertNotIn(("target", "ota install"), self.link.calls) + async def stage_and_cancel(self, *, signed=False, complete=False): + from test_ota_wizard_bundle import signed_package + package = signed_package(application()) if signed else wizard.full_package(application(), 1024) + before = await self.session.inspect() + self.link.hold_download = True + task = asyncio.create_task(self.session.install(package, before)) + await asyncio.wait_for(self.link.download_started.wait(), 2) + self.link.hold_download = not complete + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + return package + + async def test_cancel_complete_signed_fetch_clears_slot_before_restoring_trusted_policy(self): + await self.stage_and_cancel(signed=True, complete=True) + errors = await self.session.cleanup() + self.assertFalse(self.link.auto_installed) + self.assertEqual(errors, []) + self.assertEqual(self.link.image.fw_version, OLD) + self.assertIsNone(self.link.mid) + self.assertIsNone(self.link.slot_mid) + target = [text for label, text in self.link.calls if label == "target"] + self.assertEqual(target.count("ota cancel"), 1) + self.assertLess(target.index("ota cancel"), target.index("ota config autoinstall trusted")) + self.assertNotIn("ota install", target) + self.assertEqual(json.loads(self.path.read_text())["phase"], "finished") + # There is no abandoned live receive slot blocking the next update. + after = await self.session.install(wizard.full_package(application(), 1024), await self.session.inspect()) + self.assertEqual(after.version, NEW) + self.assertEqual(await self.session.cleanup(), []) + + async def test_lost_cancel_reply_never_replays_cancel_or_restores_automatic_policy(self): + await self.stage_and_cancel(signed=True, complete=True) + self.link.lost_cancel = True + self.assertTrue(await self.session.cleanup()) + self.assertTrue(await self.session.cleanup()) + self.assertEqual([text for _, text in self.link.calls].count("ota cancel"), 1) + self.assertEqual(self.link.policies["target"]["autoinstall"], "off") + self.assertEqual(self.link.policies["target"]["autofetch"], "off") + self.assertFalse(self.link.auto_installed) + self.assertEqual(json.loads(self.path.read_text())["phase"], "needs cleanup") + + async def test_legacy_download_journal_remembers_uncertain_cancel_after_restart(self): + await self.stage_and_cancel(signed=True, complete=True) + for name in ("pull_attempted", "install_attempted", "package_finalized", "cancel_sent", "cancel_confirmed"): + self.session._state.pop(name, None) + self.session.save("downloading") + self.link.lost_cancel = True + self.assertTrue(await self.session.cleanup()) + recovery = wizard.DirectSession(self.link, KEY, "secret-for-test", self.path, progress=lambda _: None) + recovery._state = json.loads(self.path.read_text()) + recovery.saved, recovery.active = recovery._state["settings"], True + self.assertTrue(recovery._state["pull_attempted"]) + self.assertTrue(recovery._state["cancel_sent"]) + self.assertTrue(await recovery.cleanup()) + self.assertEqual([text for _, text in self.link.calls].count("ota cancel"), 1) + self.assertEqual(self.link.policies["target"]["autoinstall"], "off") + self.assertEqual(self.link.policies["target"]["autofetch"], "off") + self.assertFalse(self.link.auto_installed) + + async def test_legacy_install_journal_never_becomes_cancelable_after_failed_recovery(self): + self.link.defer_install = True + before = await self.session.inspect() + with self.assertRaises(ota.OtaError): + await self.session.install(wizard.full_package(application(), 1024), before) + for name in ("pull_attempted", "install_attempted", "package_finalized", "cancel_sent", "cancel_confirmed"): + self.session._state.pop(name, None) + self.session.save("installing") + self.assertTrue(await self.session.cleanup()) + recovery = wizard.DirectSession(self.link, KEY, "secret-for-test", self.path, progress=lambda _: None) + recovery._state = json.loads(self.path.read_text()) + recovery.saved, recovery.active = recovery._state["settings"], True + self.assertTrue(recovery._state["install_attempted"]) + self.assertTrue(await recovery.cleanup()) + self.assertNotIn(("target", "ota cancel"), self.link.calls) + self.assertTrue(self.link.apply_pending) + self.assertEqual(self.link.policies["target"]["autoinstall"], "off") + + async def test_malformed_install_marker_never_cancels_armed_bootloader(self): + self.link.defer_install = True + before = await self.session.inspect() + package = wizard.Package(boot_blob(storage=14, version=BOOT_NEW), signer=TEST_SIGNER) + with self.assertRaises(ota.OtaError): + await self.session.install(package, before) + original = json.loads(json.dumps(self.session._state)) + for malformed in ("install marker", "legacy phase", "legacy missing ID"): + with self.subTest(malformed=malformed): + self.session._state = json.loads(json.dumps(original)) + self.session.active = True + if malformed == "install marker": + self.session.save("needs cleanup", install_attempted="false") + else: + for name in ("pull_attempted", "install_attempted", "package_finalized", + "cancel_sent", "cancel_confirmed"): + self.session._state.pop(name, None) + if malformed == "legacy missing ID": + self.session._state.pop("package_mid") + self.session.save(["installing"] if malformed == "legacy phase" else "installing") + self.assertTrue(await self.session.cleanup()) + recovery = wizard.DirectSession(self.link, KEY, "secret-for-test", self.path, progress=lambda _: None) + recovery._state = json.loads(self.path.read_text()) + recovery.saved, recovery.active = recovery._state["settings"], True + self.assertIs(recovery._state["install_attempted"], True) + self.assertTrue(await recovery.cleanup()) + self.assertNotIn(("target", "ota cancel"), self.link.calls) + self.assertTrue(self.link.apply_pending) + self.assertEqual(self.link.slot_mid, package.info.manifest_id) + self.assertEqual(self.link.policies["target"]["autoinstall"], "off") + self.assertEqual(self.link.policies["target"]["autofetch"], "off") + + async def test_failed_source_stop_cannot_reenable_automatic_fetch(self): + from unittest import mock + await self.stage_and_cancel(signed=True, complete=True) + original = self.session._stop_serving + with mock.patch.object(self.session, "_stop_serving", new_callable=mock.AsyncMock, + side_effect=ota.OtaError("source did not stop")): + self.assertTrue(await self.session.cleanup()) + self.assertEqual(self.link.policies["target"]["autoinstall"], "off") + self.assertEqual(self.link.policies["target"]["autofetch"], "off") + self.assertFalse(self.link.auto_installed) + await original() + self.assertEqual(await self.session.cleanup(), []) + + async def test_persistent_slot_invalidation_failure_keeps_both_automatic_policies_off(self): + package = await self.stage_and_cancel(signed=True, complete=True) + self.link.discard_fails = True + self.assertTrue(await self.session.cleanup()) + self.assertIsNone(self.link.mid) # IDLE does not prove durable erasure. + self.assertEqual(self.link.slot_mid, package.info.manifest_id) + self.assertEqual(self.link.policies["target"]["autoinstall"], "off") + self.assertEqual(self.link.policies["target"]["autofetch"], "off") + self.assertTrue(self.session.active) + self.assertFalse(self.link.auto_installed) + + async def test_foreign_existing_download_is_not_canceled_or_mutated(self): + self.link.mid = "DEADBEEF" + with self.assertRaisesRegex(ota.OtaError, "already has OTA download"): + await self.session.begin() + self.assertFalse(self.session.active) + self.assertFalse(any(text.startswith(("ota cancel", "ota config ", "set ", "tempradio ")) + for _, text in self.link.calls)) + self.assertEqual(self.link.mid, "DEADBEEF") + + async def test_download_started_during_preparation_is_not_replaced(self): + await self.session.snapshot() + self.link.mid = "DEADBEEF" + self.link.calls.clear() + with self.assertRaisesRegex(ota.OtaError, "already has OTA download"): + await self.session.begin() + self.assertFalse(self.session.active) + self.assertFalse(any(text.startswith(("ota cancel", "ota config ", "set ", "tempradio ")) + for _, text in self.link.calls)) + + async def test_uncertain_armed_install_is_not_erased_and_recovery_verifies_it(self): + self.link.defer_install = True + before = await self.session.inspect() + with self.assertRaisesRegex(ota.OtaError, "version and body hash"): + await self.session.install(wizard.full_package(application(), 1024), before) + self.assertTrue(self.link.apply_pending) + self.assertTrue(await self.session.cleanup()) + self.assertNotIn(("target", "ota cancel"), self.link.calls) + self.assertEqual(self.link.policies["target"]["autoinstall"], "off") + # A fresh process reads the saved exact expectations, never guesses + # that a timed-out install failed, and does not send install twice. + recovery = wizard.DirectSession(self.link, KEY, "secret-for-test", self.path, + progress=lambda _: None) + recovery._state = json.loads(self.path.read_text()) + recovery.saved, recovery.active = recovery._state["settings"], True + self.link.finish_install() + self.assertEqual(await recovery.cleanup(), []) + self.assertEqual(self.link.policies["target"]["autoinstall"], "trusted") + self.assertEqual([text for _, text in self.link.calls].count("ota install"), 1) + self.assertEqual(json.loads(self.path.read_text())["phase"], "finished") + + async def test_combined_lease_covers_two_long_transfers_and_reboots(self): + session = wizard.DirectSession(self.link, KEY, "secret-for-test", self.path, progress=lambda _: None) + await session.begin() + # Spend the first package's entire four-hour transfer and five-minute + # reboot budget, then reopen the rebooted target for the application. + self.link.now += 4 * 3600 + 5 * 60 + self.assertEqual(wizard.temp_state(await self.link.local_command("get tempradio")), "active") + self.link.windows["target"] = False + await session._open_window("target") + self.link.now += 4 * 3600 + 5 * 60 + self.assertEqual(wizard.temp_state(await self.link.local_command("get tempradio")), "active") + self.assertEqual(wizard.temp_state(await self.link.remote_command(KEY, "secret-for-test", "get tempradio")), "active") + source_windows = [text for label, text in self.link.calls if label == "source" and text.startswith("tempradio ")] + self.assertEqual(len(source_windows), 1) + self.assertEqual(await session.cleanup(), []) + + async def test_relative_window_acknowledgments_are_read_back_without_replay(self): + self.link.lost_window_ack = {"source", "target"} + await self.session.begin() + for label in ("source", "target"): + commands = [text for side, text in self.link.calls if side == label and text.startswith("tempradio ")] + self.assertEqual(len(commands), 1) + self.assertTrue(self.link.windows[label]) + self.assertEqual(await self.session.cleanup(), []) + async def test_cleanup_requires_policy_readback_and_can_be_retried(self): await self.session.begin() self.link.ignore_restore = True