diff --git a/docs/lora_ota_automation.md b/docs/lora_ota_automation.md index bdef28e20..5f8d0c542 100644 --- a/docs/lora_ota_automation.md +++ b/docs/lora_ota_automation.md @@ -22,10 +22,10 @@ the right container, move the participating nodes to a temporary radio channel, serve and monitor the download, request installation, restore the controller, and check the rebooted node. -The script cannot install the destination's first OTA-capable firmware. It is -also deliberately application-only: it rejects both v3 bootloader containers -and a v2 container carrying the bootloader flag, and it never sends the -privileged install command. Do the first nRF52 bootloader installation over USB. +The script cannot install the destination's first OTA-capable firmware. It can +stage signed v3 bootloader containers with `--no-install`, but never sends the +privileged bootloader install command. Invalid v2 containers carrying the +bootloader flag remain rejected. Do the first nRF52 bootloader installation over USB. An already-capable, explicitly supported internal-flash, XIAO QSPI, or exact MeshTower V2 microSD repeater may later use the separate manual [`ota bootloader install`](ota_nrf52_bootloader_update.md) workflow. @@ -77,6 +77,115 @@ 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. + +If Python cannot load `cryptography`, an interactive run offers: + +```text +Install cryptography privately and retry signature verification? [y/N] +``` + +After approval, it uses the **exact running Python interpreter** to install +`cryptography==50.0.1` and its dependencies from PyPI into a separate per-user, +per-interpreter cache. Binary wheels only are accepted; the repair does not +compile native code, overwrite system/virtual-environment packages, change PATH, +use `sudo`, or bypass externally-managed Python protections. It checks Ed25519 +against a known valid signature and a tampered message, then retries the original +full package verification. A later run can reuse a working cache without another +download. This works for direct `.mota` files and bootloader members inside ZIPs; +ordinary application updates and raw `--tcp-cli` commands do not need this library. + +The interpreter must have working pip. If pip is missing, the script gives an +exact-interpreter `ensurepip` command and operating-system pip/venv guidance; +it does not silently install system tools. No compatible binary wheel, an old +pip, network/download failures, an unwritable cache, or a failed signature check +stop the run before radio access. Pip output is shown and installation is bounded +at ten minutes. The radio admin-password environment variable is not passed to +pip. Partial installation files may remain in the displayed private directory. +The cache is under `meshcore-lora-ota/python//cryptography==50.0.1` +in the same per-user cache base described below for `motatool`. + +Only `y` or `yes` approves this repair; `--yes` is **not** installation consent. +Noninteractive runs print a copyable private-install command and stop. After +running that command manually, rerun the original OTA command; no Python path +changes are needed. Declining never skips signature verification. + +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 motatool 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 | @@ -104,9 +213,19 @@ that was actually flashed. A matching filename or version alone is not enough. ## 1. Install the host tools Install Python 3.10 or newer, [Rust](https://rustup.rs/), Git, the official -[`meshcore-cli`](https://github.com/meshcore-dev/meshcore-cli), and the official +[`meshcore-cli`](https://github.com/meshcore-dev/meshcore-cli) **1.6.0 or newer**, and the official [`motatool`](https://github.com/vk496/motatool). +For signed bootloader staging, see the automatic `cryptography` and +bootloader-capable `motatool` repairs above. Pip is needed only when installing +the missing Python dependency; Rust/Cargo and a native linker are needed only +when building motatool, not when using an already compatible binary. The +runner checks the selected meshcli version before opening radios. Keep its +serial/BLE dependencies in meshcli's own environment (pipx handles these); +installing cryptography in meshcli's pipx environment does not make it available +to the separate Python interpreter running this wrapper. A ready bootloader +`.mota` does not need the Python `detools` or `intelhex` build dependencies. + On Bash: ```bash @@ -559,7 +678,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 b54c24109..9a66c9f12 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 6d97a06ae..7b4f31969 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 @@ -32,11 +33,14 @@ import stat import struct import subprocess import sys +import sysconfig 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 +142,13 @@ 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 +CRYPTOGRAPHY_REPAIR_REQUIREMENT = "cryptography==50.0.1" +CRYPTOGRAPHY_REPAIR_TIMEOUT_SECONDS = 10 * 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 @@ -160,6 +171,10 @@ class OtaError(RuntimeError): """Expected, actionable operator error.""" +class BootloaderCryptoError(OtaError): + """The CLI may offer dependency repair; library parsing never installs code.""" + + class TransmissionError(OtaError): """A command may not have reached its destination or returned a reply.""" @@ -193,6 +208,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 +216,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 +255,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 +835,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 BootloaderCryptoError( + "bootloader signature verification requires Python cryptography; " + f"the dependency could not load in {sys.executable}: {exc}" + ) 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 +892,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 +1141,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 +1153,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 +1164,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 +1240,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 +1252,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 +1435,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 +1454,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 +1472,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 +1541,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 +1582,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 +2160,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 +2224,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 +2375,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 +2503,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 +2714,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 +2731,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 +4157,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 +4172,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 +4201,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 +6411,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 +6864,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 +6888,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 +6938,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 +6994,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,6 +7248,407 @@ 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 host_repair_cache() -> 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("host tool repair needs an absolute user cache directory") + return cache / "meshcore-lora-ota" + + +def motatool_repair_root() -> Path: + return host_repair_cache() / "motatool" / MOTATOOL_REPAIR_REVISION + + +def cryptography_repair_root() -> Path: + # Do not mix wheels between virtual environments, architectures or ABIs. + identity = "|".join(( + sys.executable, sys.version, sysconfig.get_platform(), + str(sysconfig.get_config_var("SOABI")), + )) + tag = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:16] + return host_repair_cache() / "python" / tag / CRYPTOGRAPHY_REPAIR_REQUIREMENT + + +def cryptography_repair_command(root: Path) -> list[str]: + return [ + sys.executable, "-m", "pip", "--isolated", "--disable-pip-version-check", + "install", "--no-input", "--only-binary=:all:", "--no-cache-dir", + "--index-url", "https://pypi.org/simple", "--upgrade", + "--target", str(root), CRYPTOGRAPHY_REPAIR_REQUIREMENT, + ] + + +def activate_cached_cryptography(root: Path) -> None: + """Load a private wheel installation without unloading live native modules.""" + old_path = sys.path[:] + sys.path.insert(0, str(root)) + importlib.invalidate_caches() + try: + import cryptography + from cryptography.exceptions import InvalidSignature + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + + if not Path(cryptography.__file__).resolve().is_relative_to(root.resolve()): + raise ImportError( + "private installation did not provide cryptography; a conflicting or " + "partially loaded system installation may need repair in your Python environment" + ) + # RFC 8032 test 1: a working import is not enough to prove Ed25519 works. + key = Ed25519PublicKey.from_public_bytes(bytes.fromhex( + "d75a980182b10ab7d54bfed3c964073a0ee172f3daa62325af021a68f707511a" + )) + signature = bytes.fromhex( + "e5564300c360ac729086e2cc806e828a84877f1eb8e5d974d873e065224901555f" + "b8821590a33bacc61e39701cf9b46bd25bf5f0595bbe24655141438e7a100b" + ) + key.verify(signature, b"") + try: + key.verify(signature, b"tampered") + except InvalidSignature: + pass + else: + raise ValueError("Ed25519 accepted a modified message") + except Exception as exc: + # Never unload/reload Rust/CFFI extensions: their retained exception + # classes may then differ from Python's, breaking signature rejection. + sys.path[:] = old_path + importlib.invalidate_caches() + raise OtaError(f"cached cryptography failed its Ed25519 check: {exc}") from exc + + +def offer_cryptography_repair(failure: BootloaderCryptoError) -> None: + root = cryptography_repair_root() + print(f"[host] {failure}") + if root.is_dir(): + try: + activate_cached_cryptography(root) + except OtaError as exc: + print(f"[host] {exc}") + else: + print(f"[host] using previously installed, rechecked cryptography: {root}") + return + + command = cryptography_repair_command(root) + print(f"[repair] Python: {sys.executable}") + print(f"[repair] Download {CRYPTOGRAPHY_REPAIR_REQUIREMENT} and its wheel dependencies from PyPI.") + print(f"[repair] Private install: {root}") + print("[repair] System/virtual-environment packages and PATH will not be changed.") + print(f"[repair] Manual install command: {display_host_command(command)}") + print("[repair] After a manual install, rerun the same OTA command; it rechecks this cache.") + # Probe this interpreter, not an unrelated `pip` executable on PATH. + install_env = os.environ.copy() + install_env.pop("MESHCORE_ADMIN_PASSWORD", None) + install_env["PIP_CONFIG_FILE"] = os.devnull + try: + pip = subprocess.run( + [sys.executable, "-m", "pip", "--isolated", "--version"], + stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + timeout=30, check=False, env=install_env, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise OtaError(f"cannot check pip in {sys.executable}: {exc}; no installation attempted") from exc + if pip.returncode != 0: + bootstrap = display_host_command([sys.executable, "-m", "ensurepip", "--upgrade"]) + raise OtaError( + f"pip is unavailable in this Python environment. Enable it with {bootstrap}, " + "or install your operating system's Python pip/venv package and rerun. " + "No installation or radio changes were made" + ) + if not sys.stdin.isatty(): + raise OtaError( + "cryptography repair needs separate interactive approval (--yes does not " + "approve software installation); rerun in a terminal or use the printed " + "manual command. No installation was attempted" + ) + try: + approved = input("Install cryptography privately and retry signature verification? [y/N] ").strip().lower() in ("y", "yes") + except EOFError: + approved = False + if not approved: + raise OtaError("cryptography repair declined; no installation or radio changes were made") + try: + root.mkdir(parents=True, exist_ok=True) + print("[repair] Installing binary wheels (pip output follows)...", flush=True) + result = subprocess.run( + command, stdin=subprocess.DEVNULL, timeout=CRYPTOGRAPHY_REPAIR_TIMEOUT_SECONDS, + check=False, env=install_env, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise OtaError(f"cryptography repair failed: {exc}; stopping before radio access") from exc + if result.returncode != 0: + raise OtaError( + f"cryptography repair failed (pip exit {result.returncode}); check network, " + "disk space and wheel support for this Python/platform. If pip is too old, " + "update it in your Python environment. No source builds, system-package " + f"overrides or radio changes were attempted; partial files may remain in {root}" + ) + activate_cached_cryptography(root) + print("[repair] Ed25519 self-test passed; retrying full package/signature verification.") + + +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 BootloaderCryptoError: + raise # A missing dependency is not an invalid ZIP member. + 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_package_tools(args: argparse.Namespace) -> None: + 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") + if args.package_kind == "bootloader": + require_bootloader_tool_support(args) + else: + require_command(args.motatool, "motatool") + + def preflight_inputs(args: argparse.Namespace) -> None: if not args.package.is_file(): raise OtaError(f"package does not exist: {args.package.resolve()}") @@ -7027,7 +7661,14 @@ 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") + try: + preflight_package_tools(args) + except BootloaderCryptoError as exc: + # Refuse unsupported actions before offering software installation. Retry + # once only, and never skip the original package/signature validation. + require_package_action(args, True) + offer_cryptography_repair(exc) + preflight_package_tools(args) if not args.prepare_only: require_meshcli_version(args.meshcli) @@ -7252,6 +7893,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 +8161,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 42f20d28e..e6190d4cb 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 000000000..4aed554ab --- /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_crypto_install.py b/tools/lora_ota/test_lora_ota_crypto_install.py new file mode 100644 index 000000000..ef7c0ea4b --- /dev/null +++ b/tools/lora_ota/test_lora_ota_crypto_install.py @@ -0,0 +1,88 @@ +"""Opt-in real wheel installation test, isolated from system packages and radios. + +Set MOTA_CRYPTO_INSTALL_TEST=1 to allow PyPI downloads into a temporary test cache. +""" + +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + +from test_lora_ota_bootloader import boot_blob + + +CHILD = r''' +import importlib.util +from pathlib import Path +import sys +from unittest import mock + +sys.path.insert(0, sys.argv[1]) +import lora_ota as ota +root, package, mode = Path(sys.argv[2]), Path(sys.argv[3]), sys.argv[4] +assert importlib.util.find_spec("cryptography") is None, "venv must start without crypto" +args = ota.build_parser().parse_args([str(package), "remote", "--no-install"]) +with ( + mock.patch.object(ota, "cryptography_repair_root", return_value=root), + mock.patch.object(ota.sys.stdin, "isatty", return_value=True), + mock.patch("builtins.input", return_value="yes") as approval, + mock.patch.object(ota, "check_bootloader_tool"), + mock.patch.object(ota, "require_meshcli_version"), + mock.patch.object(ota, "Controller") as controller, + mock.patch.object(ota, "source_cli_command") as source, +): + if mode == "install": + ota.preflight_inputs(args) + approval.assert_called_once() + else: + with mock.patch.object(ota.subprocess, "run", side_effect=AssertionError("cache must not reinstall")): + ota.preflight_inputs(args) + approval.assert_not_called() + controller.assert_not_called() + source.assert_not_called() +import cryptography +assert Path(cryptography.__file__).is_relative_to(root) +assert cryptography.__version__ == ota.CRYPTOGRAPHY_REPAIR_REQUIREMENT.split("==")[1] +assert args.package_kind == "bootloader" +bad = bytearray(package.read_bytes()) +bad[137] ^= 1 +try: + ota.parse_mota(bytes(bad)) +except ota.OtaError as exc: + assert "invalid bootloader mOTA" in str(exc), str(exc) +else: + raise AssertionError("tampered signature was accepted") +print("REAL CRYPTO REPAIR OK:", mode) +''' + + +@unittest.skipUnless(os.environ.get("MOTA_CRYPTO_INSTALL_TEST") == "1", "opt-in PyPI wheel download") +class CryptoInstallIntegrationTests(unittest.TestCase): + def test_clean_venv_installs_then_new_process_reuses_cache(self): + with tempfile.TemporaryDirectory(prefix="meshcore-crypto-install-test-") as directory: + folder = Path(directory) + environment = folder / "venv" + child_env = os.environ.copy() + child_env.pop("MESHCORE_ADMIN_PASSWORD", None) + subprocess.run( + [sys.executable, "-m", "venv", str(environment)], + env=child_env, check=True, timeout=120, + ) + python = environment / ("Scripts/python.exe" if os.name == "nt" else "bin/python") + package = folder / "boot.mota" + package.write_bytes(boot_blob()) + for mode in ("install", "reuse"): + result = subprocess.run( + [str(python), "-I", "-c", CHILD, str(Path(__file__).parent.resolve()), + str(folder / "private-crypto"), str(package), mode], + env=child_env, check=False, timeout=660, text=True, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + ) + self.assertEqual(result.returncode, 0, result.stdout) + self.assertIn("REAL CRYPTO REPAIR OK: " + mode, result.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/lora_ota/test_lora_ota_crypto_repair.py b/tools/lora_ota/test_lora_ota_crypto_repair.py new file mode 100644 index 000000000..edaf1d254 --- /dev/null +++ b/tools/lora_ota/test_lora_ota_crypto_repair.py @@ -0,0 +1,286 @@ +"""Host dependency repair regressions. No network, installers or radio access.""" + +import contextlib +import io +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest import mock +import zipfile + +import lora_ota as ota +from test_lora_ota import firmware, mota_blob, VERSION_NEW +from test_lora_ota_bootloader import boot_blob + + +class CryptographyRepairTests(unittest.TestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.root = Path(directory.name) / "private-python" + self.failure = ota.BootloaderCryptoError("No module named cryptography") + stack = contextlib.ExitStack() + self.addCleanup(stack.close) + stack.enter_context(mock.patch.object(ota, "cryptography_repair_root", return_value=self.root)) + 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.activate = stack.enter_context(mock.patch.object(ota, "activate_cached_cryptography")) + self.run = stack.enter_context(mock.patch.object( + ota.subprocess, "run", return_value=subprocess.CompletedProcess([], 0), + )) + self.output = stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + + def repair(self): + ota.offer_cryptography_repair(self.failure) + + def test_approved_install_uses_same_interpreter_private_wheels_and_rechecks(self): + with mock.patch.dict(os.environ, {"MESHCORE_ADMIN_PASSWORD": "do-not-inherit"}): + self.repair() + self.prompt.assert_called_once() + self.assertTrue(self.root.is_dir()) + self.assertEqual(self.run.call_count, 2) + probe, install = self.run.call_args_list + self.assertEqual(probe.args[0][:3], [sys.executable, "-m", "pip"]) + self.assertIn("--version", probe.args[0]) + self.assertEqual(install.args[0], ota.cryptography_repair_command(self.root)) + self.assertIn("--only-binary=:all:", install.args[0]) + self.assertIn("--isolated", install.args[0]) + self.assertNotIn("--break-system-packages", install.args[0]) + for call in (probe, install): + self.assertNotIn("MESHCORE_ADMIN_PASSWORD", call.kwargs["env"]) + self.assertEqual(call.kwargs["env"]["PIP_CONFIG_FILE"], os.devnull) + self.assertEqual(call.kwargs["stdin"], subprocess.DEVNULL) + self.assertEqual(install.kwargs["timeout"], ota.CRYPTOGRAPHY_REPAIR_TIMEOUT_SECONDS) + self.activate.assert_called_once_with(self.root) + + def test_blank_decline_and_eof_never_install(self): + for answer in ("", "no", "unexpected", EOFError()): + with self.subTest(answer=answer): + self.prompt.side_effect = answer if isinstance(answer, Exception) else None + self.prompt.return_value = answer + self.run.reset_mock() + with self.assertRaisesRegex(ota.OtaError, "declined"): + self.repair() + self.assertFalse(self.root.exists()) + self.assertEqual(self.run.call_count, 1) # Read-only pip probe. + self.activate.assert_not_called() + + def test_noninteractive_never_installs_and_prints_manual_command(self): + 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.assertEqual(self.run.call_count, 1) + self.assertFalse(self.root.exists()) + self.assertIn("Manual install command:", self.output.getvalue()) + self.assertIn(str(self.root), self.output.getvalue()) + + def test_missing_pip_reports_exact_interpreter_bootstrap(self): + self.run.return_value = subprocess.CompletedProcess([], 1) + with self.assertRaisesRegex(ota.OtaError, "ensurepip") as error: + self.repair() + self.assertIn(sys.executable, str(error.exception)) + self.prompt.assert_not_called() + self.assertFalse(self.root.exists()) + self.activate.assert_not_called() + + def test_pip_probe_failure_is_bounded_and_does_not_install(self): + for error in (OSError("not executable"), subprocess.TimeoutExpired("pip", 30)): + with self.subTest(error=error): + self.run.side_effect = error + with self.assertRaisesRegex(ota.OtaError, "cannot check pip"): + self.repair() + self.prompt.assert_not_called() + self.assertFalse(self.root.exists()) + + def test_failed_install_never_activates_partial_files(self): + self.run.side_effect = [subprocess.CompletedProcess([], 0), subprocess.CompletedProcess([], 1)] + with self.assertRaisesRegex(ota.OtaError, "pip exit 1"): + self.repair() + self.activate.assert_not_called() + + def test_install_timeout_and_launch_failure_stop(self): + for error in (OSError("cannot start"), subprocess.TimeoutExpired("pip", 600)): + with self.subTest(error=error): + self.run.side_effect = [subprocess.CompletedProcess([], 0), error] + with mock.patch.object(Path, "is_dir", return_value=False), self.assertRaisesRegex(ota.OtaError, "repair failed"): + self.repair() + self.activate.assert_not_called() + + def test_unwritable_cache_is_an_actionable_error(self): + with mock.patch.object(Path, "mkdir", side_effect=PermissionError("read-only cache")): + with self.assertRaisesRegex(ota.OtaError, "read-only cache"): + self.repair() + self.assertEqual(self.run.call_count, 1) + self.activate.assert_not_called() + + def test_successful_installer_is_not_enough_if_crypto_check_fails(self): + self.activate.side_effect = ota.OtaError("Ed25519 check failed") + with self.assertRaisesRegex(ota.OtaError, "Ed25519 check failed"): + self.repair() + self.assertNotIn("self-test passed", self.output.getvalue()) + + def test_cached_install_is_rechecked_without_pip_or_prompt(self): + self.root.mkdir() + with mock.patch.object(ota.sys.stdin, "isatty", return_value=False): + self.repair() + self.activate.assert_called_once_with(self.root) + self.run.assert_not_called() + self.prompt.assert_not_called() + + def test_bad_cache_offers_repair_and_rechecks_again(self): + self.root.mkdir() + self.activate.side_effect = [ota.OtaError("broken wheel"), None] + self.repair() + self.prompt.assert_called_once() + self.assertEqual(self.run.call_count, 2) + self.assertEqual(self.activate.call_count, 2) + + +class CryptoPreflightTests(unittest.TestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.folder = Path(directory.name) + self.path = self.folder / "boot.mota" + self.blob = boot_blob() + self.path.write_bytes(self.blob) + self.library = ota.bootloader_library() + self.real_verify = self.library.verify + self.missing = True + stack = contextlib.ExitStack() + self.addCleanup(stack.close) + self.verify = stack.enter_context(mock.patch.object(self.library, "verify", side_effect=self.verify_package)) + self.repair = stack.enter_context(mock.patch.object(ota, "offer_cryptography_repair", side_effect=self.fix_crypto)) + self.tool = stack.enter_context(mock.patch.object(ota, "check_bootloader_tool")) + self.meshcli = stack.enter_context(mock.patch.object(ota, "require_meshcli_version")) + self.controller = stack.enter_context(mock.patch.object(ota, "Controller")) + self.source = stack.enter_context(mock.patch.object(ota, "source_cli_command")) + stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + self.stderr = stack.enter_context(contextlib.redirect_stderr(io.StringIO())) + + def verify_package(self, parsed): + if self.missing: + raise ImportError("No module named cryptography") + return self.real_verify(parsed) + + def fix_crypto(self, failure): + self.assertIsInstance(failure, ota.BootloaderCryptoError) + self.missing = False + + def args(self, *extra): + return ota.build_parser().parse_args([str(self.path), "remote", "--no-install", *extra]) + + def test_direct_bootloader_repaired_then_fully_verified(self): + args = self.args() + ota.preflight_inputs(args) + self.repair.assert_called_once() + self.assertEqual(self.verify.call_count, 2) + self.assertEqual(args.package_kind, "bootloader") + self.tool.assert_called_once() + self.meshcli.assert_called_once() + self.controller.assert_not_called() + self.source.assert_not_called() + + def test_zip_dependency_failure_is_not_misreported_as_invalid_member(self): + self.path = self.folder / "boot.zip" + with zipfile.ZipFile(self.path, "w") as archive: + archive.writestr("board/boot.mota", self.blob) + ota.preflight_inputs(self.args("--zip-member", "board/boot.mota")) + self.repair.assert_called_once() + self.assertEqual(self.verify.call_count, 2) + self.tool.assert_called_once() + + def test_working_crypto_never_offers_install(self): + self.missing = False + ota.preflight_inputs(self.args()) + self.repair.assert_not_called() + self.tool.assert_called_once() + + def test_application_updates_do_not_require_crypto(self): + self.path.write_bytes(mota_blob(firmware(b"app" * 1000, VERSION_NEW))) + with mock.patch.object(ota, "require_command"): + ota.preflight_inputs(self.args()) + self.verify.assert_not_called() + self.repair.assert_not_called() + + def test_raw_tcp_command_does_not_require_crypto_or_package_tools(self): + with mock.patch.object(ota, "tcp_cli_main", return_value=0) as raw: + self.assertEqual(ota.main(["--tcp-cli", "127.0.0.1", "ota status"]), 0) + raw.assert_called_once() + self.verify.assert_not_called() + self.repair.assert_not_called() + + def test_library_parsing_never_prompts_or_installs(self): + with self.assertRaises(ota.BootloaderCryptoError): + ota.parse_mota(self.blob) + self.repair.assert_not_called() + + def test_unsafe_boot_action_rejected_before_offering_install(self): + args = self.args("--yes") + args.no_install = False + with self.assertRaisesRegex(ota.OtaError, "use --no-install"): + ota.preflight_inputs(args) + self.repair.assert_not_called() + self.tool.assert_not_called() + + def test_repair_is_attempted_only_once(self): + self.repair.side_effect = None # Simulate installer returning without fixing anything. + with self.assertRaises(ota.BootloaderCryptoError): + ota.preflight_inputs(self.args()) + self.repair.assert_called_once() + self.tool.assert_not_called() + + def test_invalid_signature_still_fails_after_dependency_repair(self): + bad = bytearray(self.blob) + bad[137] ^= 1 + self.path.write_bytes(bad) + with self.assertRaisesRegex(ota.OtaError, "invalid bootloader mOTA"): + ota.preflight_inputs(self.args()) + self.repair.assert_called_once() + self.tool.assert_not_called() + self.meshcli.assert_not_called() + + def test_declined_repair_in_main_stops_before_any_radio_access(self): + self.repair.side_effect = ota.OtaError("cryptography repair declined") + status = ota.main([ + str(self.path), "remote", "--no-install", "--yes", + "--controller-serial", "controller", "--source-serial", "source", + ]) + self.assertEqual(status, 2) + self.assertIn("cryptography repair declined", self.stderr.getvalue()) + self.controller.assert_not_called() + self.source.assert_not_called() + self.tool.assert_not_called() + + +class CryptoCacheTests(unittest.TestCase): + def test_interpreter_identity_changes_cache(self): + original = ota.cryptography_repair_root() + with mock.patch.object(ota.sys, "executable", "another-python"): + self.assertNotEqual(original, ota.cryptography_repair_root()) + with mock.patch.object(ota.sysconfig, "get_platform", return_value="another-architecture"): + self.assertNotEqual(original, ota.cryptography_repair_root()) + + def test_empty_cache_cannot_use_system_crypto_and_rolls_back_imports(self): + import cryptography + old_path = sys.path[:] + with tempfile.TemporaryDirectory() as directory: + with self.assertRaisesRegex(ota.OtaError, "private installation did not provide"): + ota.activate_cached_cryptography(Path(directory)) + self.assertEqual(sys.path, old_path) + self.assertIs(sys.modules["cryptography"], cryptography) + + def test_real_ed25519_known_vector_and_tampered_message(self): + import cryptography + root = Path(cryptography.__file__).parent.parent + with mock.patch.object(sys, "path", sys.path[:]): + ota.activate_cached_cryptography(root) + self.assertEqual(sys.path[0], str(root)) + + +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 000000000..edc13b0d4 --- /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 d04c8adfd..90b0e031f 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.