From e285214349f197f08da35ca4e07bc11eea02fc7a Mon Sep 17 00:00:00 2001 From: mikecarper Date: Wed, 21 Oct 2026 04:31:07 -0700 Subject: [PATCH] Verify normal radio return for every OTA participant --- docs/lora_ota_automation.md | 29 ++ tools/lora_ota/lora_ota.py | 416 +++++++++++++----- tools/lora_ota/test_lora_ota.py | 11 +- tools/lora_ota/test_lora_ota_bootloader.py | 33 +- tools/lora_ota/test_lora_ota_hops.py | 29 +- .../test_lora_ota_relay_temp_cleanup.py | 171 +++++++ .../test_lora_ota_target_temp_cleanup.py | 176 ++++++++ 7 files changed, 754 insertions(+), 111 deletions(-) create mode 100644 tools/lora_ota/test_lora_ota_relay_temp_cleanup.py create mode 100644 tools/lora_ota/test_lora_ota_target_temp_cleanup.py diff --git a/docs/lora_ota_automation.md b/docs/lora_ota_automation.md index 106311f6a..0b18cec73 100644 --- a/docs/lora_ota_automation.md +++ b/docs/lora_ota_automation.md @@ -1067,6 +1067,35 @@ TempRadio window ends, or after putting the controller back on that tuple), rerunning the same package recognizes its manifest ID and resumes the existing session instead of clearing it. +The destination and each controlled relay also receive separate protected +`target-tempradio-cleanup.json` or `relay-tempradio-cleanup-.json` +records. Each record is flushed before its single one-minute shortening attempt. +A lost acknowledgement remains unconfirmed; a reply on the temporary channel +does not establish that the shortening arrived. Restoring the controller or +reaching the destination through another path does not establish that every +relay returned either. + +Default cleanup succeeds only after the controller is back on its normal +radio and every owned participant independently returns its bound full public +key and an inactive `get tempradio` result. Shared Full Companions must also +report their live TempRadio override inactive; Binary radio settings alone +show the saved preference. Unresolved participants retain their own record and +the run reports failure. A deliberately staged `--leave-controller-radio` run +preserves its windows and records. After an install, that option proves all +participants during the normal-radio verification interval before returning +the controller to the requested temporary tuple. + +After a process interruption, retain these records and use a new work directory. +Existing records cannot be overwritten to reset a shortening attempt. Never +repeat a relative `tempradio` command whose attempt is recorded. On the normal +route, use the recorded full key and a separately supplied admin password to +read `get public.key` and `get tempradio`. Both must confirm that same participant +and an inactive override; a timeout, changed key, active profile, or malformed +reply leaves its obligation pending. Wait for its original finite lease and +repeat only these read-only checks. Keep the original record and a separate +private restoration receipt binding its file hash and successful proofs. These +records contain no admin passwords and do not authorize reboot or installation. + When a chained run has already proved the newly running body hash, its retained previous package can briefly report `verifying staged blocks` after TempRadio reactivates the OTA manager. The runner waits only through the configured diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index 012e5c238..347de7946 100755 --- a/tools/lora_ota/lora_ota.py +++ b/tools/lora_ota/lora_ota.py @@ -157,6 +157,9 @@ LEGACY_OTA_CACHE_ROUNDS = 2 LEGACY_RELAY_REBOOT_MIN_REMAINING_SECONDS = 8 * 60 TARGET_RXPS_RECOVERY_FILE = "target-rxps-settings.json" SOURCE_RXPS_RECOVERY_FILE = "source-rxps-settings.json" +TARGET_TEMP_RECOVERY_FILE = "target-tempradio-cleanup.json" +RELAY_TEMP_RECOVERY_PREFIX = "relay-tempradio-cleanup-" +TARGET_TEMP_NORMAL_PROOF_SECONDS = 90 MIN_MESHCLI_VERSION = (1, 6, 0) # Tested format-3 inspect/verify/serve support. Keep automatic repairs pinned; # neither a package nor a moving branch may choose code to install on the host. @@ -7706,112 +7709,289 @@ def request_install( def shorten_relay_temp_windows( - controller: Controller, - args: argparse.Namespace, + controller: Controller, args: argparse.Namespace, relay_values: list[tuple[str, str]] | None = None, expected_public_keys: dict[str, str] | None = None, -) -> None: + *, cleanup_states: dict[str, RelayTempCleanup] | None = None, +) -> bool: + """Send each relative setter once; ACKs alone never establish normal return. + + Production passes durable per-participant ownership. Standalone callers + receive only the aggregate acknowledgement result, never a return proof. + """ values = args.relay_values if relay_values is None else relay_values - if not values: - return - command = temp_radio_command_for_minutes(args, TEMP_RADIO_RETURN_MINUTES) - print( - f"[relays] scheduling return to the normal channel in " - f"{TEMP_RADIO_RETURN_MINUTES} minute" - ) - errors: list[str] = [] - interrupted: KeyboardInterrupt | None = None - for relay_name, relay_password in values: + if cleanup_states is not None: + for key, _ in values: + state = cleanup_states.get(key) + if (state is None or (expected_public_keys is not None + and expected_public_keys.get(key) != state.key)): + raise OtaError("relay cleanup is missing bound durable ownership") + errors = [] + interrupted = None + acknowledged = True + for key, password in values: try: - try: - reply = controller.remote_command( - relay_name, - command, - password=relay_password, - retry=False, - ) - except TransmissionError as exc: - # Never replay a relative lease. Prove which authenticated - # relay remained reachable on this tuple; its original window - # is still bounded if the one-minute write did not arrive. - actual_key = read_remote_public_key_bounded( - controller, - relay_name, - password=relay_password, - ) - expected_key = ( - expected_public_keys.get(relay_name) - if expected_public_keys is not None - else None - ) - if expected_key is not None and actual_key != expected_key: - raise OtaError( - f"identity changed after lost cleanup reply: expected " - f"{expected_key}, got {actual_key}" - ) - print( - f"[relays] {relay_name} one-minute TempRadio reply was " - f"lost ({exc}); exact identity reconciled the one-shot " - "request, and its original lease remains bounded" - ) + if cleanup_states is None: + ack = shorten_remote_temp_window(controller, args, key, + password=password, label="relay") else: - require_temp_radio_reply( - relay_name, reply, TEMP_RADIO_RETURN_MINUTES - ) + state = cleanup_states[key] + state.shorten_once(controller, args) + ack = state.report["shorten_acknowledged"] + acknowledged = acknowledged and ack except KeyboardInterrupt as exc: - # Continue through the remaining independently owned relays. The - # interrupt is re-raised only after each has received one attempt. - if interrupted is None: - interrupted = exc - errors.append(f"{relay_name}: interrupted") + interrupted = interrupted or exc + errors.append(f"{key}: interrupted") except (OtaError, OSError) as exc: - errors.append(f"{relay_name}: {exc}") + errors.append(f"{key}: " + redact_text(str(exc), (password,))) + if interrupted is not None: + raise interrupted if errors: - detail = "; ".join(errors) - if interrupted is not None: - print( - "WARNING: relay cleanup continued after interruption: " - + detail, - file=sys.stderr, - ) - raise interrupted - raise OtaError("could not shorten every relay TempRadio lease: " + detail) + raise OtaError("could not shorten every relay TempRadio lease: " + "; ".join(errors)) + return acknowledged -def shorten_target_temp_window( - controller: Controller, - args: argparse.Namespace, -) -> None: +def shorten_remote_temp_window( + controller: Controller, args: argparse.Namespace, key: str, + *, password: str | None = None, label: str = "destination", +) -> bool: + """Attempt one relative shortening; return whether its ACK was proven. + + Reachability on the temporary tuple cannot prove that a lost setter was + delivered. The caller retains a separate normal-return obligation either + way, and must not replay this duration after a missing acknowledgement. + """ command = temp_radio_command_for_minutes(args, TEMP_RADIO_RETURN_MINUTES) print( - f"[destination] scheduling return to the normal channel in " + f"[{label}] scheduling return to the normal channel in " f"{TEMP_RADIO_RETURN_MINUTES} minute" ) try: - reply = controller.remote_command( - args.target, command, retry=False - ) + options = {"password": password} if password is not None else {} + reply = controller.remote_command(key, command, retry=False, **options) except TransmissionError as exc: - identity = retry_transmission_bounded( - lambda: controller.remote_command( - args.target, "ota self", retry=False - ), - f"cleanup identity reconciliation for {args.target}", - ) - if re.search(r"\bbase_hash=[0-9A-Fa-f]{16}\b", identity) is None: - raise OtaError( - "destination replied after a lost cleanup request but did not " - "return a valid running EndF identity" - ) print( - "[destination] one-minute TempRadio reply was lost " - f"({exc}); exact identity reconciled the one-shot request, and the " - "original lease remains bounded" + f"WARNING: {label} one-minute TempRadio acknowledgement was " + f"lost ({redact_text(str(exc), (password,) if password else ())}); shortening is unconfirmed and was not replayed. " + "The original long lease may remain active; exact normal-channel " + "identity and inactive TempRadio must still be proven.", + file=sys.stderr, ) - return - require_temp_radio_reply( - args.target, reply, TEMP_RADIO_RETURN_MINUTES - ) + return False + require_temp_radio_reply(key, reply, TEMP_RADIO_RETURN_MINUTES) + return True + + +class RemoteTempCleanup: + """One bound participant's durable return obligation and bounded read proof.""" + + def __init__(self, path: Path, args: argparse.Namespace, normal_radio: RadioSettings, + key: str, role: str, password: str | None = None): + if path.exists() or path.is_symlink(): + raise OtaError("existing TempRadio recovery obligation must not be overwritten: " + str(path)) + self.path = path + self.normal_radio = normal_radio + self.key, self.role, self.password = key, role, password + self.sensitive_values = tuple(value for value in (password, getattr(args, "password", None)) if value) + self.report: dict = { + "schema": 1, + "target": key, + "role": role, + "normal_source_radio": normal_radio.meshcli_value(), + "temporary_radio": args.temp_radio, + "cleanup_required": True, + "shorten_attempted": False, + "shorten_acknowledged": False, + "normal_proof_attempted": False, + "normal_return_verified": False, + "proof_records": [], + } + self.persist() + + def redact_error(self, exc: BaseException) -> str: + return redact_text(str(exc), self.sensitive_values) + + def _shorten(self, controller: Controller, args: argparse.Namespace) -> bool: + raise NotImplementedError + + def persist(self) -> None: + write_private_recovery_file( + self.path, json.dumps(self.report, indent=2, sort_keys=True) + "\n" + ) + + def require_binding(self, args: argparse.Namespace) -> None: + expected = self.report["target"] + bound = (args.target == expected if self.role == "destination" else + any(key == expected for key, _ in args.relay_values)) + if (not re.fullmatch(r"[0-9A-Fa-f]{64}", expected) + or self.key != expected or not bound): + raise OtaError("normal-return proof requires the unchanged full participant key") + + def shorten_once(self, controller: Controller, args: argparse.Namespace) -> None: + if self.report["shorten_attempted"]: + return + self.require_binding(args) + self.report["shorten_attempted"] = True + self.persist() + try: + self.report["shorten_acknowledged"] = ( + self._shorten(controller, args) is True + ) + except BaseException as exc: + self.report["shorten_error"] = self.redact_error(exc) + raise + finally: + self.persist() + + def prove_normal(self, controller: Controller, args: argparse.Namespace) -> None: + if not self.report["cleanup_required"]: + return + if self.report["normal_proof_attempted"]: + raise OtaError(f"{self.role} normal return remains unproven; inspect " + str(self.path)) + self.require_binding(args) + expected = self.key + self.report["normal_proof_attempted"] = True + self.persist() + deadline = time.monotonic() + TARGET_TEMP_NORMAL_PROOF_SECONDS + + def read(command: str) -> str: + remaining = deadline - time.monotonic() + if remaining < 1: + raise TransmissionError(f"{self.role} normal-return proof deadline expired") + record = {"command": command, "started_at": time.monotonic(), + "operation_timeout": min(30.0, remaining)} + self.report["proof_records"].append(record) + self.persist() + try: + reply = controller.remote_command( + expected, command, retry=False, + **({"password": self.password} if self.password is not None else {}), + reply_timeout=min(20, int(remaining)), + operation_timeout=record["operation_timeout"], + ) + record["reply"] = redact_text(reply, self.sensitive_values) + return reply + except BaseException as exc: + record["error"] = self.redact_error(exc) + raise + finally: + self.persist() + + def proof() -> None: + remaining = deadline - time.monotonic() + if remaining < 1: + raise TransmissionError(f"{self.role} normal-return proof deadline expired") + if getattr(args, "source_shares_controller", False): + # Binary GET_RADIO reports saved preferences on a Full + # Companion, even while its live TempRadio override is active. + record = {"command": "shared source tempradio", "started_at": time.monotonic()} + self.report["proof_records"].append(record) + self.persist() + try: + source_status = read_source_temp_radio_status(args, deadline=deadline) + record["state"] = source_status[0] if source_status is not None else "unsupported" + except BaseException as exc: + record["error"] = self.redact_error(exc) + raise + finally: + self.persist() + if source_status is None: + raise OtaError("shared controller live TempRadio state is unavailable") + if source_status[0] != "inactive": + raise TransmissionError("shared controller live TempRadio remains active; saved normal settings are insufficient") + remaining = deadline - time.monotonic() + if remaining < 1: + raise TransmissionError("shared controller normal-return proof deadline expired") + require_radio_settings( + controller.get_radio(timeout=min(30.0, remaining)), + self.normal_radio, f"{self.role} cleanup normal controller", + ) + public = parse_cli_public_key(read("get public.key"), f"{self.role} cleanup") + if public.lower() != expected.lower(): + raise OtaError(f"{self.role} normal-return full public key changed") + status = read("get tempradio").strip() + status = status[1:].strip() if status.startswith(">") else status + if status == "off": + return + parts = status.split(",") + try: + active = (len(parts) == 5 + and all(math.isfinite(float(x)) for x in parts) + and 150 <= float(parts[0]) <= 2500 + and float(parts[1]) > 0 + and 5 <= int(parts[2]) <= 12 + and 5 <= int(parts[3]) <= 8 + and 1 <= int(parts[4]) <= 65535) + except ValueError: + active = False + if not active: + raise OtaError(f"{self.role} returned malformed TempRadio state during cleanup") + raise TransmissionError(f"{self.role} still reports active TempRadio; normal return is unproven") + + try: + # A confirmed one-minute lease can remain active during the first + # checks. Poll only replay-safe reads within one absolute bound. + retry_transmission_before_deadline( + proof, f"{self.role} normal-return proof", deadline, + retry_limit=12, deadline_label=f"90-second {self.role} cleanup window", + ) + except BaseException as exc: + self.report["normal_proof_error"] = self.redact_error(exc) + raise + else: + self.report.update(cleanup_required=False, normal_return_verified=True) + print(f"[{self.role}] exact full identity and inactive TempRadio verified on the normal channel") + finally: + self.persist() + + +class TargetTempCleanup(RemoteTempCleanup): + def __init__(self, work_dir: Path, args: argparse.Namespace, normal_radio: RadioSettings, + password: str | None = None): + super().__init__(work_dir / TARGET_TEMP_RECOVERY_FILE, args, normal_radio, + args.target, "destination", password) + + def _shorten(self, controller: Controller, args: argparse.Namespace) -> bool: + return shorten_target_temp_window(controller, args, password=self.password) + + +class RelayTempCleanup(RemoteTempCleanup): + def __init__(self, work_dir: Path, args: argparse.Namespace, normal_radio: RadioSettings, + key: str, password: str, expected_key: str): + if (not re.fullmatch(r"[0-9A-Fa-f]{64}", key) + or key != expected_key or not any(k == key for k, _ in args.relay_values)): + raise OtaError("relay cleanup requires its unchanged bound full public key") + super().__init__(work_dir / f"{RELAY_TEMP_RECOVERY_PREFIX}{key}.json", args, + normal_radio, key, "relay", password) + + def _shorten(self, controller: Controller, args: argparse.Namespace) -> bool: + return shorten_remote_temp_window(controller, args, self.key, + password=self.password, label="relay") + + +def shorten_target_temp_window(controller: Controller, args: argparse.Namespace, + *, password: str | None = None) -> bool: + return shorten_remote_temp_window(controller, args, args.target, password=password) + + +def prove_owned_temp_cleanups(controller: Controller, args: argparse.Namespace, + target: TargetTempCleanup | None, + relays: dict[str, RelayTempCleanup]) -> None: + """Attempt each independent normal-return proof even if another fails.""" + errors = [] + interrupted = None + for state in ([target] if target is not None else []) + list(relays.values()): + try: + state.prove_normal(controller, args) + except KeyboardInterrupt as exc: + interrupted = interrupted or exc + errors.append(f"{state.role}: interrupted; recovery {state.path}") + except (OtaError, OSError) as exc: + errors.append(f"{state.role}: {state.redact_error(exc)}; recovery {state.path}") + if interrupted is not None: + raise interrupted + if errors: + raise OtaError("normal return remains unproven: " + "; ".join(errors)) def shorten_source_temp_window( @@ -9234,6 +9414,8 @@ def main( source_rxps_changed = False source_rxps_recovery_path: Path | None = None target_temp_owned = False + target_temp_cleanup: TargetTempCleanup | None = None + relay_temp_cleanups: dict[str, RelayTempCleanup] = {} target_rxps_saved: RxpsSettings | None = None target_rxps_changed = False armed_relay_values: list[tuple[str, str]] = [] @@ -9599,6 +9781,14 @@ def main( print(f"[legacy] relay reboot fallback eligible: " f"{args.relay_reboot_normal_hops} normal-channel hops") + target_temp_cleanup = TargetTempCleanup(work_dir, args, original_radio, password=password) + for relay_key, relay_password in args.relay_values: + if relay_key in relay_temp_cleanups: + raise OtaError("duplicate bound relay cleanup identity") + relay_temp_cleanups[relay_key] = RelayTempCleanup( + work_dir, args, original_radio, relay_key, relay_password, + relay_public_keys[relay_key], + ) target_temp_armed_at = time.monotonic() target_temp_owned = True if args.source_shares_controller: @@ -9728,11 +9918,11 @@ def main( "--leave-controller-radio was requested" ) else: - # Retire one-shot ownership before the write. If its reply is - # lost, reconciliation is read-only and outer cleanup must not - # send the relative duration a second time. - target_temp_owned = False - shorten_target_temp_window(controller, args) + # One-shot write ownership and the outstanding normal-return + # obligation are separate. A missing ACK cannot retire either + # the durable target recovery record or the return proof. + assert target_temp_cleanup is not None + target_temp_cleanup.shorten_once(controller, args) relays_to_shorten = list(armed_relay_values) armed_relay_values.clear() shorten_relay_temp_windows( @@ -9740,6 +9930,7 @@ def main( args, relays_to_shorten, relay_public_keys, + cleanup_states=relay_temp_cleanups, ) if target_rxps_changed and target_rxps_saved is not None: if args.leave_controller_radio: @@ -9770,6 +9961,10 @@ def main( controller_changed = False restore_source_rxps_once("stage-only cleanup") retire_local_source_rxps_recovery() + if not args.leave_controller_radio: + assert target_temp_cleanup is not None + prove_owned_temp_cleanups(controller, args, target_temp_cleanup, relay_temp_cleanups) + target_temp_owned = False return 0 install_confirmed = request_install(controller, args, package) @@ -9796,6 +9991,7 @@ def main( args, relays_to_shorten, relay_public_keys, + cleanup_states=relay_temp_cleanups, ) # Seeding already stopped before policy restoration and installation. @@ -9829,6 +10025,9 @@ def main( target.base_hash, identity_reply, ) + if args.leave_controller_radio: + assert target_temp_cleanup is not None + prove_owned_temp_cleanups(controller, args, target_temp_cleanup, relay_temp_cleanups) if target_rxps_changed and target_rxps_saved is not None: restore_remote_rxps( controller, args.target, target_rxps_saved @@ -9842,6 +10041,9 @@ def main( controller_changed = True restore_source_rxps_once("successful update cleanup") retire_local_source_rxps_recovery() + if not args.leave_controller_radio: + assert target_temp_cleanup is not None + prove_owned_temp_cleanups(controller, args, target_temp_cleanup, relay_temp_cleanups) return 0 except KeyboardInterrupt: print("\nInterrupted; any partial target download remains resumable.", file=sys.stderr) @@ -9858,7 +10060,8 @@ def main( if seeder_attempted and args.source_serial: source_cli_command(args, "ota folder off", check=False) needs_remote_cleanup = ( - target_temp_owned + (target_temp_owned and target_temp_cleanup is not None + and not target_temp_cleanup.report["shorten_attempted"]) or bool(armed_relay_values) or bool(relay_timing_settings) ) @@ -9901,12 +10104,11 @@ def main( file=sys.stderr, ) if target_temp_owned and not args.leave_controller_radio: - # This relative cleanup write is one-shot even when its - # acknowledgement is lost; retire ownership first so a - # later finally path cannot extend it again. - target_temp_owned = False + # Persist the one-shot attempt before sending without + # dropping the independent target return obligation. + assert target_temp_cleanup is not None try: - shorten_target_temp_window(controller, args) + target_temp_cleanup.shorten_once(controller, args) except (OtaError, OSError) as exc: print( "WARNING: could not shorten the destination " @@ -9925,6 +10127,7 @@ def main( args, relays_to_shorten, relay_public_keys, + cleanup_states=relay_temp_cleanups, ) except (OtaError, OSError) as exc: print( @@ -10022,6 +10225,21 @@ def main( "after checking the current policy and radio channel", file=sys.stderr, ) + pending_temp = ([target_temp_cleanup] if target_temp_cleanup is not None else []) + list(relay_temp_cleanups.values()) + if any(state.report["cleanup_required"] for state in pending_temp) and not args.leave_controller_radio: + if controller is not None and not controller_changed and not source_temp_owned: + try: + prove_owned_temp_cleanups(controller, args, target_temp_cleanup, relay_temp_cleanups) + target_temp_owned = False + except (OtaError, OSError) as exc: + print(f"CRITICAL: participant normal return is unproven: {exc}", file=sys.stderr) + for state in pending_temp: + if state.report["cleanup_required"]: + print( + f"CRITICAL: restoring the controller does not prove the {state.role} returned. " + "Its original long TempRadio lease may remain active; no relative cleanup setter was replayed. " + f"Retained recovery obligation: {state.path}", file=sys.stderr, + ) if not source_rxps_changed: try: retire_local_source_rxps_recovery() diff --git a/tools/lora_ota/test_lora_ota.py b/tools/lora_ota/test_lora_ota.py index b913c0f64..9f46ffd59 100644 --- a/tools/lora_ota/test_lora_ota.py +++ b/tools/lora_ota/test_lora_ota.py @@ -6543,7 +6543,7 @@ class ReliabilityTests(unittest.TestCase): [("offline", False), ("online", False)], ) - def test_lost_relay_cleanup_reconciles_identity_without_replay(self) -> None: + def test_lost_relay_cleanup_does_not_claim_return_from_temporary_identity(self) -> None: key = "AB" * 32 class Controller: @@ -6565,11 +6565,10 @@ class ReliabilityTests(unittest.TestCase): relay_values=[("relay", "secret")], temp_values=(909.95, 250.0, 5, 5, 120), ) - ota.shorten_relay_temp_windows( - controller, - args, - expected_public_keys={"relay": key.lower()}, + acknowledged = ota.shorten_relay_temp_windows( + controller, args, expected_public_keys={"relay": key.lower()}, ) + self.assertFalse(acknowledged) relative_writes = [ call for call in controller.calls if call[1].startswith("tempradio ") ] @@ -6577,7 +6576,7 @@ class ReliabilityTests(unittest.TestCase): relative_writes, [("relay", "tempradio 909.95,250,5,5,1", False)], ) - self.assertIn(("relay", "get public.key", False), controller.calls) + self.assertEqual(controller.calls, relative_writes) def test_managed_relay_timing_is_saved_guarded_and_restored(self) -> None: class Controller: diff --git a/tools/lora_ota/test_lora_ota_bootloader.py b/tools/lora_ota/test_lora_ota_bootloader.py index 46b113f72..9508c910e 100644 --- a/tools/lora_ota/test_lora_ota_bootloader.py +++ b/tools/lora_ota/test_lora_ota_bootloader.py @@ -4,6 +4,7 @@ import argparse import contextlib from dataclasses import replace import io +import json import os from pathlib import Path import shutil @@ -328,10 +329,22 @@ class BootloaderStagingTests(unittest.TestCase): source_key: {"adv_name": "source", "public_key": source_key}, }] normal = ota.RadioSettings(910.525, 62.5, 7, 5, False) - controller.get_radio.return_value = normal + current_radio = normal + cleanup_proof_radios = [] + + def set_radio(settings, _label): + nonlocal current_radio + current_radio = settings + + controller.get_radio.side_effect = lambda **_kwargs: current_radio + controller.set_radio.side_effect = set_radio def remote_command(station, command, **_kwargs): self.assertEqual(station, destination_key) + if command in ("get public.key", "get tempradio"): + self.assertEqual(current_radio, normal) + cleanup_proof_radios.append(current_radio) + return f"> {destination_key}" if command == "get public.key" else "> off" replies = { "ota status": "OTA | no download | target:1234ABCD", "ota ls": "Updates 1", @@ -356,7 +369,7 @@ class BootloaderStagingTests(unittest.TestCase): "require_command", "require_meshcli_version", "require_bootloader_tool_support", "preflight_source_cli", "ensure_source_clock_gate_safe", "ensure_controller_clock_safe", "verify_with_motatool", "arm_target_temp_radio", - "switch_controller_to_temp_radio", "source_cli_command", + "source_cli_command", ): stack.enter_context(mock.patch.object(ota, name)) stack.enter_context(mock.patch.object(ota, "read_source_public_key_bounded", return_value=source_key)) @@ -387,8 +400,22 @@ class BootloaderStagingTests(unittest.TestCase): rehearsal.assert_called_once() seeder.start.assert_called_once() cleanup.assert_called_once() - controller.set_radio.assert_called_once_with(normal, "restore controller radio after staging") + temporary = ota.RadioSettings(*ota.parse_temp_radio(ota.DEFAULT_LAB_TEMP_RADIO)[:4], normal.repeat) + self.assertEqual(controller.set_radio.call_args_list, [ + mock.call(temporary, "switch controller to TempRadio"), + mock.call(normal, "restore controller radio after staging"), + ]) + saved_cleanup = json.loads((root / "work" / ota.TARGET_TEMP_RECOVERY_FILE).read_text()) + self.assertEqual(saved_cleanup["target"], destination_key) + self.assertTrue(saved_cleanup["shorten_attempted"]) + self.assertTrue(saved_cleanup["normal_return_verified"]) + self.assertFalse(saved_cleanup["cleanup_required"]) + self.assertEqual([item["command"] for item in saved_cleanup["proof_records"]], + ["get public.key", "get tempradio"]) + self.assertEqual(cleanup_proof_radios, [normal, normal]) install.assert_not_called() + self.assertFalse(any("install" in call.args[1].split() + for call in controller.remote_command.call_args_list)) self.assertIn( mock.call(destination_key, f"ota pull {package.manifest_id} flash", retry=False), controller.remote_command.call_args_list, diff --git a/tools/lora_ota/test_lora_ota_hops.py b/tools/lora_ota/test_lora_ota_hops.py index 5ff83e3ac..edd5607d1 100644 --- a/tools/lora_ota/test_lora_ota_hops.py +++ b/tools/lora_ota/test_lora_ota_hops.py @@ -24,8 +24,10 @@ class Stations: self.overrides = {} self.seeding = False self.normal = ota.RadioSettings(910.525, 62.5, 7, 5, False) + self.actual_radio = self.normal def command(self, name, command, **options): + name = {"a" * 64: "relay0", "b" * 64: "remote"}.get(name, name) self.calls.append((name, command, options)) if name in self.fail_reads and command == "ota config": raise ota.TransmissionError("lost config read") @@ -57,6 +59,7 @@ class Stations: return self.normal def set_radio(self, _radio, _label): + self.actual_radio = _radio self.events.append("controller-restore") def close(self): @@ -327,6 +330,8 @@ class OtaHopLifecycleTests(unittest.TestCase): def phase(name): def action(*_args, **_kwargs): stations.events.append(name) + if name == "switch": + stations.actual_radio = ota.RadioSettings(909.5, 500, 5, 5, False) if failure == name: if name == "interrupt": raise KeyboardInterrupt @@ -348,15 +353,24 @@ class OtaHopLifecycleTests(unittest.TestCase): raise ota.OtaError("synthetic install failure") return True - argv = ["release.mota", "remote", "--controller-serial", "/dev/controller", + argv = ["release.mota", "b" * 64, "--controller-serial", "/dev/controller", "--source-serial", "/dev/source", "--password", "secret", - "--relay", "relay0=relay-secret", "--work-dir", str(Path(directory) / "work"), + "--relay", "a" * 64 + "=relay-secret", "--work-dir", str(Path(directory) / "work"), "--yes"] if no_install: argv.append("--no-install") if leave: argv.append("--leave-controller-radio") with contextlib.ExitStack() as stack: + # This fixture isolates hop lifecycle using bound fixture keys. + # Exact full-key/TempRadio proof has separate transport regressions. + stack.enter_context(mock.patch.object( + ota.RemoteTempCleanup, "prove_normal", autospec=True, + side_effect=lambda cleanup, *_: ( + self.assertTrue(stations.normal.matches(stations.actual_radio)), + stations.events.append("normal-proof:" + cleanup.role), + cleanup.report.update(cleanup_required=False, normal_return_verified=True)), + )) for name in ("preflight_inputs", "preflight_source_cli", "bind_contact_selectors", "verify_shared_source_identity", "ensure_controller_clock_safe", "restore_relay_timings", "shorten_relay_temp_windows", @@ -377,7 +391,7 @@ class OtaHopLifecycleTests(unittest.TestCase): stack.enter_context(mock.patch.object(ota, "arm_relay_temp_radio_once", return_value=True)) stack.enter_context(mock.patch.object(ota, "arm_source_temp_radio_once", return_value=True)) stack.enter_context(mock.patch.object(ota, "switch_controller_to_temp_radio", side_effect=phase("switch"))) - stack.enter_context(mock.patch.object(ota, "read_relay_timing", return_value=ota.RelayTimingSettings("relay0", "relay-secret", 0, 0.3))) + stack.enter_context(mock.patch.object(ota, "read_relay_timing", return_value=ota.RelayTimingSettings("a" * 64, "relay-secret", 0, 0.3))) stack.enter_context(mock.patch.object(ota, "read_remote_public_key_bounded", return_value="a" * 64)) stack.enter_context(mock.patch.object(ota, "shorten_source_temp_window", return_value=True)) stack.enter_context(mock.patch.object(ota, "SeederProcess", return_value=seeder)) @@ -411,6 +425,15 @@ class OtaHopLifecycleTests(unittest.TestCase): self.assertLess(events.index("set:remote:1"), events.index("arm-target")) self.assertLess(events.index("seeder-stop"), events.index("set:source:0")) + def test_install_leave_controller_proves_every_participant_while_actually_normal(self): + with tempfile.TemporaryDirectory() as directory: + result,stations=self.run_main(directory,leave=True,no_install=False) + self.assertEqual(result,0) + self.assertIn("normal-proof:destination",stations.events) + self.assertIn("normal-proof:relay",stations.events) + self.assertFalse(stations.normal.matches(stations.actual_radio)) + self.assertGreater(stations.events.index("normal-proof:relay"),stations.events.index("install")) + def test_rehearsal_and_confirmation_failures_never_write_hop_settings(self): for failure in ("confirm", "rehearse"): with self.subTest(failure=failure), tempfile.TemporaryDirectory() as directory: diff --git a/tools/lora_ota/test_lora_ota_relay_temp_cleanup.py b/tools/lora_ota/test_lora_ota_relay_temp_cleanup.py new file mode 100644 index 000000000..2a14fac34 --- /dev/null +++ b/tools/lora_ota/test_lora_ota_relay_temp_cleanup.py @@ -0,0 +1,171 @@ +"""Offline per-relay lease obligations share the production target proof.""" +import argparse +import contextlib +import io +import json +from pathlib import Path +import tempfile +import subprocess +import sys +import types +import unittest +from unittest import mock +import lora_ota as ota + +TARGET='ab'*32 +RELAYS=['cd'*32,'ef'*32,'12'*32] +NORMAL=ota.RadioSettings(910.525,62.5,7,5,False) + +class RelayCleanupTests(unittest.TestCase): + def args(self): + return argparse.Namespace(target=TARGET,temp_radio='909.5,500,7,5,180', + temp_values=(909.5,500,7,5,180),relay_values=[(key,'private-'+str(i)) for i,key in enumerate(RELAYS)]) + def state(self,directory,args,key): + return ota.RelayTempCleanup(Path(directory),args,NORMAL,key,dict(args.relay_values)[key],key) + def controller(self,args,modes=None): + modes=modes or {};calls=[] + def remote(key,command,**kwargs): + self.assertIs(kwargs['retry'],False) + if key!=TARGET:self.assertEqual(kwargs['password'],dict(args.relay_values)[key]) + calls.append((key,command)) + mode=modes.get(key,'off') + if command.startswith('tempradio '): + if mode in ('lost-delivered','lost-undelivered'):raise ota.TransmissionError('lost ACK') + if mode=='error':raise ota.OtaError('failed private-0') + if mode=='interrupt':raise KeyboardInterrupt() + return 'OK - temp params for 1 mins' + if command=='get public.key':return '> '+('34'*32 if mode=='changed-key' else key) + if command=='get tempradio':return '> 909.500,500.000,7,5,32' if mode in ('active','lost-undelivered') else '> off' + raise AssertionError(command) + return types.SimpleNamespace(remote_command=remote,get_radio=lambda **kwargs:NORMAL,calls=calls) + def clock(self): + clock=types.SimpleNamespace(now=100.0) + def sleep(seconds):clock.now+=seconds + return clock,types.SimpleNamespace(monotonic=lambda:clock.now,sleep=sleep) + def test_lost_delivered_relay_ack_resolves_only_from_independent_normal_off(self): + args=self.args();key=RELAYS[0];c=self.controller(args,{key:'lost-delivered'}) + with tempfile.TemporaryDirectory() as d,contextlib.redirect_stderr(io.StringIO()): + state=self.state(d,args,key) + self.assertFalse(ota.shorten_relay_temp_windows(c,args,[(key,'private-0')],{key:key},cleanup_states={key:state})) + self.assertTrue(state.report['cleanup_required']) + self.assertFalse(state.report['shorten_acknowledged']) + state.prove_normal(c,args) + self.assertTrue(state.report['normal_return_verified']) + self.assertEqual(c.calls,[(key,'tempradio 909.5,500,7,5,1'),(key,'get public.key'),(key,'get tempradio')]) + self.assertNotIn('private-0',state.path.read_text()) + def test_lost_undelivered_relay_ack_keeps_its_own_obligation(self): + args=self.args();key=RELAYS[0];c=self.controller(args,{key:'lost-undelivered'});clock,fake=self.clock() + with tempfile.TemporaryDirectory() as d,contextlib.redirect_stderr(io.StringIO()),contextlib.redirect_stdout(io.StringIO()): + state=self.state(d,args,key) + state.shorten_once(c,args) + with mock.patch.object(ota,'time',fake),self.assertRaises(ota.TransmissionError):state.prove_normal(c,args) + saved=json.loads(state.path.read_text()) + self.assertTrue(saved['cleanup_required']);self.assertFalse(saved['normal_return_verified']) + self.assertLessEqual(clock.now,190) + state.shorten_once(c,args) + self.assertEqual(sum(cmd.startswith('tempradio ') for _,cmd in c.calls),1) + def test_changed_relay_full_identity_is_terminal_and_never_resolves(self): + args=self.args();key=RELAYS[0];c=self.controller(args,{key:'changed-key'}) + with tempfile.TemporaryDirectory() as d: + state=self.state(d,args,key) + with self.assertRaisesRegex(ota.OtaError,'public key changed'):state.prove_normal(c,args) + self.assertEqual(c.calls,[(key,'get public.key')]);self.assertTrue(state.report['cleanup_required']) + def test_target_off_does_not_prove_a_relay_off_and_other_proofs_continue(self): + args=self.args();c=self.controller(args,{RELAYS[0]:'active'});clock,fake=self.clock() + with tempfile.TemporaryDirectory() as d,contextlib.redirect_stdout(io.StringIO()): + target=ota.TargetTempCleanup(Path(d),args,NORMAL) + states={key:self.state(d,args,key) for key in RELAYS[:2]} + with mock.patch.object(ota,'time',fake),self.assertRaisesRegex(ota.OtaError,'normal return remains unproven'): + ota.prove_owned_temp_cleanups(c,args,target,states) + self.assertFalse(target.report['cleanup_required']) + self.assertTrue(states[RELAYS[0]].report['cleanup_required']) + self.assertFalse(states[RELAYS[1]].report['cleanup_required']) + before=list(c.calls) + with self.assertRaises(ota.OtaError):ota.prove_owned_temp_cleanups(c,args,target,states) + self.assertEqual(c.calls,before) # No repeated failed proof nor relative setter. + def test_each_independent_shortening_attempt_survives_errors_and_interrupts_once(self): + args=self.args();modes={RELAYS[0]:'error',RELAYS[1]:'interrupt',RELAYS[2]:'lost-delivered'} + c=self.controller(args,modes) + with tempfile.TemporaryDirectory() as d,contextlib.redirect_stderr(io.StringIO()): + states={key:self.state(d,args,key) for key in RELAYS} + original=c.remote_command + def before_send(key,command,**kwargs): + self.assertTrue(json.loads(states[key].path.read_text())['shorten_attempted']) + return original(key,command,**kwargs) + c.remote_command=before_send + with self.assertRaises(KeyboardInterrupt):ota.shorten_relay_temp_windows(c,args,cleanup_states=states) + self.assertEqual([key for key,_ in c.calls],RELAYS) + self.assertFalse(ota.shorten_relay_temp_windows(c,args,cleanup_states=states)) + self.assertEqual(len(c.calls),3) + for key,state in states.items(): + self.assertTrue(state.report['shorten_attempted']);self.assertTrue(state.report['cleanup_required']) + self.assertNotIn(dict(args.relay_values)[key],state.path.read_text()) + self.assertEqual(state.path.stat().st_mode&0o777,0o600) + def test_missing_or_rebound_ownership_stops_before_any_write(self): + args=self.args();c=self.controller(args) + with tempfile.TemporaryDirectory() as d: + states={RELAYS[0]:self.state(d,args,RELAYS[0])} + with self.assertRaises(ota.OtaError):ota.shorten_relay_temp_windows(c,args,cleanup_states=states) + self.assertEqual(c.calls,[]) + with self.assertRaises(ota.OtaError):ota.RelayTempCleanup(Path(d),args,NORMAL,RELAYS[1],'private-1',RELAYS[2]) + args.relay_values=[] + with self.assertRaises(ota.OtaError):states[RELAYS[0]].prove_normal(c,args) + self.assertEqual(c.calls,[]) + + def test_fresh_process_keeps_attempt_and_constructor_cannot_reset_it(self): + args=self.args();key=RELAYS[0];c=self.controller(args,{key:'lost-delivered'}) + with tempfile.TemporaryDirectory() as d,contextlib.redirect_stderr(io.StringIO()): + state=self.state(d,args,key);state.shorten_once(c,args) + before=state.path.read_bytes() + probe=subprocess.run([sys.executable,'-B','-c', + 'import json,sys;d=json.load(open(sys.argv[1]));assert d["shorten_attempted"] and d["cleanup_required"];assert not d["shorten_acknowledged"]',str(state.path)], + capture_output=True,text=True,timeout=10) + self.assertEqual(probe.returncode,0,probe.stderr) + with self.assertRaisesRegex(ota.OtaError,'must not be overwritten'):self.state(d,args,key) + self.assertEqual(state.path.read_bytes(),before) + self.assertEqual(sum(cmd.startswith('tempradio ') for _,cmd in c.calls),1) + + def test_saved_normal_radio_does_not_hide_a_shared_live_temporary_override(self): + args=self.args();args.source_shares_controller=True + c=self.controller(args);clock,fake=self.clock() + with tempfile.TemporaryDirectory() as d,contextlib.redirect_stdout(io.StringIO()): + state=self.state(d,args,RELAYS[0]) + with mock.patch.object(ota,'time',fake),mock.patch.object(ota,'read_source_temp_radio_status',return_value=('active',ota.RadioSettings(909.5,500,7,5,False))),self.assertRaises(ota.TransmissionError): + state.prove_normal(c,args) + self.assertTrue(state.report['cleanup_required']) + self.assertEqual(c.calls,[]) + self.assertTrue(all(r.get('state')=='active' for r in state.report['proof_records'])) + + def test_authenticated_reply_capture_never_persists_private_password_text(self): + args=self.args();key=RELAYS[0];c=self.controller(args);original=c.remote_command + def reply(key,command,**kwargs): + value=original(key,command,**kwargs) + return value+'\nprivate-0' if command=='get public.key' else value + c.remote_command=reply + with tempfile.TemporaryDirectory() as d: + state=self.state(d,args,key);state.prove_normal(c,args) + self.assertFalse(state.report['cleanup_required']) + self.assertNotIn('private-0',state.path.read_text()) + + def test_target_env_or_getpass_secret_is_redacted_without_args_password(self): + for fail in (False,True): + with self.subTest(fail=fail),tempfile.TemporaryDirectory() as d: + args=self.args();args.password=None;secret='resolved-getpass-fixture' + def remote(key,command,**kwargs): + self.assertEqual(kwargs['password'],secret) + if command.startswith('tempradio '):raise ota.TransmissionError('lost '+secret) + if command=='get public.key':return '> '+TARGET+'\n'+secret + if fail:raise ota.OtaError('proof failed '+secret) + return '> off' + c=types.SimpleNamespace(remote_command=remote,get_radio=lambda **kwargs:NORMAL) + state=ota.TargetTempCleanup(Path(d),args,NORMAL,password=secret) + warning=io.StringIO() + with contextlib.redirect_stderr(warning):state.shorten_once(c,args) + self.assertNotIn(secret,warning.getvalue()) + if fail: + with self.assertRaises(ota.OtaError):state.prove_normal(c,args) + self.assertTrue(state.report['cleanup_required']) + else:state.prove_normal(c,args) + self.assertNotIn(secret,state.path.read_text()) + +if __name__=='__main__':unittest.main() diff --git a/tools/lora_ota/test_lora_ota_target_temp_cleanup.py b/tools/lora_ota/test_lora_ota_target_temp_cleanup.py new file mode 100644 index 000000000..55825507e --- /dev/null +++ b/tools/lora_ota/test_lora_ota_target_temp_cleanup.py @@ -0,0 +1,176 @@ +"""Offline regression tests for uncertain target TempRadio cleanup.""" +import argparse +import contextlib +import io +import json +from pathlib import Path +import tempfile +import types +import unittest +from unittest import mock +import lora_ota as ota +import test_lora_ota as fixtures + +KEY='ab'*32 +NORMAL=ota.RadioSettings(910.525,62.5,7,5,False) + +class CleanupTests(unittest.TestCase): + def args(self): + return argparse.Namespace(target=KEY,temp_radio='909.5,500,7,5,180',temp_values=(909.5,500,7,5,180)) + def controller(self,mode='off'): + calls=[] + def remote(target,command,**kwargs): + self.assertEqual(target,KEY);self.assertIs(kwargs['retry'],False);calls.append(command) + if command.startswith('tempradio '): + if mode in ('lost','long'):raise ota.TransmissionError('lost cleanup ACK') + return 'OK - temp params for 1 mins' + if command=='get public.key': + return '> '+('cd'*32 if mode=='wrong-key' else KEY) + if command=='get tempradio': + if mode=='long':raise ota.TransmissionError('still only reachable on old temporary tuple') + if mode=='malformed':return '> unexpected' + return '> off' + raise AssertionError(command) + return types.SimpleNamespace(remote_command=remote,get_radio=lambda **kwargs:NORMAL,calls=calls) + def state(self,directory):return ota.TargetTempCleanup(Path(directory),self.args(),NORMAL) + def test_lost_ack_does_not_accept_temporary_identity_as_return(self): + controller=self.controller('lost') + with contextlib.redirect_stderr(io.StringIO()) as warning: + acknowledged=ota.shorten_target_temp_window(controller,self.args()) + self.assertFalse(acknowledged) + self.assertEqual(controller.calls,['tempradio 909.5,500,7,5,1']) + self.assertIn('original long lease may remain active',warning.getvalue()) + def test_attempt_is_persisted_before_send_and_never_replayed(self): + with tempfile.TemporaryDirectory() as directory: + state=self.state(directory);controller=self.controller('lost');remote=controller.remote_command + def checked(*args,**kwargs): + self.assertTrue(json.loads(state.path.read_text())['shorten_attempted']) + return remote(*args,**kwargs) + controller.remote_command=checked + with contextlib.redirect_stderr(io.StringIO()):state.shorten_once(controller,self.args()) + state.shorten_once(controller,self.args()) + self.assertEqual(len(controller.calls),1) + self.assertTrue(state.report['cleanup_required']) + self.assertFalse(state.report['shorten_acknowledged']) + self.assertEqual(state.path.stat().st_mode&0o777,0o600) + def test_positive_ack_still_requires_independent_normal_return(self): + with tempfile.TemporaryDirectory() as directory: + state=self.state(directory);controller=self.controller() + state.shorten_once(controller,self.args()) + self.assertTrue(state.report['shorten_acknowledged']) + self.assertTrue(state.report['cleanup_required']) + state.prove_normal(controller,self.args()) + self.assertFalse(state.report['cleanup_required']) + self.assertTrue(state.report['normal_return_verified']) + self.assertEqual(controller.calls,['tempradio 909.5,500,7,5,1','get public.key','get tempradio']) + def test_lost_ack_can_resolve_only_from_full_key_and_inactive_temp_on_normal(self): + with tempfile.TemporaryDirectory() as directory: + state=self.state(directory);controller=self.controller('lost') + with contextlib.redirect_stderr(io.StringIO()):state.shorten_once(controller,self.args()) + state.prove_normal(controller,self.args()) + self.assertTrue(state.report['normal_return_verified']) + self.assertFalse(state.report['shorten_acknowledged']) + self.assertEqual(len([x for x in controller.calls if x.startswith('tempradio ')]),1) + def test_wrong_key_and_malformed_status_are_not_retried(self): + for mode,expected_count in [('wrong-key',1),('malformed',2)]: + with self.subTest(mode=mode),tempfile.TemporaryDirectory() as directory: + state=self.state(directory);controller=self.controller(mode) + with self.assertRaises(ota.OtaError):state.prove_normal(controller,self.args()) + self.assertEqual(len(controller.calls),expected_count) + self.assertTrue(json.loads(state.path.read_text())['cleanup_required']) + def test_active_profile_is_not_mistaken_for_normal_return(self): + with tempfile.TemporaryDirectory() as directory: + state=self.state(directory);controller=self.controller();remote=controller.remote_command + clock=types.SimpleNamespace(now=100.0) + def sleep(seconds):clock.now+=seconds + fake_time=types.SimpleNamespace(monotonic=lambda:clock.now,sleep=sleep) + def active_until_return(target,command,**kwargs): + if command=='get tempradio' and clock.now<175: + controller.calls.append(command);return '> 909.500,500.000,7,5,32' + return remote(target,command,**kwargs) + controller.remote_command=active_until_return + with mock.patch.object(ota,'time',fake_time),contextlib.redirect_stdout(io.StringIO()): + state.prove_normal(controller,self.args()) + self.assertGreaterEqual(clock.now,175) + self.assertTrue(state.report['normal_return_verified']) + self.assertGreater(len(state.report['proof_records']),2) + def test_unresolved_long_lease_keeps_obligation_and_bounded_reads(self): + with tempfile.TemporaryDirectory() as directory: + state=self.state(directory);controller=self.controller('long');clock=types.SimpleNamespace(now=100.0) + def sleep(seconds):clock.now+=seconds + fake_time=types.SimpleNamespace(monotonic=lambda:clock.now,sleep=sleep) + remote=controller.remote_command + def bounded_loss(target,command,**kwargs): + if command=='get tempradio':clock.now+=min(20,kwargs['operation_timeout']) + return remote(target,command,**kwargs) + controller.remote_command=bounded_loss + with mock.patch.object(ota,'time',fake_time),contextlib.redirect_stdout(io.StringIO()),self.assertRaises(ota.TransmissionError): + state.prove_normal(controller,self.args()) + saved=json.loads(state.path.read_text()) + self.assertTrue(saved['cleanup_required']);self.assertFalse(saved['normal_return_verified']) + self.assertLessEqual(clock.now,190) + self.assertTrue(all(r['operation_timeout']<=190-r['started_at'] for r in saved['proof_records'])) + self.assertFalse(any(x.startswith('tempradio ') for x in controller.calls)) + def test_wrong_controller_channel_stops_before_target_read(self): + with tempfile.TemporaryDirectory() as directory: + state=self.state(directory);controller=self.controller() + controller.get_radio=lambda **kwargs:ota.RadioSettings(909.5,500,7,5,False) + with self.assertRaises(ota.OtaError):state.prove_normal(controller,self.args()) + self.assertEqual(controller.calls,[]) + self.assertTrue(state.report['cleanup_required']) + + def test_failed_main_keeps_obligation_after_lost_cleanup_ack(self): + controller=mock.Mock() + controller.get_radio.return_value=NORMAL + calls=[];clock=types.SimpleNamespace(now=100.0) + def remote(target,command,**kwargs): + self.assertEqual(target,KEY);self.assertIs(kwargs['retry'],False) + calls.append(command) + if command.startswith('tempradio '): + raise ota.TransmissionError('lost one-minute cleanup ACK') + self.assertEqual(command,'get public.key') + clock.now+=min(20,kwargs['operation_timeout']) + raise ota.TransmissionError('target remains on the original long lease') + controller.remote_command.side_effect=remote + def sleep(seconds):clock.now+=seconds + fake_time=types.SimpleNamespace(monotonic=lambda:clock.now,sleep=sleep,time=lambda:1800000000) + image=fixtures.firmware(b'cleanup main regression'*300,fixtures.VERSION_NEW) + package=ota.parse_mota(fixtures.mota_blob(image));seeder=mock.Mock() + with tempfile.TemporaryDirectory() as directory: + work=Path(directory)/'work' + argv=['release.mota',KEY,'--controller-serial','/dev/controller', + '--source-serial','/dev/source','--password','test-only', + '--work-dir',str(work),'--yes'] + error=io.StringIO() + with contextlib.ExitStack() as stack: + for name in ('preflight_inputs','preflight_source_cli','bind_contact_selectors', + 'ensure_source_clock_gate_safe','ensure_controller_clock_safe', + 'verify_shared_source_identity','confirm_update','run_temp_radio_preflight', + 'arm_target_temp_radio','switch_controller_to_temp_radio', + 'find_and_start_pull'): + stack.enter_context(mock.patch.object(ota,name)) + stack.enter_context(mock.patch.object(ota,'time',fake_time)) + stack.enter_context(mock.patch.object(ota,'read_source_rxps',return_value=ota.RxpsSettings(False,0,0))) + stack.enter_context(mock.patch.object(ota,'read_remote_rxps',return_value=ota.RxpsSettings(False,0,0))) + stack.enter_context(mock.patch.object(ota,'disable_source_rxps',return_value=False)) + stack.enter_context(mock.patch.object(ota,'read_lora_ota_participant_versions',return_value={})) + stack.enter_context(mock.patch.object(ota,'prepare_ota_hop_settings',return_value=[])) + stack.enter_context(mock.patch.object(ota,'query_target',return_value=fixtures.target())) + stack.enter_context(mock.patch.object(ota,'prepare_package',return_value=(Path('release.mota'),package,None))) + stack.enter_context(mock.patch.object(ota,'arm_source_temp_radio_once',return_value=True)) + stack.enter_context(mock.patch.object(ota,'shorten_source_temp_window',return_value=True)) + stack.enter_context(mock.patch.object(ota,'source_cli_command',return_value='OK')) + stack.enter_context(mock.patch.object(ota,'SeederProcess',return_value=seeder)) + stack.enter_context(mock.patch.object(ota,'monitor_download',side_effect=ota.OtaError('synthetic transfer failure'))) + stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + stack.enter_context(contextlib.redirect_stderr(error)) + result=ota.main(argv,controller_override=controller) + saved=json.loads((work/ota.TARGET_TEMP_RECOVERY_FILE).read_text()) + self.assertEqual(result,2) + self.assertEqual([c for c in calls if c.startswith('tempradio ')],['tempradio 909.5,500,5,5,1']) + self.assertTrue(saved['cleanup_required']);self.assertTrue(saved['shorten_attempted']) + self.assertFalse(saved['shorten_acknowledged']);self.assertFalse(saved['normal_return_verified']) + self.assertNotIn('ota self',calls) + self.assertIn('restoring the controller does not prove the destination returned',error.getvalue()) + +if __name__=='__main__':unittest.main()