diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index a8aaa533f..2b18b4712 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -59,6 +59,9 @@ jobs: - name: Verify old-firmware compatibility in both OTA launchers run: python3 -B -m unittest discover -s tools/lora_ota -p test_lora_ota_bootloader_version.py -v + - name: Verify OTA orchestration and automatic relay policy cleanup + run: python3 -B -m unittest discover -s tools/lora_ota -p 'test_lora_ota*.py' -v + - name: Verify all signed OTAFIX 2.4.6 release images env: MESHCORE_BOOTLOADER_MOTA_ZIP: ${{ runner.temp }}/OTAFIX-2.4.6-bootloader-mota.zip diff --git a/docs/lora_ota_automation.md b/docs/lora_ota_automation.md index 7050d36c6..f069f0450 100644 --- a/docs/lora_ota_automation.md +++ b/docs/lora_ota_automation.md @@ -665,6 +665,31 @@ matched repeater list in [the nRF52 QSPI guide](ota_nrf52_qspi.md). ## 6. Add intermediate relays +The runner automatically checks `ota config` on the destination, managed source, +and each OTA-enabled relay before confirmation. The declared `--relay` count is +the minimum required OTA reach: one relay requires at least one hop, not the +direct-only `hops=0`. After the radio rehearsal passes, it rechecks the original +policies and temporarily raises only limits below that minimum. Higher existing +limits remain unchanged. Every change needs exact readback; a lost setter reply +is reconciled by reading the policy, not blindly repeating the write. + +The original limits are restored after the download, before installation can +reboot the destination, and cleanup also attempts restoration after an error or +Ctrl-C. A serial seeder is stopped before the source text CLI is used. The +private, password-free `ota-hop-settings.json` recovery file is flushed before +any hop write. Hop limits are persisted by firmware: if the host loses power or +a radio becomes unreachable, use that file's original values once the correct +radio channel is reachable. Unexpected external policy changes are not overwritten. + +For a route containing additional, undeclared repeaters, add `--ota-hops N` +(0..8, at least the declared relay count). This is a minimum, not a command to +reduce larger limits. It does not discover or switch undeclared repeaters; +all participating radios still need overlapping TempRadio windows. Direct runs +without relays or a positive explicit minimum leave hop policies untouched. +Explicitly non-OTA relays remain opaque forwarders; unsupported/ambiguous policy +replies, authentication failures and timeouts on OTA participants stop the run. +No signer, install-security or forwarding-filter checks are bypassed. + List relays from farthest to nearest so each command is sent before its route moves to TempRadio. A bare relay name uses the destination password; use `NAME=PASSWORD` when it differs: @@ -929,7 +954,8 @@ download without replacing it. The working directory is created before a managed source can be changed, then retained and printed at exit. It contains the exact served mOTA, `motatool-serve.log`, extracted build inputs when needed, and -`controller-radio.txt`. A managed source also gets a protected +`controller-radio.txt`, plus `ota-hop-settings.json` when OTA reach needed a +temporary increase. A managed source also gets a protected `source-rxps-settings.json` containing its exact original preference and idempotent restore command. Its contents and directory entry are flushed before RXPS is disabled. When the destination diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index dc21e1956..40bd60433 100755 --- a/tools/lora_ota/lora_ota.py +++ b/tools/lora_ota/lora_ota.py @@ -144,6 +144,8 @@ INSTALL_RECONCILE_WAIT_SECONDS = 20 DEFAULT_RELAY_TX_DELAY = 0.3 RELAY_TIMING_COMMANDS_PER_RELAY = 12 RELAY_TIMING_RECOVERY_FILE = "relay-timing-settings.json" +OTA_HOPS_RECOVERY_FILE = "ota-hop-settings.json" +OTA_MAX_HOPS = 8 TARGET_RXPS_RECOVERY_FILE = "target-rxps-settings.json" SOURCE_RXPS_RECOVERY_FILE = "source-rxps-settings.json" MIN_MESHCLI_VERSION = (1, 6, 0) @@ -306,6 +308,19 @@ class RelayTimingSettings: txdelay: float +@dataclass(frozen=True) +class OtaHopSettings: + name: str + password: str | None + local_source: bool + original: int + minimum: int + + @property + def transfer(self) -> int: + return max(self.original, self.minimum) + + @dataclass(frozen=True) class RemoteClockSample: """One authenticated remote epoch with complete host-time uncertainty.""" @@ -1614,6 +1629,13 @@ def reply_matches_command(command_text: str, reply: str) -> bool: ) if command == "ota stats": return text.startswith("OTA | fw ") or is_unknown or needs_temp + if command == "ota config": + return ( + lowered.startswith("ota config:") or is_unknown or is_error + or re.search(r"\bota (?:is )?not included\b", lowered) is not None + ) + if command.startswith("ota config hops "): + return lowered.startswith("ok ota reach = ") or is_unknown or is_error if command == "get bootloader.ver": try: parse_bootloader_version_reply(text) @@ -2986,6 +3008,11 @@ def recommended_temp_radio_minutes( 0 if args.source_already_temp or args.source_shares_controller else 30 ) final_reply_count = (4 + relays) if include_install else 1 + hop_restore_seconds = 0 + if relays or (getattr(args, "ota_hops", None) or 0) > 0: + # A policy restore needs a current read, absolute setter, and readback + # on each remote participant plus the local source CLI. + hop_restore_seconds = 3 * ((1 + relays) * args.reply_timeout + 30) required_seconds = ( remote_setup_seconds + source_setup_seconds @@ -2996,6 +3023,7 @@ def recommended_temp_radio_minutes( + adaptive_poll_ceiling(args.poll_seconds) + args.reply_timeout * final_reply_count + relays * RELAY_TIMING_COMMANDS_PER_RELAY * args.reply_timeout + + hop_restore_seconds ) return required_seconds // 60 + 1 @@ -4336,6 +4364,167 @@ def parse_delay_reply(reply: str, label: str) -> float: return float(match.group(1)) +def parse_ota_hops(reply: str, label: str) -> int: + # A local serial terminal may include a welcome banner and command echo. + # Only the actual config line is authoritative, never another dotted or + # numeric value from that surrounding output. + lines = re.findall(r"(?mi)^[ \t]*(?:>[ \t]*)?ota config:[^\r\n]*", reply) + if len(lines) == 1: + values = re.findall(r"(? str: + if saved.local_source: + return source_cli_command(args, command, bounded=True, retry=retry) + + def once() -> str: + return controller.remote_command( + saved.name, command, password=saved.password, retry=False, + operation_timeout=TEMP_RADIO_PREFLIGHT_OPERATION_TIMEOUT_SECONDS, + ) + + if retry: + return retry_transmission_bounded(once, f"{command!r} on {saved.name}") + return once() + + +def read_ota_hops( + controller: Controller, args: argparse.Namespace, saved: OtaHopSettings, +) -> int: + return parse_ota_hops(ota_hops_command(controller, args, saved, "ota config"), saved.name) + + +def prepare_ota_hop_settings( + controller: Controller, args: argparse.Namespace, +) -> list[OtaHopSettings]: + """Read every policy before confirmation; never guess an unknown value.""" + minimum = max(len(args.relay_values), getattr(args, "ota_hops", None) or 0) + if minimum == 0: + return [] # Preserve direct-run behavior and existing user policy. + if minimum > OTA_MAX_HOPS: + raise OtaError(f"OTA supports at most {OTA_MAX_HOPS} relay hops") + if not has_managed_source_cli(args): + raise OtaError( + "automatic OTA hop setup requires a managed source CLI; supply " + "--source-cli-serial or --source-cli-tcp instead of --source-already-temp" + ) + source_name = ( + getattr(args, "shared_source_public_key", None) + or getattr(args, "source_contact_value", None) or "OTA source" + ) + participants = [ + OtaHopSettings(source_name, None, True, 0, minimum), + OtaHopSettings(args.target, None, False, 0, minimum), + *(OtaHopSettings(name, password, False, 0, minimum) + for name, password in args.relay_values), + ] + result = [] + for index, participant in enumerate(participants): + reply = ota_hops_command(controller, args, participant, "ota config") + # Non-OTA repeaters can forward OTA opaquely. Skip only an explicit + # absence of the build feature. An unknown command alone does not + # establish that this firmware has no OTA receive/forward policy. + if index >= 2 and re.fullmatch( + r"\s*(?:LoRa\s+)?OTA (?:is )?not included in this build" + r"(?:[;.:!][^\r\n]*)?\s*", + reply, re.IGNORECASE, + ): + print(f"[hops] {participant.name}: opaque relay; no OTA policy to change") + continue + result.append(replace(participant, original=parse_ota_hops(reply, participant.name))) + return result + + +def set_ota_hops_verified( + controller: Controller, args: argparse.Namespace, saved: OtaHopSettings, value: int, +) -> None: + try: + reply = ota_hops_command( + controller, args, saved, f"ota config hops {value}", retry=False, + ) + except TransmissionError: + # A missing acknowledgement is not permission to send another write. + # The absolute, idempotent setting is reconciled by an exact readback. + reply = None + if reply is not None and not re.fullmatch( + rf"\s*(?:>[ \t]*)?OK OTA reach = {value} hops? \(saved\)" + r"(?: - direct only)?\s*", reply, re.IGNORECASE, + ): + # Serial replies can carry a banner. Find exactly one complete ACK + # line, rather than accepting an OK from an unrelated command. + ack = re.findall( + rf"(?mi)^[ \t]*(?:>[ \t]*)?OK OTA reach = {value} hops? \(saved\)" + r"(?: - direct only)?[ \t]*\r?$", reply, + ) + if len(ack) != 1: + raise OtaError(f"{saved.name} refused OTA hop limit {value}: {reply}") + if read_ota_hops(controller, args, saved) != value: + raise OtaError(f"{saved.name} OTA hop limit did not read back as {value}") + + +def apply_ota_hop_settings( + controller: Controller, args: argparse.Namespace, settings: list[OtaHopSettings], + owned: list[OtaHopSettings], work_dir: Path, +) -> None: + if not settings: + return + # Rehearsal/preparation can take minutes. Validate ALL snapshots again + # before the first write, so another operator's change is not overwritten. + for saved in settings: + if read_ota_hops(controller, args, saved) != saved.original: + raise OtaError( + f"{saved.name} OTA hop limit changed during preparation; " + "leaving the external setting unchanged" + ) + changes = [saved for saved in settings if saved.transfer != saved.original] + if changes: + path = write_private_recovery_file(work_dir / OTA_HOPS_RECOVERY_FILE, json.dumps({ + "nodes": [{ + "name": saved.name, "local_source": saved.local_source, + "original_hops": saved.original, "transfer_hops": saved.transfer, + "restore_command": f"ota config hops {saved.original}", + } for saved in changes], + "source_endpoint": { + "serial": getattr(args, "source_cli_serial", None) or getattr(args, "source_serial", None), + "tcp": getattr(args, "source_cli_tcp", None), + }, + }, indent=2) + "\n") + print(f"[hops] recovery settings: {path}") + for saved in changes: + owned.append(saved) # Own cleanup BEFORE a possibly lost mutation. + set_ota_hops_verified(controller, args, saved, saved.transfer) + print(f"[hops] {saved.name}: temporarily {saved.original} -> {saved.transfer} (verified)") + + +def restore_ota_hop_settings( + controller: Controller, args: argparse.Namespace, owned: list[OtaHopSettings], +) -> None: + errors = [] + for saved in reversed(owned.copy()): + try: + current = read_ota_hops(controller, args, saved) + if current != saved.original: + if current != saved.transfer: + raise OtaError("OTA hop limit changed externally; leaving it unchanged") + set_ota_hops_verified(controller, args, saved, saved.original) + owned.remove(saved) # Retire ownership only after exact readback. + print(f"[hops] {saved.name}: original {saved.original} restored") + except (OtaError, OSError) as exc: + errors.append(f"{saved.name}: {exc}") + if errors: + raise OtaError("could not restore OTA hop settings: " + "; ".join(errors)) + + def read_relay_timing( controller: Controller, relay_name: str, @@ -4494,6 +4683,9 @@ def confirm_update( f" relay timing: rxdelay 0, " f"txdelay {format_decimal(args.relay_txdelay)} (saved/restored)" ) + for saved in getattr(args, "ota_hop_settings", []): + action = "unchanged" if saved.original == saved.transfer else "temporary; saved/restored" + print(f" OTA hops : {saved.name}: {saved.original} -> {saved.transfer} ({action})") print(f" action : {'stage only' if args.no_install else 'install and reboot'}") current_version = ( parse_version(target.current_version) if target.current_version else None @@ -7198,6 +7390,11 @@ def build_parser() -> argparse.ArgumentParser: "--relay", action="append", default=[], metavar="NAME_OR_KEY[=PASSWORD]", help="optional relay name/key/unique key prefix, farthest-to-nearest; repeat as needed", ) + parser.add_argument( + "--ota-hops", type=int, metavar="MINIMUM", + help=("minimum OTA reach for undeclared additional hops (0..8); otherwise " + "infer from --relay count; only raise lower limits and restore afterward"), + ) parser.add_argument( "--source-contact", metavar="NAME_OR_KEY", @@ -7387,6 +7584,10 @@ def validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) -> parser.error(f"--temp-radio: {exc}") if not math.isfinite(args.relay_txdelay) or not 0.0 <= args.relay_txdelay <= 2.0: parser.error("--relay-txdelay must be between 0 and 2") + if len(args.relay) > OTA_MAX_HOPS: + parser.error(f"at most {OTA_MAX_HOPS} --relay nodes are supported") + if args.ota_hops is not None and not len(args.relay) <= args.ota_hops <= OTA_MAX_HOPS: + parser.error("--ota-hops must be 0..8 and at least the declared --relay count") clear_manifest = args.clear_completed_manifest clear_body_hash = args.clear_completed_on_body_hash if bool(clear_manifest) != bool(clear_body_hash): @@ -8269,6 +8470,7 @@ def main( armed_relay_values: list[tuple[str, str]] = [] relay_public_keys: dict[str, str] = {} relay_timing_settings: list[RelayTimingSettings] = [] + owned_ota_hop_settings: list[OtaHopSettings] = [] password = args.password or os.environ.get("MESHCORE_ADMIN_PASSWORD", "") temp_command = f"tempradio {args.temp_radio}" @@ -8464,6 +8666,7 @@ def main( target_rxps_profile = None args.target_rxps_saved = target_rxps_saved args.target_rxps_profile = target_rxps_profile + args.ota_hop_settings = prepare_ota_hop_settings(controller, args) confirm_update(args, target, package) freq, bandwidth, sf, cr, _minutes = args.temp_values @@ -8482,6 +8685,9 @@ def main( original_radio, temp_radio, ) + apply_ota_hop_settings( + controller, args, args.ota_hop_settings, owned_ota_hop_settings, work_dir, + ) # Package generation above belongs on a workstation/build VM; a # Pi-class radio host should receive only the completed hash-verified @@ -8638,6 +8844,13 @@ def main( seeder.start() find_and_start_pull(controller, args, package, seeder) monitor_download(controller, args, package, seeder) + # A serial source is exclusively owned by motatool. Release that + # ownership before trying to restore its policy through the text CLI. + seeder.stop() + seeder = None + # Transfer is complete; restore while every participant is still on + # the same tuple, before installation can reboot the destination. + restore_ota_hop_settings(controller, args, owned_ota_hop_settings) if args.no_install: report_staged_update(controller, args, package) @@ -8677,8 +8890,6 @@ def main( controller, args.target, target_rxps_saved ) target_rxps_changed = False - seeder.stop() - seeder = None if source_temp_owned and ( not args.leave_controller_radio or not args.source_shares_controller @@ -8721,9 +8932,7 @@ def main( relay_public_keys, ) - # Stop seeding before returning the controller to its ordinary channel. - seeder.stop() - seeder = None + # Seeding already stopped before policy restoration and installation. if source_temp_owned and ( not args.leave_controller_radio or not args.source_shares_controller @@ -8808,6 +9017,11 @@ def main( file=sys.stderr, ) if controller_changed: + if owned_ota_hop_settings: + try: + restore_ota_hop_settings(controller, args, owned_ota_hop_settings) + except (OtaError, OSError) as exc: + print(f"WARNING: {exc}; retrying after radio restore", file=sys.stderr) if relay_timing_settings: try: restore_relay_timings(controller, relay_timing_settings) @@ -8931,6 +9145,15 @@ def main( f"{recovery_hint}: {exc}", file=sys.stderr, ) + if owned_ota_hop_settings and controller is not None: + try: + restore_ota_hop_settings(controller, args, owned_ota_hop_settings) + except (OtaError, OSError) as exc: + print( + f"CRITICAL: {exc}; use {work_dir / OTA_HOPS_RECOVERY_FILE} " + "after checking the current policy and radio channel", + file=sys.stderr, + ) if not source_rxps_changed: try: retire_local_source_rxps_recovery() diff --git a/tools/lora_ota/rak3401_mota_chain.py b/tools/lora_ota/rak3401_mota_chain.py index 98545ddd5..ac63bee78 100755 --- a/tools/lora_ota/rak3401_mota_chain.py +++ b/tools/lora_ota/rak3401_mota_chain.py @@ -2408,6 +2408,11 @@ def run_step( ]) if args.debug: command.append("--debug") + # The outer runner's destination policy alone cannot make a direct-only + # source/relay transport an indirect route. Carry the reach into the + # generic runner's temporary, saved/restored participant policy as well. + if getattr(args, "ota_hops", None) is not None: + command.extend(["--ota-hops", str(max(args.ota_hops, len(relays)))]) for relay_key, relay_password in relays: command.extend(["--relay", f"{relay_key}={relay_password}"]) result = ota.main(command, controller_override=controller) diff --git a/tools/lora_ota/test_lora_ota.py b/tools/lora_ota/test_lora_ota.py index 6c71da19d..c81066327 100644 --- a/tools/lora_ota/test_lora_ota.py +++ b/tools/lora_ota/test_lora_ota.py @@ -7252,6 +7252,7 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase): reboot_wait=90, relay=[], debug=False, + ota_hops=3, ) step = mock.Mock( number=1, @@ -7278,6 +7279,7 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase): ) command = nested.call_args.args[0] self.assertEqual(command.count("--debug"), int(enabled)) + self.assertEqual(command[command.index("--ota-hops") + 1], "3") self.assertEqual(command[1], target_key) self.assertEqual(command[command.index("--relay") + 1], f"{relay_key}=relay-password") diff --git a/tools/lora_ota/test_lora_ota_hops.py b/tools/lora_ota/test_lora_ota_hops.py new file mode 100644 index 000000000..5ff83e3ac --- /dev/null +++ b/tools/lora_ota/test_lora_ota_hops.py @@ -0,0 +1,438 @@ +"""Offline policy and lifecycle tests; no radio access or firmware writes.""" +import argparse +import contextlib +import io +import json +import os +from pathlib import Path +import tempfile +import unittest +from unittest import mock + +import lora_ota as ota +from test_lora_ota import firmware, mota_blob, target, VERSION_NEW + + +class Stations: + def __init__(self, relays=1, source_hops=0, target_hops=0): + self.hops = {"source": source_hops, "remote": target_hops, + **{f"relay{i}": 0 for i in range(relays)}} + self.calls = [] + self.events = [] + self.lost = set() + self.fail_reads = set() + self.overrides = {} + self.seeding = False + self.normal = ota.RadioSettings(910.525, 62.5, 7, 5, False) + + def command(self, name, command, **options): + self.calls.append((name, command, options)) + if name in self.fail_reads and command == "ota config": + raise ota.TransmissionError("lost config read") + if (name, command) in self.overrides: + return self.overrides[name, command] + if command == "ota folder off": + return "OK" + if command == "ota config": + return f"ota config: autofetch=off autoinstall=off hops={self.hops[name]} keys=0" + if command.startswith("ota config hops "): + if name == "source" and self.seeding: + raise AssertionError("source text CLI used while seeder owns serial") + value = int(command.rsplit(" ", 1)[1]) + self.hops[name] = value + self.events.append(f"set:{name}:{value}") + if (name, value) in self.lost: + self.lost.remove((name, value)) + raise ota.TransmissionError("lost setting acknowledgement") + return f"OK OTA reach = {value} hops (saved)" + (" - direct only" if value == 0 else "") + raise AssertionError((name, command)) + + def remote_command(self, name, command, **options): + return self.command(name, command, **options) + + def source_cli(self, _args, command, **options): + return self.command("source", command, **options) + + def get_radio(self): + return self.normal + + def set_radio(self, _radio, _label): + self.events.append("controller-restore") + + def close(self): + pass + + +def arguments(relays=1, minimum=None): + return argparse.Namespace( + target="remote", relay_values=[(f"relay{i}", "relay-secret") for i in range(relays)], + ota_hops=minimum, source_serial="/dev/source", source_cli_serial=None, + source_cli_tcp=None, source_contact_value="source", + ) + + +class OtaHopTests(unittest.TestCase): + def setUp(self): + self.output = contextlib.redirect_stdout(io.StringIO()) + self.output.__enter__() + self.addCleanup(self.output.__exit__, None, None, None) + self.sleep = mock.patch.object(ota.time, "sleep") + self.sleep.start() + self.addCleanup(self.sleep.stop) + + def test_config_parser_accepts_legacy_and_terminal_banners(self): + for hops in range(9): + reply = f"ota config: cache=off/0 autofetch=off advert=1440min hops={hops} keys=0" + self.assertEqual(ota.parse_ota_hops(reply, "station"), hops) + self.assertEqual(ota.parse_ota_hops(f"WELCOME\n> ota config\n> {reply}\r\n> ", "station"), hops) + + def test_parser_rejects_missing_ambiguous_malformed_and_error_values(self): + for reply in ("", "ERR permission denied", "Unknown command", "OTA | hops=0", + "ota config: hops=9", "ota config: hops=-1", "ota config: hops=1foo", + "ota config: hops=1 hops=1", "ota config: advert=1440", + "ota config: hops=1\nota config: hops=2"): + with self.subTest(reply=reply), self.assertRaises(ota.OtaError): + ota.parse_ota_hops(reply, "station") + + def test_command_reply_matching_does_not_accept_unrelated_ok(self): + self.assertTrue(ota.reply_matches_command("ota config", "ota config: hops=0")) + self.assertTrue(ota.reply_matches_command("ota config", "LoRa OTA not included in this build")) + self.assertTrue(ota.reply_matches_command("ota config", "LoRa OTA is not included in this build; tempradio cannot enable it.")) + self.assertTrue(ota.reply_matches_command("ota config hops 1", "OK OTA reach = 1 hop (saved)")) + self.assertFalse(ota.reply_matches_command("ota config hops 1", "OK - temp params for 1 mins")) + + def test_direct_run_has_no_policy_queries_or_writes(self): + stations = Stations(0) + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + self.assertEqual(ota.prepare_ota_hop_settings(stations, arguments(0)), []) + self.assertEqual(stations.calls, []) + + def test_infers_route_count_preserves_higher_limits_and_restores(self): + for count in (1, 2, 3, 8): + with self.subTest(relays=count), tempfile.TemporaryDirectory() as directory: + stations = Stations(count) + stations.hops["relay0"] = 8 + original = stations.hops.copy() + args = arguments(count) + owned = [] + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + settings = ota.prepare_ota_hop_settings(stations, args) + self.assertTrue(all(s.minimum == count for s in settings)) + ota.apply_ota_hop_settings(stations, args, settings, owned, Path(directory)) + self.assertEqual(stations.hops["remote"], count) + self.assertEqual(stations.hops["source"], count) + self.assertEqual(stations.hops["relay0"], 8) + ota.restore_ota_hop_settings(stations, args, owned) + self.assertEqual(stations.hops, original) + self.assertEqual(owned, []) + self.assertFalse(any(name == "relay0" and cmd.startswith("ota config hops ") + for name, cmd, _ in stations.calls)) + remote_writes = [options for name, cmd, options in stations.calls + if name != "source" and cmd.startswith("ota config hops ")] + self.assertTrue(all(options["retry"] is False for options in remote_writes)) + relay_options = [options for name, _cmd, options in stations.calls if name.startswith("relay")] + self.assertTrue(all(options["password"] == "relay-secret" for options in relay_options)) + + def test_explicit_minimum_handles_additional_undeclared_hops(self): + stations = Stations(0) + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + settings = ota.prepare_ota_hop_settings(stations, arguments(0, 3)) + self.assertEqual([s.transfer for s in settings], [3, 3]) + + def test_higher_limits_produce_no_recovery_file_or_changes(self): + stations = Stations(source_hops=4, target_hops=3) + stations.hops["relay0"] = 2 + with tempfile.TemporaryDirectory() as directory: + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + settings = ota.prepare_ota_hop_settings(stations, arguments()) + ota.apply_ota_hop_settings(stations, arguments(), settings, [], Path(directory)) + self.assertFalse((Path(directory) / ota.OTA_HOPS_RECOVERY_FILE).exists()) + self.assertFalse(stations.events) + + def test_unknown_source_or_relay_stops_before_any_setting_change(self): + for name in ("source", "remote", "relay0"): + for reply in ("Unknown command", "ERR permission denied", "ota config: hops=9", ""): + with self.subTest(name=name, reply=reply): + stations = Stations() + stations.overrides[name, "ota config"] = reply + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + with self.assertRaises(ota.OtaError): + ota.prepare_ota_hop_settings(stations, arguments()) + self.assertEqual(stations.events, []) + + def test_explicit_non_ota_relay_is_left_opaque(self): + stations = Stations() + stations.overrides["relay0", "ota config"] = "LoRa OTA is not included in this build; tempradio cannot enable it. Use an OTA-enabled repeater firmware" + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + settings = ota.prepare_ota_hop_settings(stations, arguments()) + self.assertEqual([s.name for s in settings], ["source", "remote"]) + + def test_setter_reply_lost_without_delivery_stops_without_replaying(self): + stations = Stations() + saved = ota.OtaHopSettings("remote", None, False, 0, 1) + with mock.patch.object(stations, "remote_command", side_effect=[ + ota.TransmissionError("not delivered"), "ota config: hops=0", + ]) as command: + with self.assertRaisesRegex(ota.OtaError, "did not read back"): + ota.set_ota_hops_verified(stations, arguments(), saved, 1) + self.assertEqual([c.args[1] for c in command.call_args_list], ["ota config hops 1", "ota config"]) + + def test_recovery_write_failure_prevents_any_mutation(self): + stations = Stations() + owned = [] + with tempfile.TemporaryDirectory() as directory: + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + settings = ota.prepare_ota_hop_settings(stations, arguments()) + with mock.patch.object(ota, "write_private_recovery_file", side_effect=OSError("disk full")): + with self.assertRaises(OSError): + ota.apply_ota_hop_settings(stations, arguments(), settings, owned, Path(directory)) + self.assertFalse(stations.events) + self.assertEqual(owned, []) + + def test_source_serial_ack_can_include_a_welcome_banner(self): + stations = Stations() + saved = ota.OtaHopSettings("source", None, True, 0, 1) + with mock.patch.object(ota, "source_cli_command", side_effect=[ + "WELCOME\n> ota config hops 1\nOK OTA reach = 1 hop (saved)\n> ", + "WELCOME\nota config: hops=1\n> ", + ]): + ota.set_ota_hops_verified(stations, arguments(), saved, 1) + + def test_unmanaged_source_and_excessive_route_are_rejected(self): + for args in (arguments(9), arguments()): + if len(args.relay_values) == 1: + args.source_serial = None + stations = Stations() + with self.assertRaises(ota.OtaError): + ota.prepare_ota_hop_settings(stations, args) + self.assertEqual(stations.calls, []) + + def test_timeout_has_finite_read_retries_and_never_means_zero(self): + stations = Stations() + stations.fail_reads.add("remote") + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + with self.assertRaises(ota.TransmissionError): + ota.prepare_ota_hop_settings(stations, arguments()) + self.assertEqual(len([c for c in stations.calls if c[0] == "remote"]), ota.TRANSMISSION_RETRY_LIMIT + 1) + self.assertEqual(stations.events, []) + + def test_external_change_during_rehearsal_aborts_all_writes(self): + stations = Stations() + with tempfile.TemporaryDirectory() as directory: + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + settings = ota.prepare_ota_hop_settings(stations, arguments()) + stations.hops["relay0"] = 4 + with self.assertRaisesRegex(ota.OtaError, "changed during preparation"): + ota.apply_ota_hop_settings(stations, arguments(), settings, [], Path(directory)) + self.assertFalse((Path(directory) / ota.OTA_HOPS_RECOVERY_FILE).exists()) + self.assertEqual(stations.events, []) + + def test_recovery_is_private_password_free_and_durable_before_writes(self): + stations = Stations() + args = arguments() + owned = [] + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + original_command = stations.command + + def command(name, text, **options): + if text.startswith("ota config hops "): + path = root / ota.OTA_HOPS_RECOVERY_FILE + self.assertTrue(path.is_file()) + self.assertTrue(any(saved.name == name for saved in owned)) + raw = path.read_text() + self.assertNotIn("relay-secret", raw) + self.assertIn("restore_command", raw) + if os.name != "nt": + self.assertEqual(path.stat().st_mode & 0o777, 0o600) + return original_command(name, text, **options) + + with mock.patch.object(stations, "command", side_effect=command): + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + settings = ota.prepare_ota_hop_settings(stations, args) + ota.apply_ota_hop_settings(stations, args, settings, owned, root) + saved = json.loads((root / ota.OTA_HOPS_RECOVERY_FILE).read_text()) + self.assertEqual(saved["source_endpoint"]["serial"], "/dev/source") + + def test_lost_set_and_restore_acknowledgements_use_readback_without_replay(self): + stations = Stations() + stations.lost = {("source", 1), ("remote", 1), ("relay0", 1), + ("source", 0), ("remote", 0), ("relay0", 0)} + with tempfile.TemporaryDirectory() as directory: + owned = [] + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + settings = ota.prepare_ota_hop_settings(stations, arguments()) + ota.apply_ota_hop_settings(stations, arguments(), settings, owned, Path(directory)) + ota.restore_ota_hop_settings(stations, arguments(), owned) + self.assertEqual(owned, []) + self.assertEqual(len(stations.events), 6) + self.assertTrue(all(value == 0 for value in stations.hops.values())) + + def test_bad_readback_is_not_success_and_partial_changes_are_restored(self): + stations = Stations() + stations.overrides["remote", "ota config hops 1"] = "OK OTA reach = 1 hop (saved)" + with tempfile.TemporaryDirectory() as directory: + owned = [] + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + settings = ota.prepare_ota_hop_settings(stations, arguments()) + with self.assertRaisesRegex(ota.OtaError, "did not read back"): + ota.apply_ota_hop_settings(stations, arguments(), settings, owned, Path(directory)) + self.assertEqual([s.name for s in owned], ["source", "remote"]) + ota.restore_ota_hop_settings(stations, arguments(), owned) + self.assertEqual(owned, []) + self.assertEqual(stations.hops["source"], 0) + + def test_restore_attempts_all_nodes_and_preserves_external_changes(self): + stations = Stations() + with tempfile.TemporaryDirectory() as directory: + owned = [] + with mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli): + settings = ota.prepare_ota_hop_settings(stations, arguments()) + ota.apply_ota_hop_settings(stations, arguments(), settings, owned, Path(directory)) + stations.hops["relay0"] = 4 + with self.assertRaisesRegex(ota.OtaError, "changed externally"): + ota.restore_ota_hop_settings(stations, arguments(), owned) + self.assertEqual(stations.hops, {"source": 0, "remote": 0, "relay0": 4}) + self.assertEqual([s.name for s in owned], ["relay0"]) + + def test_cli_minimum_must_cover_declared_route_and_stay_in_range(self): + parser = ota.build_parser() + for extra in (["--ota-hops", "-1"], ["--ota-hops", "9"], + ["--relay", "relay", "--ota-hops", "0"], + ["--relay", "relay"] * 9): + with self.subTest(extra=extra), contextlib.redirect_stderr(io.StringIO()): + args = parser.parse_args(["release.mota", "remote", *extra]) + with self.assertRaises(SystemExit): + ota.validate_args(args, parser) + + +class OtaHopLifecycleTests(unittest.TestCase): + def run_main(self, directory, *, failure=None, leave=False, no_install=True): + stations = Stations() + image = firmware(b"ota hop lifecycle" * 300, VERSION_NEW) + package = ota.parse_mota(mota_blob(image)) + seeder = mock.Mock() + + def start(): + stations.seeding = True + stations.events.append("seeder-start") + + def stop(): + stations.seeding = False + stations.events.append("seeder-stop") + + seeder.start.side_effect = start + seeder.stop.side_effect = stop + + def phase(name): + def action(*_args, **_kwargs): + stations.events.append(name) + if failure == name: + if name == "interrupt": + raise KeyboardInterrupt + raise ota.OtaError("synthetic " + name) + return action + + def pull(*_args): + stations.events.append("pull") + if failure == "interrupt": + raise KeyboardInterrupt + if failure == "pull": + raise ota.OtaError("synthetic discovery failure") + self.assertTrue(all(hops == 1 for hops in stations.hops.values())) + + def install(*_args): + stations.events.append("install") + self.assertTrue(all(hops == 0 for hops in stations.hops.values())) + if failure == "install": + raise ota.OtaError("synthetic install failure") + return True + + argv = ["release.mota", "remote", "--controller-serial", "/dev/controller", + "--source-serial", "/dev/source", "--password", "secret", + "--relay", "relay0=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: + 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", + "shorten_target_temp_window", "enforce_relay_timing", "report_staged_update", + "wait_for_post_install_identity", "verify_installed"): + stack.enter_context(mock.patch.object(ota, name)) + stack.enter_context(mock.patch.object(ota, "ensure_source_clock_gate_safe", return_value=(1800000000, 1800000059))) + stack.enter_context(mock.patch.object(ota, "read_source_rxps", return_value=ota.RxpsSettings(False, 100, 100))) + stack.enter_context(mock.patch.object(ota, "disable_source_rxps", return_value=False)) + stack.enter_context(mock.patch.object(ota, "query_target", return_value=target())) + stack.enter_context(mock.patch.object(ota, "prepare_package", return_value=(Path("release.mota"), package, None))) + stack.enter_context(mock.patch.object(ota, "read_lora_ota_participant_versions", return_value={})) + stack.enter_context(mock.patch.object(ota, "read_remote_rxps", return_value=ota.RxpsSettings(False, 100, 100))) + stack.enter_context(mock.patch.object(ota, "confirm_update", side_effect=phase("confirm"))) + stack.enter_context(mock.patch.object(ota, "run_temp_radio_preflight", side_effect=phase("rehearse"))) + stack.enter_context(mock.patch.object(ota, "source_cli_command", side_effect=stations.source_cli)) + stack.enter_context(mock.patch.object(ota, "arm_target_temp_radio", side_effect=phase("arm-target"))) + 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_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)) + stack.enter_context(mock.patch.object(ota, "find_and_start_pull", side_effect=pull)) + stack.enter_context(mock.patch.object(ota, "monitor_download", side_effect=phase("monitor"))) + stack.enter_context(mock.patch.object(ota, "request_install", side_effect=install)) + stack.enter_context(mock.patch.object(ota.time, "sleep")) + stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + stack.enter_context(contextlib.redirect_stderr(io.StringIO())) + # The production source name is a bound public key; this test uses + # the descriptive fallback while the source CLI fixture is local. + original_prepare = ota.prepare_ota_hop_settings + + def prepare(controller, args): + args.source_contact_value = "source" + return original_prepare(controller, args) + + stack.enter_context(mock.patch.object(ota, "prepare_ota_hop_settings", side_effect=prepare)) + result = ota.main(argv, controller_override=stations) + return result, stations + + def test_success_restores_after_seeder_stop_even_when_temp_radio_is_preserved(self): + for leave in (False, True): + with self.subTest(leave=leave), tempfile.TemporaryDirectory() as directory: + result, stations = self.run_main(directory, leave=leave) + self.assertEqual(result, 0) + self.assertEqual(stations.hops, {"source": 0, "remote": 0, "relay0": 0}) + events = stations.events + self.assertLess(events.index("confirm"), events.index("rehearse")) + self.assertLess(events.index("rehearse"), events.index("set:source:1")) + self.assertLess(events.index("set:remote:1"), events.index("arm-target")) + self.assertLess(events.index("seeder-stop"), events.index("set:source:0")) + + 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: + result, stations = self.run_main(directory, failure=failure) + self.assertEqual(result, 2) + self.assertFalse(any(event.startswith("set:") for event in stations.events)) + + def test_discovery_failure_and_interrupt_restore_all_original_settings(self): + for failure in ("pull", "interrupt", "arm-target"): + with self.subTest(failure=failure), tempfile.TemporaryDirectory() as directory: + result, stations = self.run_main(directory, failure=failure) + self.assertEqual(result, 130 if failure == "interrupt" else 2) + self.assertEqual(stations.hops, {"source": 0, "remote": 0, "relay0": 0}) + + def test_install_runs_only_after_policies_are_restored(self): + for failure in (None, "install"): + with self.subTest(failure=failure), tempfile.TemporaryDirectory() as directory: + result, stations = self.run_main(directory, failure=failure, no_install=False) + self.assertEqual(result, 2 if failure else 0) + self.assertEqual(stations.hops, {"source": 0, "remote": 0, "relay0": 0}) + self.assertLess(stations.events.index("set:source:0"), stations.events.index("install")) + + +if __name__ == "__main__": + unittest.main()