Support bootloader mOTA staging and guided host tool repair

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/<revision>` 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 <MID8> <HASH16>
+```
+
+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

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

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("<IIII", blob, 11)
     block_size_log2 = blob[27]
@@ -1001,6 +1062,27 @@ def read_zip_member(archive: zipfile.ZipFile, member: zipfile.ZipInfo) -> 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 <MID8> <HASH16>`; 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 <MID8> <HASH16>` 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:

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,

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("<II", image, 0, 0x20040000, ml.XIAO_BOOT_IMAGE_START + 0x101)
+    struct.pack_into("<8sHHB3x", image, 0x80, ml.XIAO_BOOT_CAPS_MAGIC, 3, 5, storage)
+    offset = ml.BOOT_CANDIDATE_MANIFEST_OFFSET
+    struct.pack_into(
+        "<8sHHIII16sI", image, offset, ml.XIAO_BOOT_MANIFEST_MAGIC,
+        ml.XIAO_BOOT_MANIFEST_VERSION, ml.XIAO_BOOT_MANIFEST_SIZE,
+        ml.XIAO_BOOT_IMAGE_START, len(image), board, name.encode("ascii"), 0,
+    )
+    struct.pack_into(
+        "<8sHHIHHIHHI", image, offset + ml.XIAO_BOOT_MANIFEST_SIZE,
+        ml.BOOT_CONTINUITY_MAGIC, ml.BOOT_CONTINUITY_VERSION, ml.BOOT_CONTINUITY_SIZE,
+        version, ml.BOOT_CONTINUITY_FAMILY_S140,
+        0x0123 if app_base == ml.NRF52_APP_BASE_S140_V7 else 0x00B6,
+        app_base, ml.BOOT_CONTINUITY_LAYOUT_ABI, 0, 0,
+    )
+    struct.pack_into("<I", image, offset + 40, zlib.crc32(image) & 0xFFFFFFFF)
+    image = bytes(image)
+    manifest = ml.build_manifest(
+        target_id=ml.bootloader_target_id(board, name), fw_version=version,
+        image_size=len(image), payload=image, block_size=1024,
+        image_hash=ml.mh32(image), codec_id=ml.CODEC_FULL, is_full=True,
+        sign_priv=Ed25519PrivateKey.from_private_bytes(b"\x01" * 32), bootloader=True,
+    )
+    return ml.build_container(manifest, image)
+
+
+def boot_target(package):
+    return replace(
+        target(platform="nrf52", boot_codecs=5), bootloader_abi=3,
+        boot_target_id=package.target_id, boot_hw_id=package.hw_id,
+        boot_storage=package.bootloader_storage,
+        nrf_sd=package.bootloader_storage == 9,
+        nrf_qspi=package.bootloader_storage == 14,
+    )
+
+
+class BootloaderStagingTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.blob = boot_blob()
+        cls.package = ota.parse_mota(cls.blob)
+
+    def test_signed_full_bootloader_is_identified(self):
+        package = self.package
+        self.assertTrue(package.is_bootloader)
+        self.assertEqual(package.kind, "bootloader full")
+        self.assertEqual(package.target_id, 0x23818A80)
+        self.assertEqual(package.block_size, 1024)
+        self.assertEqual(package.payload_size, 40960)
+
+    def test_all_supported_storage_profiles(self):
+        for storage in (10, 14, 9):
+            with self.subTest(storage=storage):
+                package = ota.parse_mota(boot_blob(storage=storage))
+                self.assertEqual(ota.compatible_mota(package, boot_target(package)), (True, ""))
+
+    def test_malformed_bootloaders_fail_closed(self):
+        for offset in (9, 10, 11, 27, 28, 105, 137, 201, 205, 600):
+            with self.subTest(offset=offset):
+                bad = bytearray(self.blob)
+                bad[offset] ^= 1
+                with self.assertRaises(ota.OtaError):
+                    ota.parse_mota(bytes(bad))
+
+    def test_bootloader_is_not_an_application_full_exception(self):
+        application = ota.parse_mota(mota_blob(firmware(b"test" * 1000, VERSION_NEW)))
+        self.assertFalse(ota.compatible_mota(application, target(platform="nrf52"))[0])
+
+    def test_bootloader_compatibility_requires_exact_live_contract(self):
+        live = boot_target(self.package)
+        for change in (
+            dict(platform="esp32"), dict(boot_target_id=None),
+            dict(boot_target_id=self.package.target_id ^ 1), dict(boot_hw_id="wrong"),
+            dict(bootloader_abi=2), dict(bootloader_codecs=4),
+            dict(boot_storage=14), dict(nrf_qspi=True), dict(max_block_size=512),
+        ):
+            with self.subTest(change=change):
+                self.assertFalse(ota.compatible_mota(self.package, replace(live, **change))[0])
+
+    def test_stage_only_is_required_even_with_yes(self):
+        args = argparse.Namespace(no_install=False, yes=True)
+        with self.assertRaisesRegex(ota.OtaError, "use --no-install"):
+            ota.require_package_action(args, True)
+        args.no_install = True
+        ota.require_package_action(args, True)
+        for key, value in (("prepare_only", True), ("allow_non_upgrade", True), ("base", Path("base.bin"))):
+            with self.subTest(option=key):
+                with self.assertRaises(ota.OtaError):
+                    ota.require_package_action(argparse.Namespace(no_install=True, **{key: value}), True)
+
+    def test_default_run_rejects_boot_before_contacting_radios(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "not-an-app.mota"
+            path.write_bytes(self.blob)
+            with (
+                mock.patch.object(ota, "Controller") as controller,
+                mock.patch.object(ota, "source_cli_command") as source,
+                mock.patch.object(ota, "require_command") as require,
+                contextlib.redirect_stdout(io.StringIO()),
+                contextlib.redirect_stderr(io.StringIO()) as error,
+            ):
+                result = ota.main([
+                    str(path), "remote", "--controller-serial", "controller",
+                    "--source-serial", "source", "--yes",
+                ])
+            self.assertEqual(result, 2)
+            self.assertIn("use --no-install", error.getvalue())
+            controller.assert_not_called()
+            source.assert_not_called()
+            require.assert_not_called()
+
+    def test_prepare_boot_stage_uses_motatool_verification(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "boot.mota"
+            source.write_bytes(self.blob)
+            args = prepare_args(source, "motatool")
+            args.no_install = True
+            with mock.patch.object(ota, "verify_with_motatool") as verify:
+                path, package, body_hash = ota.prepare_package(args, boot_target(self.package), root)
+            self.assertTrue(package.is_bootloader)
+            self.assertIsNone(body_hash)
+            self.assertEqual(path.read_bytes(), self.blob)
+            verify.assert_called_once_with("motatool", path, None)
+
+    def test_mixed_archive_requires_explicit_choice(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "mixed.zip"
+            with zipfile.ZipFile(path, "w") as archive:
+                archive.writestr("renamed.mota", self.blob)
+                archive.writestr("firmware.bin", firmware(b"test" * 1000, VERSION_NEW))
+            with self.assertRaisesRegex(ota.OtaError, "both application and bootloader"):
+                ota.inspect_package_kind(path, None)
+            self.assertEqual(ota.inspect_package_kind(path, "renamed.mota"), "bootloader")
+            self.assertEqual(ota.inspect_package_kind(path, "firmware.bin"), "application")
+            with self.assertRaisesRegex(ota.OtaError, "not found"):
+                ota.inspect_package_kind(path, "missing.mota")
+
+    def test_boot_archive_selects_matching_newest_and_never_falls_back(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "boot.zip"
+            with zipfile.ZipFile(path, "w") as archive:
+                archive.writestr("old.mota", self.blob)
+                archive.writestr("new.mota", boot_blob(version=0x0117010E))
+                archive.writestr("xiao.mota", boot_blob(storage=14))
+            with zipfile.ZipFile(path) as archive:
+                selected = ota.select_mota_from_zip(archive, boot_target(self.package), None, package_kind="bootloader")
+                self.assertEqual(selected[1], "new.mota")
+                with self.assertRaisesRegex(ota.OtaError, "unusable"):
+                    ota.select_mota_from_zip(archive, target(platform="esp32"), None, package_kind="bootloader")
+
+    def test_duplicate_zip_member_cannot_be_selected_by_name(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "duplicate.zip"
+            with warnings.catch_warnings(), zipfile.ZipFile(path, "w") as archive:
+                warnings.simplefilter("ignore", UserWarning)
+                archive.writestr("boot.mota", self.blob)
+                archive.writestr("boot.mota", self.blob)
+            with self.assertRaisesRegex(ota.OtaError, "duplicated"):
+                ota.inspect_package_kind(path, "boot.mota")
+
+    def test_query_uses_boot_identity_not_application_id(self):
+        controller = mock.Mock()
+        controller.remote_command.return_value = (
+            "BL board=239A0029 target=23818A80 name=3401_DFU "
+            "crc=12345678 abi=3 caps=0A | staged:none mid=- hash=-"
+        )
+        live = replace(target(platform="nrf52", boot_codecs=5), bootloader_abi=3)
+        result = ota.query_bootloader_target(controller, live)
+        self.assertEqual(result.target_id, live.target_id)
+        self.assertEqual(result.boot_target_id, self.package.target_id)
+        self.assertEqual(result.boot_hw_id, self.package.hw_id)
+        for reply in (
+            "ERR this build cannot update its bootloader over LoRa",
+            controller.remote_command.return_value.replace("23818A80", "00000000"),
+            controller.remote_command.return_value.replace("abi=3", "abi=2"),
+        ):
+            controller.remote_command.return_value = reply
+            with self.assertRaises(ota.OtaError):
+                ota.query_bootloader_target(controller, live)
+
+    def test_staging_prints_confirmation_but_never_installs(self):
+        package = self.package
+        args = argparse.Namespace(target="remote")
+        controller = mock.Mock()
+        reply = (
+            "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()}"
+        )
+        controller.remote_command.return_value = reply
+        with contextlib.redirect_stdout(io.StringIO()) as output:
+            ota.report_staged_update(controller, args, package)
+        controller.remote_command.assert_called_once_with("remote", "ota bootloader status")
+        self.assertIn("NOT installed", output.getvalue())
+        self.assertIn(f"ota bootloader install {package.manifest_id}", output.getvalue())
+        for bad in (reply.replace("staged:ready", "staged:none"), reply.replace(package.manifest_id, "00000000"), reply.replace(package.image_hash[:8].hex(), "0000000000000000")):
+            controller.remote_command.return_value = bad
+            with self.assertRaisesRegex(ota.OtaError, "exact staged"):
+                ota.report_staged_update(controller, args, package)
+        controller.reset_mock()
+        with self.assertRaisesRegex(ota.OtaError, "explicit"):
+            ota.request_install(controller, args, package)
+        controller.remote_command.assert_not_called()
+
+    def test_target_query_only_adds_boot_probe_for_boot_packages(self):
+        replies = {
+            "ota status": "OTA | no download | target:1234ABCD hw=TestBoard",
+            "get bootloader.ver": "> 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()

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()

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.