diff --git a/docs/lora_ota_automation.md b/docs/lora_ota_automation.md index bdef28e2..5f591db2 100644 --- a/docs/lora_ota_automation.md +++ b/docs/lora_ota_automation.md @@ -77,6 +77,83 @@ mOTA ownership state as nRF52. The exact `ota folder on` line transfers the USB port from startup ASCII or idle Binary Companion mode before binary mOTA frames begin; neither console mode shares that port until the folder detaches. +## Bootloader packages and station identifiers + +Both wrappers use the same package-aware runner. It identifies application +versus bootloader updates from container metadata, not filenames. A ZIP mixing +application inputs and bootloader mOTAs requires `--zip-member`; their versions +and target IDs are independent and must not be ranked together. + +Signed format-3 nRF52 bootloader packages can be **staged with `--no-install`**. +Use the same controller/source/TempRadio arguments as an application transfer, +but provide the bootloader `.mota` (or its ZIP member) and `--no-install`. +Without that flag, even `--yes` stops before opening a radio. Bootloader staging +requires Python `cryptography` in the interpreter running the wrapper and a +current `motatool` supporting format 3. The runner reuses `tools/mota/motalib.py` +for strict geometry, embedded identity/CRC, capability, hash and signature +validation, and still runs `motatool verify` including any `--public-key` pin. +It probes actual `motatool inspect` support before radio access: older and newer +builds may advertise the same version. If the tool is missing or incompatible, +an interactive run offers an automatic repair: + +```text +Install this bootloader-capable motatool and continue after verification? [y/N] +``` + +Only `y` or `yes` approves installation. **`--yes` does not approve host software +installation.** Noninteractive runs stop and print a copyable manual installation +command plus the `--motatool` argument to use afterward. + +The repair builds the [mikecarper/motatool](https://github.com/mikecarper/motatool) +fork at pinned revision `8c38369e7d35ad50cf74261869676d52dd24adf7`, with Cargo's +`--locked` dependencies, in a separate per-user cache directory. It needs +Rust/Cargo and a native linker already installed; otherwise it points to +[Rust installation](https://rustup.rs/) and stops without installing anything. +Source/dependency downloads, build time and disk use are disclosed before +confirmation. Cargo progress is displayed, with a one-hour build timeout. +The radio admin-password environment variable is not passed to the build. + +The installer does not overwrite the selected/system `motatool` or alter PATH. +It may replace only a previous copy in its displayed private install directory. +After installation it repeats the bootloader-package capability check and uses +the new binary for this run only if that check passes. A later run using the +default `motatool` selection can reuse the checked cache without downloading +again; switching from an explicit `--motatool` path still asks permission. +Refusal, a failed build, or a failed recheck stops before opening any radio. +Failed build files may remain in the displayed cache directory for diagnostics. + +The install root is `meshcore-lora-ota/motatool/` under `%LOCALAPPDATA%` +on Windows, `~/Library/Caches` on macOS, or `$XDG_CACHE_HOME` (default +`~/.cache`) on Linux. Cargo uses an explicit +[`--root`, `--rev`, and `--locked`](https://doc.rust-lang.org/cargo/commands/cargo-install.html) +instead of replacing a global installation or following a moving branch. + +Before transfer it queries the destination's `ota bootloader status`, checks its +bootloader-specific target/hardware identity, installed ABI/codecs and matching +internal/QSPI/SD storage profile. It does not compare an application version or +application target ID to a bootloader package. After transfer it checks the +destination's exact staged MID and image-hash confirmation, then prints the +manual command; it **never sends a bootloader install command**: + +```text +ota bootloader status +ota bootloader install +``` + +Review and send those commands on the **destination**, not the source. The +device independently enforces signer authorization, safe live storage, +continuity and upgrade-only policy at installation. `--base`, +`--allow-non-upgrade`, and `--prepare-only` remain application-only. See the +[bootloader prerequisites and recovery limits](ota_nrf52_bootloader_update.md). + +`TARGET_NODE`, `--relay`, and `--source-contact` accept a contact name, full +public key, or unique hexadecimal key prefix. The runner reads the controller's +existing contact table and binds each selection to a full key before remote +commands, so emoji names need not be typed and later name changes cannot +redirect the run. Duplicate names, ambiguous prefixes, or the same radio under +different participant aliases stop with an actionable error. Missing contacts +must first be imported or discovered; a key alone does not create a contact. + ## Destination requirements | Destination | Package installed | One-time prerequisite | Raw ZIP handling | @@ -559,7 +636,7 @@ Useful controls: If the version gate required RXPS off, it stays off while that topology is preserved; use `target-rxps-settings.json` to restore it only after sending `normalradio`. -- `--allow-non-upgrade` deliberately permits the same or an older version. +- `--allow-non-upgrade` deliberately permits the same or an older application version. - `--replace-active-download` deliberately discards a different update already downloading or staged on the target. Without it, that update is preserved. - `--source-shares-controller` is for a Full Companion whose USB Binary API is diff --git a/docs/ota_nrf52_bootloader_update.md b/docs/ota_nrf52_bootloader_update.md index b54c2410..9a66c9f1 100644 --- a/docs/ota_nrf52_bootloader_update.md +++ b/docs/ota_nrf52_bootloader_update.md @@ -197,6 +197,13 @@ closed instead of selecting smaller or overlapping geometry. ## Explicit install workflow +The `tools/lora_ota/lora_ota.sh` and `.ps1` runners can perform the discovery, +transfer, verification and radio cleanup for a bootloader mOTA with +`--no-install`. They detect format 3 automatically and check the destination's +bootloader-specific identity/capabilities. They do not install it: the final +MID/hash confirmation below remains an explicit operator action. A mixed +application/bootloader ZIP requires `--zip-member`. + Check the installed identity and capability marker: ```text diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index 6d97a06a..f8c51916 100755 --- a/tools/lora_ota/lora_ota.py +++ b/tools/lora_ota/lora_ota.py @@ -17,6 +17,7 @@ import argparse import atexit import getpass import hashlib +import importlib.util import json import math import os @@ -35,8 +36,10 @@ import sys import tempfile import threading import time -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import datetime, timezone +from functools import lru_cache +from types import ModuleType from typing import Callable, TypeVar import zipfile @@ -138,6 +141,11 @@ RELAY_TIMING_RECOVERY_FILE = "relay-timing-settings.json" TARGET_RXPS_RECOVERY_FILE = "target-rxps-settings.json" SOURCE_RXPS_RECOVERY_FILE = "source-rxps-settings.json" 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. +MOTATOOL_REPAIR_REPOSITORY = "https://github.com/mikecarper/motatool.git" +MOTATOOL_REPAIR_REVISION = "8c38369e7d35ad50cf74261869676d52dd24adf7" +MOTATOOL_REPAIR_TIMEOUT_SECONDS = 60 * 60 # v1.17.1.5 is the first release-version contract in which every packet, # including retries, uses the same tuple-selected physical preamble: normally # 32 symbols at SF5-SF8, then 64 or 128 only where each shorter choice cannot @@ -193,6 +201,7 @@ class MotaInfo: hw_id: str base_hash: bytes payload_offset: int + bootloader_storage: int | None = None @property def is_full(self) -> bool: @@ -200,8 +209,14 @@ class MotaInfo: @property def kind(self) -> str: + if self.is_bootloader: + return "bootloader full" return "full" if self.is_full else "delta" + @property + def is_bootloader(self) -> bool: + return bool(self.flags & MOTA_FLAG_BOOTLOADER) + @property def version(self) -> str: return format_version(self.fw_version) @@ -233,6 +248,9 @@ class TargetInfo: nrf_qspi: bool = False # Firmware predating the live maxblk marker had 1 KiB receive buffers. max_block_size: int = LEGACY_TARGET_MAX_BLOCK_SIZE + boot_target_id: int | None = None + boot_hw_id: str | None = None + boot_storage: int | None = None @property def nrf_external(self) -> bool: @@ -810,6 +828,52 @@ def merkle_root(leaves: list[bytes]) -> bytes: return level[0] +@lru_cache(maxsize=1) +def bootloader_library() -> ModuleType: + """Reuse the repository's strict v3 reference validator, not a second codec.""" + path = Path(__file__).resolve().parents[1] / "mota" / "motalib.py" + name = "_meshcore_lora_ota_motalib" + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise OtaError(f"bootloader validator is unavailable: {path}") + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module # dataclasses resolves annotations through this registry + try: + spec.loader.exec_module(module) + except (OSError, ImportError) as exc: + sys.modules.pop(name, None) + raise OtaError(f"cannot load bootloader validator: {exc}") from exc + return module + + +def parse_bootloader_mota(blob: bytes, path: Path | None) -> MotaInfo: + library = bootloader_library() + try: + parsed = library.parse_container(blob) + problems = library.verify(parsed) + if problems: + raise ValueError("; ".join(problems)) + except ImportError as exc: + raise OtaError( + "bootloader signature verification requires Python cryptography; " + "install it in the Python environment running this script" + ) from exc + except (ValueError, struct.error) as exc: + raise OtaError(f"invalid bootloader mOTA: {exc}") from exc + manifest = parsed.manifest + return MotaInfo( + path=path, blob=blob, flags=manifest.flags, + target_id=manifest.target_id, fw_version=manifest.fw_version, + image_size=manifest.image_size, payload_size=manifest.payload_size, + block_size=manifest.block_size, merkle_root=manifest.merkle_root, + image_hash=manifest.image_hash, codec_id=manifest.codec_id, + hw_id=manifest.hw_id.rstrip(b"\0").decode("ascii"), + base_hash=manifest.base_hash, + payload_offset=len(blob) - len(MOTA_TRAILER) - manifest.payload_size, + bootloader_storage=library.bootloader_caps_storage(parsed.payload), + ) + + def parse_mota(blob: bytes, path: Path | None = None) -> MotaInfo: if len(blob) < 8 + MOTA_FIXED_MANIFEST_SIZE + len(MOTA_TRAILER): raise OtaError("mOTA is truncated") @@ -821,17 +885,14 @@ def parse_mota(blob: bytes, path: Path | None = None) -> MotaInfo: f"mOTA size field is {declared_size}, but the file is {len(blob)} bytes" ) flags = blob[9] - if blob[8] == MOTA_BOOT_FORMAT_VERSION and flags & MOTA_FLAG_BOOTLOADER: - raise OtaError( - "bootloader mOTA packages require the device's explicit `ota bootloader install` " - "workflow; this application-update runner deliberately refuses them" - ) + if blob[10] != 0x12: + raise OtaError(f"unsupported mOTA hash algorithm 0x{blob[10]:02x}") + if blob[8] == MOTA_BOOT_FORMAT_VERSION: + return parse_bootloader_mota(blob, path) if blob[8] != MOTA_FORMAT_VERSION: raise OtaError(f"unsupported mOTA format version {blob[8]}") if flags & MOTA_FLAG_BOOTLOADER or flags & ~MOTA_KNOWN_FLAGS: raise OtaError("invalid flags in v2 application mOTA") - if blob[10] != 0x12: - raise OtaError(f"unsupported mOTA hash algorithm 0x{blob[10]:02x}") target_id, fw_version, image_size, payload_size = struct.unpack_from(" bytes: def compatible_mota(info: MotaInfo, target: TargetInfo) -> tuple[bool, str]: + if info.is_bootloader: + if target.platform != "nrf52" or target.boot_target_id is None: + return False, "destination has no verified `ota bootloader status` capability" + if info.target_id != target.boot_target_id or info.hw_id != target.boot_hw_id: + return False, ( + f"bootloader target {info.target_id:08X} hw={info.hw_id}, need " + f"{target.boot_target_id:08X} hw={target.boot_hw_id}" + ) + if target.bootloader_abi is None or target.bootloader_abi < 3: + return False, "bootloader self-update requires installed ABI 3 or newer" + if target.bootloader_codecs != 0x5: + return False, "bootloader self-update requires FULL|INPLACE codecs 0x5" + expected_storage = 0x09 if target.nrf_sd else 0x0E if target.nrf_qspi else 0x0A + if ( + target.boot_storage != expected_storage + or info.bootloader_storage != expected_storage + ): + return False, "bootloader package, installed capabilities, and application storage differ" + if info.block_size > target.max_block_size: + return False, f"bootloader block exceeds destination maxblk:{target.max_block_size}" + return True, "" if info.target_id != target.target_id: return False, f"target {info.target_id:08X}, need {target.target_id:08X}" if info.hw_id and target.hw_id and info.hw_id != target.hw_id: @@ -1052,6 +1134,8 @@ def select_mota_from_zip( archive: zipfile.ZipFile, target: TargetInfo, requested_member: str | None, + *, + package_kind: str = "application", ) -> tuple[MotaInfo, str] | None: candidates: list[tuple[MotaInfo, str]] = [] rejected: list[str] = [] @@ -1062,6 +1146,9 @@ def select_mota_from_zip( continue try: info = parse_mota(read_zip_member(archive, member)) + if info.is_bootloader != (package_kind == "bootloader"): + rejected.append(f"{member.filename}: different update type; select it explicitly") + continue good, reason = compatible_mota(info, target) if good: candidates.append((info, member.filename)) @@ -1070,7 +1157,9 @@ def select_mota_from_zip( except (OtaError, zipfile.BadZipFile) as exc: rejected.append(f"{member.filename}: {exc}") if not candidates: - if requested_member and requested_member.lower().endswith(".mota"): + if package_kind == "bootloader" or ( + requested_member and requested_member.lower().endswith(".mota") + ): details = "; ".join(rejected) or "member not found" raise OtaError(f"requested ZIP mOTA is unusable: {details}") return None @@ -1144,8 +1233,8 @@ def load_base_image(path: Path, target: TargetInfo) -> EndFInfo: info = parse_mota( read_bounded_file(path, MAX_ARCHIVE_MEMBER_SIZE, "mOTA file"), path ) - if not info.is_full: - raise OtaError("--base mOTA must be a full-image container") + if info.is_bootloader or not info.is_full: + raise OtaError("--base mOTA must be a full application-image container") identity = parse_endf(info.payload) elif suffix == ".zip": identities: list[tuple[EndFInfo, str]] = [] @@ -1156,7 +1245,7 @@ def load_base_image(path: Path, target: TargetInfo) -> EndFInfo: try: if suffix == ".mota": candidate = parse_mota(read_zip_member(archive, member)) - if not candidate.is_full: + if candidate.is_bootloader or not candidate.is_full: continue candidate_identity = parse_endf(candidate.payload) elif suffix in (".bin", ".hex"): @@ -1339,6 +1428,10 @@ def prepare_package( source = args.package.resolve() if not source.is_file(): raise OtaError(f"package does not exist: {source}") + package_kind = getattr(args, "package_kind", None) or inspect_package_kind( + source, args.zip_member + ) + require_package_action(args, package_kind == "bootloader") served_dir = work_dir / "served" served_dir.mkdir(parents=True, exist_ok=False) selected: MotaInfo | None = None @@ -1354,7 +1447,8 @@ def prepare_package( try: with zipfile.ZipFile(source) as archive: mota_member = select_mota_from_zip( - archive, target, args.zip_member + archive, target, args.zip_member, + package_kind=package_kind, ) if mota_member is not None: selected, member_name = mota_member @@ -1371,6 +1465,9 @@ def prepare_package( raise OtaError("PACKAGE must be a .mota or .zip file") if selected is not None: + if selected.is_bootloader != (package_kind == "bootloader"): + raise OtaError("package type changed after preflight; restart with the intended input") + require_package_action(args, selected.is_bootloader) good, reason = compatible_mota(selected, target) if not good: raise OtaError(f"package is not installable on {target.name}: {reason}") @@ -1437,7 +1534,7 @@ def prepare_package( verify_with_motatool(args.motatool, output, args.public_key) expected_body_hash: bytes | None = None - if selected.is_full: + if selected.is_full and not selected.is_bootloader: try: expected_body_hash = parse_endf(selected.payload).body_hash except OtaError: @@ -1478,6 +1575,14 @@ def reply_matches_command(command_text: str, reply: str) -> bool: needs_temp = lowered.startswith("lora ota needs temp radio") if command == "ota status": return text.startswith("OTA |") or "not included" in lowered or is_unknown + if command in ("ota bootloader", "ota bootloader status"): + return ( + text.startswith("BL board=") + or lowered.startswith("bootloader update unavailable:") + or is_unknown + or needs_temp + or (is_error and "bootloader" in lowered) + ) if command == "ota self": return ( lowered.startswith("self ") @@ -2048,9 +2153,7 @@ class Controller: contact_keys = [ value["public_key"].lower() for value in json_objects(after) - if value.get("adv_name") == target - and isinstance(value.get("public_key"), str) - and re.fullmatch(r"[0-9A-Fa-f]{64}", value["public_key"]) + if contact_matches_selector(value, target) ] if contact_keys != [expected_public_key.lower()]: raise OtaError( @@ -2114,9 +2217,7 @@ class Controller: contact_keys = [ value["public_key"].lower() for value in post_objects - if value.get("adv_name") == target - and isinstance(value.get("public_key"), str) - and re.fullmatch(r"[0-9A-Fa-f]{64}", value["public_key"]) + if contact_matches_selector(value, target) ] if len(contact_keys) != 1: if any( @@ -2267,16 +2368,15 @@ class Controller: for item in objects ): raise OtaError(f"controller has no contact named {target!r}") - target_key = None - for item in objects: - if ( - item.get("adv_name") == target - and isinstance(item.get("public_key"), str) - ): - target_key = item["public_key"].lower() - break - if target_key is None: + target_keys = { + item["public_key"].lower() for item in objects + if contact_matches_selector(item, target) + } + if len(target_keys) > 1: + raise OtaError(f"ambiguous contact identity for {target}; use a full public key") + if not target_keys: raise TransmissionError(f"meshcli did not return contact identity for {target}") + target_key = target_keys.pop() messages = [ item for item in post_objects @@ -2396,6 +2496,111 @@ def parse_target_max_block_size(status: str, self_status: str) -> int: return value +def contact_matches_selector(contact: dict, selector: str) -> bool: + key = contact.get("public_key") + if not isinstance(key, str) or not re.fullmatch(r"[0-9A-Fa-f]{64}", key): + return False + name = contact.get("adv_name") + return ( + isinstance(name, str) and name.casefold() == selector.casefold() + ) or ( + re.fullmatch(r"[0-9A-Fa-f]{1,64}", selector) is not None + and key.lower().startswith(selector.lower()) + ) + + +def bind_contact_selectors(controller: Controller, args: argparse.Namespace) -> None: + """Resolve the local contact table once, before any remote command is sent. + + meshcli already accepts keys. Pin unique selectors to complete keys so names + with emojis, renames, duplicate names, and prefix collisions cannot redirect + maintenance halfway through a run. Reading contacts sends no LoRa packets. + """ + objects = controller._run(["contacts"], "resolve OTA station identifiers") + contacts: dict[str, dict] = {} + for obj in objects: + for key, value in obj.items(): + if ( + isinstance(value, dict) + and isinstance(key, str) + and re.fullmatch(r"[0-9A-Fa-f]{64}", key) + and isinstance(value.get("public_key"), str) + and value["public_key"].lower() == key.lower() + ): + contacts[key.lower()] = value + + def resolve(selector: str) -> str: + matches = [ + key for key, value in contacts.items() + if contact_matches_selector(value, selector) + ] + if not matches: + raise OtaError( + f"controller has no contact matching {selector!r}; import or discover it first" + ) + if len(matches) != 1: + raise OtaError( + f"ambiguous station {selector!r}: {len(matches)} contacts match; " + "use a full public key" + ) + key = matches[0] + print(f"[contact] {selector} -> {contacts[key].get('adv_name', '?')} [{key}]") + return key + + target = resolve(args.target) + relays = [(resolve(name), password) for name, password in args.relay_values] + source_selector = args.source_contact_value + if ( + not source_selector and not args.source_shares_controller + and has_managed_source_cli(args) + ): + source_selector = read_source_name_bounded(args) + source = resolve(source_selector) if source_selector else None + keys = [target, *(key for key, _password in relays)] + if source: + keys.append(source) + if len(keys) != len(set(keys)): + raise OtaError("destination, relays, and source contact must identify different radios") + args.target = target + args.relay_values = relays + args.source_contact_value = source + + +def query_bootloader_target( + controller: Controller, target: TargetInfo, +) -> TargetInfo: + reply = controller.remote_command(target.name, "ota bootloader status") + match = re.fullmatch( + r"BL board=([0-9A-Fa-f]{8}) target=([0-9A-Fa-f]{8}) name=(\S+) " + r"crc=([0-9A-Fa-f]{8}) abi=(\d+) caps=([0-9A-Fa-f]{2}) " + r"\| staged:(?:ready|none) mid=(?:-|[0-9A-Fa-f]{8}) hash=(?:-|[0-9A-Fa-f]{16})", + reply.strip(), + ) + if match is None: + raise OtaError( + f"destination cannot stage a bootloader update: {reply}; " + "requires exact-board ABI-3 self-update bootloader and capable application" + ) + board, target_id, name, _crc, abi, caps = match.groups() + library = bootloader_library() + try: + hw_id = ( + library.bootloader_hw_id(int(board, 16), name) + .rstrip(b"\0").decode("ascii") + ) + derived_target = library.bootloader_target_id(int(board, 16), name) + except (ValueError, UnicodeError) as exc: + raise OtaError(f"invalid destination bootloader identity: {exc}") from exc + if derived_target != int(target_id, 16): + raise OtaError("destination bootloader board/name does not match its target ID") + if int(abi) < 3 or int(abi) != target.bootloader_abi: + raise OtaError("destination reports inconsistent or unsupported bootloader ABI") + return replace( + target, boot_target_id=derived_target, boot_hw_id=hw_id, + boot_storage=int(caps, 16), + ) + + def query_target( controller: Controller, args: argparse.Namespace, @@ -2502,7 +2707,7 @@ def query_target( ) current_version = format_version(version_value) current_version_source = "ver" - return TargetInfo( + target = TargetInfo( name=args.target, target_id=target_id, base_hash=base_hash, @@ -2519,6 +2724,9 @@ def query_target( nrf_qspi=nrf_qspi, max_block_size=max_block_size, ) + if getattr(args, "package_kind", "application") == "bootloader": + return query_bootloader_target(controller, target) + return target def parse_temp_radio(value: str) -> tuple[float, float, int, int, int]: @@ -3942,6 +4150,7 @@ def confirm_update( target: TargetInfo, package: MotaInfo, ) -> None: + require_package_action(args, package.is_bootloader) print("\nValidated update plan:") print(f" destination : {target.name} ({target.target_id:08X}, {target.platform})") print(f" running base: {target.base_hash.hex().upper()}") @@ -3956,6 +4165,9 @@ def confirm_update( print(" bootloader : not required") print(f" update : {package.version} {package.kind} hw={package.hw_id or '?'}") print(f" mOTA id : {package.manifest_id}") + if package.is_bootloader: + print(f" boot target : {package.target_id:08X}; stage only, explicit install required") + print(" boot safety : destination rechecks trust, continuity and upgrade-only policy at manual install") print(f" TempRadio : {args.temp_radio}") saved_rxps = getattr(args, "target_rxps_saved", None) rxps_profile = getattr(args, "target_rxps_profile", None) @@ -3982,7 +4194,10 @@ def confirm_update( current_version = ( parse_version(target.current_version) if target.current_version else None ) - if current_version is not None and current_version >= package.fw_version: + if ( + not package.is_bootloader and current_version is not None + and current_version >= package.fw_version + ): print( f" warning : destination reports {target.current_version}; " f"package is {package.version}" @@ -6189,6 +6404,11 @@ def request_install( package: MotaInfo, ) -> bool: """Request install without ever blindly replaying an uncertain command.""" + if package.is_bootloader: + raise OtaError( + "bootloader installation requires explicit " + "`ota bootloader install `; this runner only stages it" + ) cycle_started = time.monotonic() retries = 0 confirm_ready_to_install(controller, args, package) @@ -6637,7 +6857,7 @@ def build_parser() -> argparse.ArgumentParser: ), ) parser.add_argument("package", type=Path, metavar="PACKAGE", help=".mota or .zip") - parser.add_argument("target", metavar="TARGET_NODE", help="destination contact name") + parser.add_argument("target", metavar="TARGET_NODE", help="destination contact name, full public key, or unique key prefix") controller = parser.add_mutually_exclusive_group() controller.add_argument("--controller-serial", metavar="PORT") controller.add_argument("--controller-tcp", metavar="HOST[:PORT]") @@ -6661,14 +6881,14 @@ def build_parser() -> argparse.ArgumentParser: help="destination admin password (prefer the MESHCORE_ADMIN_PASSWORD environment variable)", ) parser.add_argument( - "--relay", action="append", default=[], metavar="NAME[=PASSWORD]", - help="optional relay, ordered farthest-to-nearest; repeat as needed", + "--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( "--source-contact", - metavar="NAME", + metavar="NAME_OR_KEY", help=( - "controller contact for a separate OTA source, used for the " + "controller contact name/key/unique key prefix for a separate OTA source, used for the " "three-minute on-air proof; defaults to the source's local name " "(no remote-admin password is required)" ), @@ -6711,7 +6931,13 @@ def build_parser() -> argparse.ArgumentParser: help=argparse.SUPPRESS, ) parser.add_argument("--meshcli", default="meshcli") - parser.add_argument("--motatool", default="motatool") + parser.add_argument( + "--motatool", default="motatool", + help=( + "host packaging tool; missing/incompatible bootloader support offers " + "a separately confirmed private install (--yes does not approve it)" + ), + ) parser.add_argument( "--package-build-timeout", type=int, @@ -6761,7 +6987,7 @@ def build_parser() -> argparse.ArgumentParser: "--leave-controller-radio", action="store_true", help="leave the controller on --temp-radio instead of restoring it", ) - parser.add_argument("--no-install", action="store_true", help="download and verify, but do not install") + parser.add_argument("--no-install", action="store_true", help="download and verify, but do not install (required for bootloader mOTA)") parser.add_argument( "--replace-active-download", action="store_true", help="discard a different update already staged on the destination", @@ -7015,11 +7241,272 @@ def require_meshcli_version(command: str) -> tuple[int, int, int]: return version +def require_package_action(args: argparse.Namespace, bootloader: bool) -> None: + if not bootloader: + return + if not getattr(args, "no_install", False): + raise OtaError( + "detected a bootloader mOTA, not an application update; use --no-install " + "to stage it. Installation remains the device's explicit " + "`ota bootloader install ` workflow" + ) + if getattr(args, "prepare_only", False): + raise OtaError( + "bootloader staging requires live destination capability checks; " + "--prepare-only is application-only" + ) + if getattr(args, "base", None) or getattr(args, "allow_non_upgrade", False): + raise OtaError( + "--base and --allow-non-upgrade are application-only; " + "bootloader continuity and upgrade policy cannot be overridden" + ) + + +def inspect_package_kind(source: Path, requested_member: str | None) -> str: + """Choose the update family before opening radios; never rank app vs BL versions.""" + if source.suffix.lower() == ".mota": + info = parse_mota( + read_bounded_file(source, MAX_ARCHIVE_MEMBER_SIZE, "mOTA file"), source + ) + return "bootloader" if info.is_bootloader else "application" + kinds: set[str] = set() + found = False + try: + with zipfile.ZipFile(source) as archive: + for member in archive.infolist(): + if member.is_dir() or ( + requested_member and member.filename != requested_member + ): + continue + if requested_member and found: + raise OtaError(f"ZIP member name is duplicated: {requested_member}") + found = True + suffix = Path(member.filename).suffix.lower() + if suffix == ".mota": + # Classification is deliberately cheap and conservative. Full + # validation (including signatures for BL) follows selection. + # A malformed boot declaration must never fall back to an app. + with archive.open(member) as stream: + header = stream.read(10) + is_boot = len(header) == 10 and header[:4] == MOTA_MAGIC and ( + header[8] == MOTA_BOOT_FORMAT_VERSION + or header[9] & MOTA_FLAG_BOOTLOADER + ) + kinds.add("bootloader" if is_boot else "application") + elif suffix in (".bin", ".hex"): + kinds.add("application") + except (OSError, RuntimeError, zipfile.BadZipFile) as exc: + raise OtaError(f"cannot inspect ZIP archive {source}: {exc}") from exc + if requested_member and not found: + raise OtaError(f"requested ZIP member not found: {requested_member}") + if len(kinds) > 1: + raise OtaError( + "ZIP contains both application and bootloader inputs; choose the intended " + "update with --zip-member (their versions and target IDs are unrelated)" + ) + return next(iter(kinds), "application") + + +def report_staged_update( + controller: Controller, args: argparse.Namespace, package: MotaInfo, +) -> None: + if not package.is_bootloader: + print(f"{args.target} is ready to install; leaving the verified update staged.") + return + reply = controller.remote_command(args.target, "ota bootloader status") + match = re.search( + r"\btarget=([0-9A-Fa-f]{8}) .*\| staged:ready " + r"mid=([0-9A-Fa-f]{8}) hash=([0-9A-Fa-f]{16})$", reply.strip(), + ) + if match is None or ( + int(match[1], 16) != package.target_id + or match[2].upper() != package.manifest_id + or match[3].lower() != package.image_hash[:8].hex() + ): + raise OtaError(f"destination did not confirm the exact staged bootloader MID/hash: {reply}") + print(f"[staged] bootloader {package.version} on {args.target}; NOT installed") + print("[manual] Review `ota bootloader status` on the destination, then explicitly send:") + print(f" ota bootloader install {package.manifest_id} {package.image_hash[:8].hex().upper()}") + print("[manual] Do not use `ota install`. The device still enforces signer trust, continuity and upgrade-only checks.") + + +def motatool_repair_root() -> Path: + if os.name == "nt": + cache = Path(os.environ.get("LOCALAPPDATA") or Path.home() / "AppData" / "Local") + elif sys.platform == "darwin": + cache = Path.home() / "Library" / "Caches" + else: + cache = Path(os.environ.get("XDG_CACHE_HOME") or Path.home() / ".cache") + if not cache.is_absolute(): + raise OtaError("motatool repair needs an absolute user cache directory") + return cache / "meshcore-lora-ota" / "motatool" / MOTATOOL_REPAIR_REVISION + + +def motatool_repair_command(root: Path, cargo: str = "cargo") -> list[str]: + return [ + cargo, "install", "--git", MOTATOOL_REPAIR_REPOSITORY, + "--rev", MOTATOOL_REPAIR_REVISION, "--locked", "--bin", "motatool", + "--root", str(root), "--target-dir", str(root / "build"), "--force", + ] + + +def display_host_command(command: list[str]) -> str: + if os.name == "nt": + return "& " + " ".join("'" + word.replace("'", "''") + "'" for word in command) + return shlex.join(command) + + +def check_bootloader_tool(command: str, probe: Path) -> None: + run_checked( + [command, "inspect", str(probe)], + label="check motatool bootloader support", timeout=30, + ) + + +def offer_motatool_repair( + args: argparse.Namespace, probe: Path, failure: OtaError, +) -> None: + """Install only after separate consent; never replace the selected/PATH binary.""" + root = motatool_repair_root() + binary = root / "bin" / ("motatool.exe" if os.name == "nt" else "motatool") + print(f"[host] {args.motatool} cannot inspect this format-3 bootloader package: {failure}") + cached = False + if binary.is_file(): + try: + check_bootloader_tool(str(binary), probe) + cached = True + except OtaError as exc: + print(f"[host] cached motatool also failed its capability check: {exc}") + if cached and args.motatool == "motatool": + args.motatool = str(binary) + print(f"[host] using previously installed, rechecked motatool: {binary}") + return + + cargo = shutil.which("cargo") + command = motatool_repair_command(root, cargo or "cargo") + manual = display_host_command(command) + selection = display_host_command(["--motatool", str(binary)]) + # Display an argument, not a PowerShell call operator for an option. + if os.name == "nt": + selection = selection.removeprefix("& ") + print(f"[repair] source: {MOTATOOL_REPAIR_REPOSITORY}") + print(f"[repair] pinned revision: {MOTATOOL_REPAIR_REVISION}") + print(f"[repair] private install: {root}") + print("[repair] Existing motatool installations and PATH will not be changed.") + print(f"[repair] Manual install command: {manual}") + print(f"[repair] To select this build explicitly, append: {selection}") + if not cached and cargo is None: + raise OtaError( + "automatic motatool repair requires Rust/Cargo and a native linker; " + "install the toolchain from https://rustup.rs and rerun, or select a " + "bootloader-capable build with --motatool. No installation was attempted" + ) + if not sys.stdin.isatty(): + raise OtaError( + "motatool repair needs separate interactive approval (--yes does not approve " + "software installation); rerun in a terminal or use the printed manual " + "command and --motatool. No installation was attempted" + ) + if cached: + question = "Use the checked cached motatool for this run instead? [y/N] " + else: + print( + "[repair] This downloads and compiles the pinned source and locked dependencies. " + "It can take several minutes and use substantial disk space; " + "only a previous copy in the private install folder may be replaced." + ) + question = "Install this bootloader-capable motatool and continue after verification? [y/N] " + try: + approved = input(question).strip().lower() in ("y", "yes") + except EOFError: + approved = False + if not approved: + raise OtaError("motatool repair declined; no installation or radio changes were made") + + if not cached: + # No cache directories, downloads or builds before affirmative consent. + root.mkdir(parents=True, exist_ok=True) + print("[repair] Building motatool (Cargo output follows)...", flush=True) + build_env = os.environ.copy() + build_env.pop("MESHCORE_ADMIN_PASSWORD", None) + try: + result = subprocess.run( + command, cwd=str(root), stdin=subprocess.DEVNULL, + timeout=MOTATOOL_REPAIR_TIMEOUT_SECONDS, check=False, + env=build_env, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise OtaError( + f"motatool repair failed: {exc}; existing installations are unchanged. " + f"Build files may remain in {root}; fix the toolchain/network and rerun" + ) from exc + if result.returncode != 0: + raise OtaError( + f"motatool repair failed (Cargo exit {result.returncode}); see the build " + f"output above. Existing installations are unchanged; build files remain in {root}" + ) + if not binary.is_file(): + raise OtaError(f"motatool installer returned success but did not produce {binary}") + # Recheck even after Cargo success; do not silently fall back to the old binary. + try: + check_bootloader_tool(str(binary), probe) + except OtaError as exc: + raise OtaError( + f"repaired motatool still failed bootloader verification: {exc}; " + "stopping before radio access" + ) from exc + args.motatool = str(binary) + print(f"[repair] bootloader support verified; continuing with {binary}") + + +def require_bootloader_tool_support(args: argparse.Namespace) -> None: + """Probe actual format support: old and new motatool builds can share a version.""" + if args.package.suffix.lower() == ".mota": + probe = args.package + temporary_blob = None + else: + temporary_blob = None + failures: list[str] = [] + try: + with zipfile.ZipFile(args.package) as archive: + for member in archive.infolist(): + if not member.filename.lower().endswith(".mota") or ( + args.zip_member and member.filename != args.zip_member + ): + continue + try: + info = parse_mota(read_zip_member(archive, member)) + if info.is_bootloader: + temporary_blob = info.blob + break + except OtaError as exc: + failures.append(f"{member.filename}: {exc}") + except zipfile.BadZipFile as exc: + raise OtaError(f"invalid ZIP archive: {args.package}") from exc + if temporary_blob is None: + raise OtaError("ZIP contains no valid bootloader mOTA: " + "; ".join(failures)) + + # This is a local parser probe, never a serve/pull operation. Do not require + # every hardware variant in an archive to share the operator's signer pin; + # the selected package is independently verified with that pin later. + with tempfile.TemporaryDirectory(prefix="meshcore-boot-probe-") as directory: + if temporary_blob is not None: + probe = Path(directory) / "bootloader.mota" + probe.write_bytes(temporary_blob) + try: + check_bootloader_tool(args.motatool, probe) + except OtaError as exc: + offer_motatool_repair(args, probe, exc) + + def preflight_inputs(args: argparse.Namespace) -> None: if not args.package.is_file(): raise OtaError(f"package does not exist: {args.package.resolve()}") if args.package.suffix.lower() not in (".zip", ".mota"): raise OtaError("PACKAGE must be a .zip or .mota file") + args.package_kind = inspect_package_kind(args.package, args.zip_member) + print(f"[package] detected {args.package_kind} update from container metadata") + require_package_action(args, args.package_kind == "bootloader") for label, path in ( ("--base", args.base), ("--sign-key", args.sign_key), @@ -7027,7 +7514,10 @@ def preflight_inputs(args: argparse.Namespace) -> None: ): if path is not None and not path.is_file(): raise OtaError(f"{label} file does not exist: {path.resolve()}") - require_command(args.motatool, "motatool") + if args.package_kind == "bootloader": + require_bootloader_tool_support(args) + else: + require_command(args.motatool, "motatool") if not args.prepare_only: require_meshcli_version(args.meshcli) @@ -7252,6 +7742,7 @@ def main( ) if controller is None: controller = Controller(args, password) + bind_contact_selectors(controller, args) verify_shared_source_identity(controller, args) # These gates are intentionally before query_target() or any other # on-air remote operation. First prove that advancing the managed @@ -7519,7 +8010,7 @@ def main( monitor_download(controller, args, package, seeder) if args.no_install: - print(f"{args.target} is ready to install; leaving the verified update staged.") + report_staged_update(controller, args, package) restore_relay_timings(controller, relay_timing_settings) relay_timing_settings.clear() if args.leave_controller_radio: diff --git a/tools/lora_ota/test_lora_ota.py b/tools/lora_ota/test_lora_ota.py index 42f20d28..e6190d4c 100644 --- a/tools/lora_ota/test_lora_ota.py +++ b/tools/lora_ota/test_lora_ota.py @@ -143,11 +143,11 @@ class FormatTests(unittest.TestCase): with self.assertRaisesRegex(ota.OtaError, "block hashes"): ota.parse_mota(bytes(blob)) - def test_bootloader_packages_are_explicitly_refused(self) -> None: + def test_relabelling_application_does_not_make_a_bootloader(self) -> None: blob = bytearray(mota_blob(firmware(b"B" * 5000, VERSION_NEW))) blob[8] = ota.MOTA_BOOT_FORMAT_VERSION blob[9] |= ota.MOTA_FLAG_SIGNED | ota.MOTA_FLAG_BOOTLOADER - with self.assertRaisesRegex(ota.OtaError, "bootloader mOTA packages"): + with self.assertRaisesRegex(ota.OtaError, "invalid bootloader mOTA"): ota.parse_mota(bytes(blob)) blob[8] = ota.MOTA_FORMAT_VERSION @@ -836,6 +836,13 @@ class DebugTests(unittest.TestCase): class SourceCliTests(unittest.TestCase): + def setUp(self) -> None: + # These lifecycle tests stub the controller and source transports. + # Station resolution itself is covered by StationSelectionTests. + patcher = mock.patch.object(ota, "bind_contact_selectors") + patcher.start() + self.addCleanup(patcher.stop) + @staticmethod def source_args() -> argparse.Namespace: return argparse.Namespace( @@ -4426,6 +4433,7 @@ class TempRadioPreflightTests(unittest.TestCase): ] with ( mock.patch.object(ota, "preflight_inputs"), + mock.patch.object(ota, "bind_contact_selectors"), mock.patch.object(ota, "preflight_source_cli"), mock.patch.object( ota, diff --git a/tools/lora_ota/test_lora_ota_bootloader.py b/tools/lora_ota/test_lora_ota_bootloader.py new file mode 100644 index 00000000..4aed554a --- /dev/null +++ b/tools/lora_ota/test_lora_ota_bootloader.py @@ -0,0 +1,457 @@ +"""Offline bootloader staging and station-selection regression tests. No radios.""" + +import argparse +import contextlib +from dataclasses import replace +import io +import os +from pathlib import Path +import shutil +import struct +import tempfile +import unittest +from unittest import mock +import warnings +import zipfile +import zlib + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +import lora_ota as ota +from test_lora_ota import firmware, mota_blob, prepare_args, target, VERSION_NEW + + +def boot_blob(*, storage=0x0A, version=0x0117010D): + """Build a signed test-only candidate using the canonical reference builder.""" + ml = ota.bootloader_library() + board, name, app_base = 0x239A0029, "3401_DFU", ml.NRF52_APP_BASE_S140_V6 + if storage == 0x0E: + board, name, app_base = 0x28860044, "XIAO-SENSE-DFU", ml.NRF52_APP_BASE_S140_V7 + name = ml.XIAO_BOOT_DEVICE_NAME.rstrip(b"\0").decode("ascii") + elif storage == 0x09: + board, name = 0x239A0071, "TOWER_V2_OTA" + image = bytearray(b"\xff" * ml.XIAO_BOOT_IMAGE_SIZE) + struct.pack_into(" 0.11.0-OTAFIX2.4.6", + "ota self": "self body=100 image=156 base_hash=0011223344556677 | bootloader: abi=3 codecs=0x5", + "ota stats": "OTA | fw v1.17.1.5", + "ota bootloader status": ( + "BL board=239A0029 target=23818A80 name=3401_DFU " + "crc=12345678 abi=3 caps=0A | staged:none mid=- hash=-" + ), + } + for kind in ("application", "bootloader"): + controller = mock.Mock() + controller.remote_command.side_effect = lambda _target, command, **_kwargs: replies[command] + with contextlib.redirect_stdout(io.StringIO()): + result = ota.query_target(controller, argparse.Namespace(target="remote", package_kind=kind)) + if kind == "bootloader": + self.assertEqual(result.boot_target_id, self.package.target_id) + self.assertEqual(ota.compatible_mota(self.package, result), (True, "")) + else: + self.assertIsNone(result.boot_target_id) + self.assertNotIn(mock.call("remote", "ota bootloader status"), controller.remote_command.call_args_list) + + def test_application_version_is_not_compared_to_bootloader_version(self): + args = ota.build_parser().parse_args(["boot.mota", "remote", "--no-install", "--yes"]) + live = replace(boot_target(self.package), current_version="v99.0.0") + with contextlib.redirect_stdout(io.StringIO()): + ota.confirm_update(args, live, self.package) + + def test_boot_status_reply_filter(self): + for reply in ( + "BL board=239A0029 target=23818A80", "Bootloader update unavailable: bad CRC", + "ERR this build cannot update its bootloader over LoRa", + ): + self.assertTrue(ota.reply_matches_command("ota bootloader status", reply)) + self.assertFalse(ota.reply_matches_command("ota bootloader status", "OTA | no download | target:1234ABCD")) + + @unittest.skipUnless(os.environ.get("MOTATOOL_TEST_BIN") or shutil.which("motatool"), "motatool is not installed") + def test_real_motatool_verifies_bootloader_and_rejects_wrong_signer_pin(self): + tool = os.environ.get("MOTATOOL_TEST_BIN") or "motatool" + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + package = root / "boot.mota" + package.write_bytes(self.blob) + pin = root / "signer.pub" + pin.write_text(self.blob[105:137].hex(), encoding="ascii") + ota.require_bootloader_tool_support(argparse.Namespace(package=package, motatool=tool)) + archive_path = root / "boot.zip" + with zipfile.ZipFile(archive_path, "w") as archive: + archive.writestr("boot.mota", self.blob) + ota.require_bootloader_tool_support(argparse.Namespace( + package=archive_path, zip_member="boot.mota", motatool=tool, + )) + ota.verify_with_motatool(tool, package, pin) + pin.write_text("02" * 32, encoding="ascii") + with self.assertRaises(ota.OtaError): + ota.verify_with_motatool(tool, package, pin) + + def test_old_motatool_stops_before_any_radio_access(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "boot.mota" + path.write_bytes(self.blob) + args = ota.build_parser().parse_args([str(path), "remote", "--no-install"]) + with ( + mock.patch.object(ota, "require_command"), + mock.patch.object(ota, "require_meshcli_version") as meshcli, + mock.patch.object(ota, "run_checked", side_effect=ota.OtaError("unsupported format_ver 3")), + mock.patch.object(ota, "motatool_repair_root", return_value=Path(directory) / "cache"), + mock.patch.object(ota.shutil, "which", return_value="cargo"), + mock.patch.object(ota.sys.stdin, "isatty", return_value=False), + self.assertRaisesRegex(ota.OtaError, "separate interactive approval"), + ): + ota.preflight_inputs(args) + meshcli.assert_not_called() + + def test_simulated_main_stages_bootloader_and_restores_radios(self): + package = self.package + destination_key, source_key = "a1" * 32, "b2" * 32 + controller = mock.Mock() + controller._run.return_value = [{ + destination_key: {"adv_name": "remote", "public_key": destination_key}, + 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 + + def remote_command(station, command, **_kwargs): + self.assertEqual(station, destination_key) + replies = { + "ota status": "OTA | no download | target:1234ABCD", + "ota ls": "Updates 1", + f"ota pull {package.manifest_id} flash": f"OK pulling mid={package.manifest_id} -> flash", + "ota bootloader status": ( + "BL board=239A0029 target=23818A80 name=3401_DFU crc=12345678 abi=3 caps=0A " + f"| staged:ready mid={package.manifest_id} hash={package.image_hash[:8].hex()}" + ), + } + if command not in replies: + self.fail(f"unexpected remote command: {command}") + return replies[command] + + controller.remote_command.side_effect = remote_command + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source = root / "boot.mota" + source.write_bytes(self.blob) + seeder = mock.Mock() + with contextlib.ExitStack() as stack: + for name in ( + "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", + ): + stack.enter_context(mock.patch.object(ota, name)) + stack.enter_context(mock.patch.object(ota, "read_source_name_bounded", return_value="source")) + 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, "query_target", return_value=replace(boot_target(package), name=destination_key), + )) + stack.enter_context(mock.patch.object(ota, "read_lora_ota_participant_versions", return_value={})) + rehearsal = stack.enter_context(mock.patch.object(ota, "run_temp_radio_preflight")) + stack.enter_context(mock.patch.object(ota, "arm_source_temp_radio_once", return_value=True)) + stack.enter_context(mock.patch.object(ota, "SeederProcess", return_value=seeder)) + stack.enter_context(mock.patch.object(ota, "monitor_download")) + cleanup = stack.enter_context(mock.patch.object(ota, "shorten_target_temp_window")) + stack.enter_context(mock.patch.object(ota, "shorten_source_temp_window", return_value=True)) + install = stack.enter_context(mock.patch.object(ota, "request_install")) + stack.enter_context(mock.patch.object(ota.time, "sleep")) + output = stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + error = stack.enter_context(contextlib.redirect_stderr(io.StringIO())) + result = ota.main([ + str(source), destination_key[:12], "--controller-serial", "controller", + "--source-serial", "source", "--password", "test-only", + "--no-install", "--yes", "--work-dir", str(root / "work"), + ], controller_override=controller) + self.assertEqual(result, 0, error.getvalue()) + self.assertIn("detected bootloader update", output.getvalue()) + self.assertIn("NOT installed", output.getvalue()) + 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") + install.assert_not_called() + self.assertIn( + mock.call(destination_key, f"ota pull {package.manifest_id} flash", retry=False), + controller.remote_command.call_args_list, + ) + + +class StationSelectionTests(unittest.TestCase): + key = "3ee21f453f8f" + "ab" * 26 + name = "Ashport \U0001f4e1" + + def args(self, selector): + return argparse.Namespace( + target=selector, relay_values=[], source_contact_value=None, + source_shares_controller=True, + ) + + def controller(self): + controller = mock.Mock() + controller._run.return_value = [{self.key: {"public_key": self.key, "adv_name": self.name}}] + return controller + + def test_names_full_keys_and_prefixes_bind_to_one_full_key(self): + for selector in (self.name, self.name.upper(), self.key, self.key.upper(), self.key[:12]): + with self.subTest(selector=selector): + args = self.args(selector) + controller = self.controller() + with contextlib.redirect_stdout(io.StringIO()): + ota.bind_contact_selectors(controller, args) + self.assertEqual(args.target, self.key) + controller._run.assert_called_once_with(["contacts"], "resolve OTA station identifiers") + controller.remote_command.assert_not_called() + + def test_missing_and_ambiguous_identifiers_are_refused(self): + controller = self.controller() + with self.assertRaisesRegex(ota.OtaError, "no contact"): + ota.bind_contact_selectors(controller, self.args("missing")) + second = self.key[:-2] + "cd" + controller._run.return_value[0][second] = {"public_key": second, "adv_name": self.name} + for selector in (self.name, self.key[:12]): + with self.assertRaisesRegex(ota.OtaError, "ambiguous"): + ota.bind_contact_selectors(controller, self.args(selector)) + + def test_same_radio_cannot_be_target_and_relay_under_aliases(self): + args = self.args(self.name) + args.relay_values = [(self.key[:12], "unused")] + with contextlib.redirect_stdout(io.StringIO()), self.assertRaisesRegex(ota.OtaError, "different radios"): + ota.bind_contact_selectors(self.controller(), args) + + def test_remote_reply_accepts_key_selector_with_emoji_name(self): + for selector in (self.name, self.key, self.key[:12]): + controller = object.__new__(ota.Controller) + controller.reply_timeout = 5 + controller._authenticated_targets = {selector} + controller._run_marked = mock.Mock(return_value=( + [{"adv_name": self.name, "public_key": self.key}], + [{"txt_type": 1, "text": "OTA | no download | target:1234ABCD", "pubkey_prefix": self.key[:12]}], + )) + with contextlib.redirect_stdout(io.StringIO()): + self.assertIn("OTA", controller._remote_command_once(selector, "ota status", "unused")) + + def test_clock_and_ack_accept_key_selector(self): + controller = object.__new__(ota.Controller) + controller._execute = mock.Mock(side_effect=lambda commands, _label: argparse.Namespace(stdout=( + commands[1] + '\n{"adv_name": "renamed", "public_key": "' + self.key + '"}\n1800000000\n' + ))) + self.assertEqual(controller.get_contact_clock(self.key, self.key), 1800000000) + controller._run_marked = mock.Mock(return_value=([], [ + {"adv_name": "renamed", "public_key": self.key}, + {"expected_ack": "12345678"}, {"code": "12345678"}, + ])) + controller.prove_contact_ack(self.key, self.key, "proof") + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/lora_ota/test_lora_ota_tool_repair.py b/tools/lora_ota/test_lora_ota_tool_repair.py new file mode 100644 index 00000000..edc13b0d --- /dev/null +++ b/tools/lora_ota/test_lora_ota_tool_repair.py @@ -0,0 +1,206 @@ +"""Automatic host-tool repair tests. All installer execution is mocked.""" + +import argparse +import contextlib +import io +import os +from pathlib import Path +import subprocess +import tempfile +import unittest +from unittest import mock + +import lora_ota as ota +from test_lora_ota_bootloader import boot_blob + + +class MotatoolRepairTests(unittest.TestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.folder = Path(directory.name) + self.root = self.folder / "private-install" + self.binary = self.root / "bin" / ("motatool.exe" if os.name == "nt" else "motatool") + self.probe = self.folder / "boot.mota" + self.probe.write_bytes(boot_blob()) + self.args = argparse.Namespace(motatool="old-motatool", yes=True, package=self.probe) + self.failure = ota.OtaError("unsupported format_ver 3") + stack = contextlib.ExitStack() + self.addCleanup(stack.close) + stack.enter_context(mock.patch.object(ota, "motatool_repair_root", return_value=self.root)) + stack.enter_context(mock.patch.object(ota.shutil, "which", return_value="cargo")) + stack.enter_context(mock.patch.object(ota.sys.stdin, "isatty", return_value=True)) + self.prompt = stack.enter_context(mock.patch("builtins.input", return_value="yes")) + self.check = stack.enter_context(mock.patch.object(ota, "check_bootloader_tool")) + self.install = stack.enter_context(mock.patch.object(ota.subprocess, "run", side_effect=self.fake_install)) + self.output = stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + + def fake_install(self, command, **kwargs): + self.binary.parent.mkdir(parents=True, exist_ok=True) + self.binary.write_bytes(b"test-only installer output, not executable") + return subprocess.CompletedProcess(command, 0) + + def repair(self): + ota.offer_motatool_repair(self.args, self.probe, self.failure) + + def test_consent_installs_pinned_private_tool_rechecks_and_selects_it(self): + with mock.patch.dict(os.environ, {"MESHCORE_ADMIN_PASSWORD": "never-pass-to-build"}): + self.repair() + self.prompt.assert_called_once() + self.install.assert_called_once() + command = self.install.call_args.args[0] + self.assertEqual(command, ota.motatool_repair_command(self.root)) + self.assertIn(ota.MOTATOOL_REPAIR_REPOSITORY, command) + self.assertIn(ota.MOTATOOL_REPAIR_REVISION, command) + self.assertIn("--locked", command) + self.assertNotIn("MESHCORE_ADMIN_PASSWORD", self.install.call_args.kwargs["env"]) + self.assertEqual(self.install.call_args.kwargs["stdin"], subprocess.DEVNULL) + self.assertEqual(self.install.call_args.kwargs["timeout"], ota.MOTATOOL_REPAIR_TIMEOUT_SECONDS) + self.check.assert_called_once_with(str(self.binary), self.probe) + self.assertEqual(self.args.motatool, str(self.binary)) + self.assertIn("Existing motatool installations and PATH will not be changed", self.output.getvalue()) + + def test_decline_or_blank_never_creates_install_directory(self): + for answer in ("", "no", "n", "anything else"): + with self.subTest(answer=answer): + self.prompt.return_value = answer + with self.assertRaisesRegex(ota.OtaError, "declined"): + self.repair() + self.assertFalse(self.root.exists()) + self.install.assert_not_called() + self.assertEqual(self.args.motatool, "old-motatool") + + def test_eof_does_not_approve_install(self): + self.prompt.side_effect = EOFError + with self.assertRaisesRegex(ota.OtaError, "declined"): + self.repair() + self.assertFalse(self.root.exists()) + self.install.assert_not_called() + + def test_yes_flag_never_approves_repair_in_noninteractive_run(self): + self.assertTrue(self.args.yes) + with mock.patch.object(ota.sys.stdin, "isatty", return_value=False): + with self.assertRaisesRegex(ota.OtaError, "--yes does not approve"): + self.repair() + self.prompt.assert_not_called() + self.install.assert_not_called() + self.assertIn("Manual install command:", self.output.getvalue()) + self.assertIn("--motatool", self.output.getvalue()) + self.assertFalse(self.root.exists()) + + def test_missing_cargo_gives_prerequisite_without_installing_anything(self): + with mock.patch.object(ota.shutil, "which", return_value=None): + with self.assertRaisesRegex(ota.OtaError, "https://rustup.rs"): + self.repair() + self.install.assert_not_called() + self.prompt.assert_not_called() + self.assertFalse(self.root.exists()) + + def test_cargo_failure_stops_and_preserves_original_selection(self): + self.install.side_effect = None + self.install.return_value = subprocess.CompletedProcess([], 101) + with self.assertRaisesRegex(ota.OtaError, "Cargo exit 101"): + self.repair() + self.check.assert_not_called() + self.assertEqual(self.args.motatool, "old-motatool") + + def test_cargo_timeout_or_launch_failure_stops(self): + for error in (OSError("cannot start compiler"), subprocess.TimeoutExpired("cargo", 3600)): + with self.subTest(error=error): + self.install.side_effect = error + with self.assertRaisesRegex(ota.OtaError, "repair failed"): + self.repair() + self.check.assert_not_called() + self.assertEqual(self.args.motatool, "old-motatool") + + def test_installer_success_without_binary_is_not_success(self): + self.install.side_effect = None + self.install.return_value = subprocess.CompletedProcess([], 0) + with self.assertRaisesRegex(ota.OtaError, "did not produce"): + self.repair() + self.check.assert_not_called() + self.assertEqual(self.args.motatool, "old-motatool") + + def test_failed_postinstall_check_does_not_select_bad_tool(self): + self.check.side_effect = ota.OtaError("still unsupported") + with self.assertRaisesRegex(ota.OtaError, "still failed bootloader verification"): + self.repair() + self.assertEqual(self.args.motatool, "old-motatool") + + def test_default_tool_reuses_rechecked_cache_without_download_or_prompt(self): + self.args.motatool = "motatool" + self.fake_install([]) + with mock.patch.object(ota.sys.stdin, "isatty", return_value=False): + self.repair() + self.prompt.assert_not_called() + self.install.assert_not_called() + self.check.assert_called_once_with(str(self.binary), self.probe) + self.assertEqual(self.args.motatool, str(self.binary)) + + def test_custom_tool_is_not_silently_replaced_even_with_valid_cache(self): + self.fake_install([]) + self.prompt.return_value = "no" + with self.assertRaisesRegex(ota.OtaError, "declined"): + self.repair() + self.assertIn("cached motatool", self.prompt.call_args.args[0]) + self.assertEqual(self.args.motatool, "old-motatool") + self.install.assert_not_called() + + def test_approved_cached_tool_needs_no_cargo(self): + self.fake_install([]) + with mock.patch.object(ota.shutil, "which", return_value=None): + self.repair() + self.install.assert_not_called() + self.assertEqual(self.check.call_count, 2) + self.assertEqual(self.args.motatool, str(self.binary)) + + def test_bad_cache_requires_new_consent_and_recheck(self): + self.args.motatool = "motatool" + self.fake_install([]) + self.check.side_effect = [ota.OtaError("old cached tool"), None] + self.repair() + self.prompt.assert_called_once() + self.install.assert_called_once() + self.assertEqual(self.check.call_count, 2) + + def test_working_tool_does_not_offer_or_install(self): + with mock.patch.object(ota, "offer_motatool_repair") as offer: + ota.require_bootloader_tool_support(self.args) + offer.assert_not_called() + self.install.assert_not_called() + + def test_missing_or_incompatible_tool_reaches_the_offer(self): + self.check.side_effect = self.failure + with mock.patch.object(ota, "offer_motatool_repair") as offer: + ota.require_bootloader_tool_support(self.args) + offer.assert_called_once_with(self.args, self.probe, self.failure) + + def test_repair_failure_in_main_never_contacts_radios(self): + self.check.side_effect = self.failure + self.prompt.return_value = "no" + with ( + mock.patch.object(ota, "Controller") as controller, + mock.patch.object(ota, "source_cli_command") as source, + contextlib.redirect_stderr(io.StringIO()), + ): + status = ota.main([ + str(self.probe), "remote", "--controller-serial", "controller", + "--source-serial", "source", "--no-install", "--yes", + ]) + self.assertEqual(status, 2) + self.prompt.assert_called_once() + self.install.assert_not_called() + controller.assert_not_called() + source.assert_not_called() + + def test_repaired_binary_is_selected_before_meshcli_preflight(self): + self.check.side_effect = [self.failure, None] + args = ota.build_parser().parse_args([str(self.probe), "remote", "--no-install"]) + with mock.patch.object(ota, "require_meshcli_version") as meshcli: + ota.preflight_inputs(args) + self.assertEqual(args.motatool, str(self.binary)) + meshcli.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/mota/README.md b/tools/mota/README.md index d04c8adf..90b0e031 100644 --- a/tools/mota/README.md +++ b/tools/mota/README.md @@ -82,4 +82,6 @@ from the generated application target table. This library does not authorize a device update. A capable node will only arm such a v3 package through the exact manual confirmation described in [`docs/ota_nrf52_bootloader_update.md`](../../docs/ota_nrf52_bootloader_update.md). -The ordinary `tools/lora_ota` runner rejects bootloader packages. +The `tools/lora_ota` runner accepts verified bootloader packages only with +`--no-install`, checks the destination's bootloader-specific identity and +capabilities, and leaves installation to that explicit manual confirmation.