Merge remote-tracking branch 'origin/keymindCascade' into keymindCascade

docs/lora_ota_automation.md

@@ -139,8 +139,38 @@ 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
+matching stable Cargo and rustc **1.86.0 or newer** and a native linker already
+installed; otherwise it points to
 [Rust installation](https://rustup.rs/) and stops without installing anything.
+The pinned dependency set needs more than the older `rust-version` in motatool's
+own manifest: `zeroize 1.9.0` needs edition 2024, and ICU/idna dependencies need
+Rust 1.86. Cargo 1.75 is a version problem, not a root-permission problem; no
+nightly toolchain is necessary.
+
+Before asking to build or downloading crates, the runner checks both executable
+versions. It keeps a compatible default pair; otherwise it looks for versioned
+executables on PATH (such as Ubuntu's `cargo-1.91` and `rustc-1.91`) and already
+installed stable rustup toolchains. It displays the selected paths and pins both
+for the build process, so a newer Cargo cannot silently invoke an old rustc.
+Rustup proxies are resolved to their actual toolchain binaries before changing
+to the build directory. No default toolchain, symlink, system PATH, or package
+installation is changed, and probes disable rustup's automatic toolchain install.
+Tool version probes have ten-second timeouts. A working cached motatool does not
+need Cargo or Rust checks at all.
+
+To choose a pair explicitly, append:
+
+```text
+--cargo cargo-1.91 --rustc rustc-1.91
+```
+
+These options also accept executable paths, not shell aliases or a command with
+embedded arguments. Explicit selections and `RUSTC`/`CARGO_BUILD_RUSTC` compiler
+overrides are honored; if incompatible, they fail instead of silently selecting
+another compiler. `--rustc` takes precedence over those environment variables.
+The manual build command includes the chosen compiler as well as Cargo. Native
+linker, SDK, and system-library availability are still checked by the real build;
+a passed version check is not a guarantee that every host prerequisite is present.
 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.
@@ -186,6 +216,32 @@ 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.
 
+For a separate source with a managed USB/TCP console, automatic source selection
+reads the physical source's full public key and matches it in the controller's
+contacts. Repeaters use `get public.key`. Full Companion does not implement
+that repeater command: on both USB and TCP, its key is read from a **fresh
+terminal welcome banner**, with a supported `ver` reply on the same connection
+proving the terminal is live. USB briefly uses the existing STOP/START/STOP
+terminal wrapper to obtain that banner, even for an ASCII-first Full Companion;
+TCP reads the greeting from a new connection. Identity rechecks repeat this
+exchange rather than trusting a key cached from an earlier connection.
+
+`ota status` identifies the Full Companion seeder role before this probe. For
+unrecognized firmware, only an explicit unsupported `get public.key` response
+can trigger the Companion fallback. Timeouts, permission errors, malformed keys,
+missing/ambiguous banners, and failed `ver` replies still stop the run. Firmware
+that exposes neither identity mechanism needs a supported terminal firmware;
+the runner does not guess a key from the node name or skip the identity check.
+
+Source selection does not require the source's current name to match an
+old saved advert, and duplicate/emoji names cannot redirect this lookup. An
+explicit `--source-contact` must identify that same physical source. If its key
+really is missing, import the source contact or advertise/discover it on the
+normal channel first; the normal/TempRadio on-air ACK checks still require the
+contact. The script does not silently create contacts or skip those checks.
+When the controller and source are the same TCP Full Companion, use the verified
+`--source-shares-controller` topology; it does not need a contact for itself.
+
 ## Destination requirements
 
 | Destination | Package installed | One-time prerequisite | Raw ZIP handling |

tools/lora_ota/lora_ota.py

@@ -147,6 +147,10 @@ MIN_MESHCLI_VERSION = (1, 6, 0)
 MOTATOOL_REPAIR_REPOSITORY = "https://github.com/mikecarper/motatool.git"
 MOTATOOL_REPAIR_REVISION = "8c38369e7d35ad50cf74261869676d52dd24adf7"
 MOTATOOL_REPAIR_TIMEOUT_SECONDS = 60 * 60
+# The pinned lockfile, not motatool's older package-level rust-version: zeroize
+# needs edition 2024 (1.85), and idna_adapter/ICU 2.2 require Rust 1.86.
+MOTATOOL_MIN_RUST_VERSION = (1, 86, 0)
+RUST_TOOL_PROBE_TIMEOUT_SECONDS = 10
 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,
@@ -2558,11 +2562,34 @@ def bind_contact_selectors(controller: Controller, args: argparse.Namespace) ->
     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
+        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
+        # USB/TCP identifies the physical source. Its current display name may
+        # differ from the controller's saved advert (renames, truncation, emoji).
+        # Never bind a same-named but different radio, even with --source-contact.
+        source_key = read_source_public_key_bounded(args).lower()
+        if source_selector:
+            source = resolve(source_selector)
+            if source != source_key:
+                raise OtaError(
+                    f"--source-contact identifies {source}, but the connected OTA source "
+                    f"reports {source_key}; select the contact for the physical source"
+                )
+        else:
+            if source_key not in contacts:
+                raise OtaError(
+                    f"the connected OTA source public key {source_key} is not in the "
+                    "controller's contacts; import its contact or advertise/discover it "
+                    "on the normal channel first. Renaming a different contact or passing "
+                    "--source-contact cannot replace this identity check. If source and "
+                    "controller are the same TCP Full Companion, use the verified "
+                    "--source-shares-controller topology instead"
+                )
+            source = source_key
+            print(f"[contact] OTA source -> {contacts[source].get('adv_name', '?')} [{source}]")
+    else:
+        source = resolve(source_selector) if source_selector else None
     keys = [target, *(key for key, _password in relays)]
     if source:
         keys.append(source)
@@ -2758,6 +2785,29 @@ def parse_temp_radio(value: str) -> tuple[float, float, int, int, int]:
     return freq, bandwidth, sf, cr, minutes
 
 
+def parse_source_terminal_banner(value: str) -> tuple[str, str]:
+    """Read one Full/Companion welcome identity, not an advert or cached name."""
+    text = value.replace("\r", "")
+    headers = list(re.finditer(
+        r"(?m)^===== MeshCore (?:Full )?Companion Terminal =====[ \t]*$", text,
+    ))
+    if len(headers) != 1:
+        raise OtaError("OTA source did not expose one fresh Companion terminal banner")
+    tail = text[headers[0].end():]
+    # meshcli raw output strips trailing spaces from a standalone USB prompt.
+    prompt = re.search(r"\n>(?:[ \t]|(?=\n|$))", tail)
+    if prompt is None:
+        raise OtaError("OTA source Companion banner did not reach its terminal prompt")
+    banner = tail[:prompt.start()]
+    keys = re.findall(r"(?m)^[0-9A-Fa-f]{64}$", banner)
+    welcome = re.findall(
+        r"(?:^|\n)WELCOME[^\n]*\n([0-9A-Fa-f]{64})\nCompanion [^\n]+", banner,
+    )
+    if len(keys) != 1 or welcome != keys:
+        raise OtaError("OTA source Companion banner did not expose one exact public key after WELCOME")
+    return keys[0].lower(), tail[prompt.end():]
+
+
 def source_cli_command(
     args: argparse.Namespace,
     command_text: str,
@@ -2766,6 +2816,7 @@ def source_cli_command(
     bounded: bool = False,
     deadline: float | None = None,
     retry: bool = True,
+    full_companion_identity: bool = False,
 ) -> str:
     if not command_text.strip() or any(c in command_text for c in "\r\n\0"):
         raise OtaError("source command must be one nonempty line")
@@ -2773,6 +2824,8 @@ def source_cli_command(
         raise OtaError(
             f"{command_text!r} requires state-aware retry handling"
         )
+    if full_companion_identity and command_text != "ver":
+        raise OtaError("a Companion identity probe must use the read-only ver command")
     serial_port = args.source_cli_serial or args.source_serial
     tcp_console = args.source_cli_tcp
     if not serial_port and not tcp_console:
@@ -2783,6 +2836,7 @@ def source_cli_command(
         return ""
 
     def run_once() -> str:
+        terminal_key = None
         operation_timeout = getattr(args, "source_cli_timeout", 30.0)
         if deadline is not None:
             remaining = deadline - time.monotonic()
@@ -2826,6 +2880,8 @@ def source_cli_command(
                     expected_terminal_key = getattr(
                         args, "shared_source_public_key", None
                     )
+                    if full_companion_identity:
+                        terminal_key, _ = parse_source_terminal_banner(greeting_text)
                     if expected_terminal_key is not None:
                         banner_keys = re.findall(
                             r"(?:^|\r?\n)([0-9A-Fa-f]{64})(?=\r?\n)",
@@ -2872,7 +2928,7 @@ def source_cli_command(
                 raise TransmissionError("source TCP console returned an empty reply")
         else:
             wire_command = command_text
-            if getattr(args, "source_companion_terminal", False):
+            if full_companion_identity or getattr(args, "source_companion_terminal", False):
                 # meshcli raw mode keeps one serial open while writing this
                 # compound command. STOP first makes this independent of the
                 # port's current state: ASCII consumes it and returns to
@@ -2905,6 +2961,16 @@ def source_cli_command(
         if ("error" in lowered or "unknown command" in lowered or "err " in lowered
                 or re.search(r"(?:^|\n)\s*(?:->\s*)?(?:err|fail)(?:\b|:)", lowered)):
             raise OtaError(f"source rejected {command_text!r}: {output}")
+        if full_companion_identity:
+            version_output = output.replace("\r", "")
+            if not tcp_console:
+                terminal_key, version_output = parse_source_terminal_banner(output)
+            replies = re.findall(r"(?m)^[ \t]*(?:>[ \t]*)?Companion [^\n]+", version_output)
+            if len(replies) != 1 or extract_reply_version(replies[0]) is None:
+                raise OtaError("OTA source Companion did not confirm a live ver reply after its identity banner")
+            # Normalize only after this same connection supplied a complete
+            # welcome key AND replied to ver. No key from a previous connection.
+            output = f"> {terminal_key}"
         if (
             command_text.startswith("tempradio ")
             and "ok - temp params " not in lowered
@@ -3006,6 +3072,7 @@ def preflight_source_cli(args: argparse.Namespace) -> None:
             "repeater raw text CLI, nRF52 full Companion USB port, or "
             "companion_radio_full TCP terminal."
         )
+    args.source_full_companion = "OTA seeder" in output and "install:disabled" in output
 
 
 def has_managed_source_cli(args: argparse.Namespace) -> bool:
@@ -4974,13 +5041,20 @@ def read_source_public_key_bounded(
     *,
     deadline: float | None = None,
 ) -> str:
+    if not getattr(args, "source_full_companion", False):
+        reply = optional_source_cli_command(args, "get public.key", deadline=deadline)
+        if reply is not None:
+            return parse_cli_public_key(reply, "OTA source")
+    # Full Companion has no repeater `get public.key` command. Its supported
+    # USB/TCP terminal supplies the local key in a fresh welcome banner instead.
+    # Unknown firmware can reach this path only after an explicit unsupported
+    # reply, never after a timeout, access denial, or malformed key.
     reply = source_cli_command(
-        args,
-        "get public.key",
-        bounded=True,
-        deadline=deadline,
+        args, "ver", bounded=True, deadline=deadline, full_companion_identity=True,
     )
-    return parse_cli_public_key(reply, "OTA source")
+    key = parse_cli_public_key(reply, "OTA source")
+    args.source_full_companion = True
+    return key
 
 
 def prove_shared_source_terminal_bounded(
@@ -6938,6 +7012,12 @@ def build_parser() -> argparse.ArgumentParser:
         help=argparse.SUPPRESS,
     )
     parser.add_argument("--meshcli", default="meshcli")
+    parser.add_argument(
+        "--cargo", help="Cargo executable for motatool repair (default: find a compatible installed toolchain)",
+    )
+    parser.add_argument(
+        "--rustc", help="Rust compiler for motatool repair (default: pair with selected Cargo)",
+    )
     parser.add_argument(
         "--motatool", default="motatool",
         help=(
@@ -7480,6 +7560,195 @@ def offer_cryptography_repair(failure: BootloaderCryptoError) -> None:
     print("[repair] Ed25519 self-test passed; retrying full package/signature verification.")
 
 
+@dataclass(frozen=True)
+class RustBuildTools:
+    cargo: str
+    rustc: str
+    cargo_version: tuple[int, int, int]
+    rustc_version: tuple[int, int, int]
+
+    def environment(self) -> dict[str, str]:
+        env = rust_probe_environment()
+        # Pin both executables for this child process. In particular, a versioned
+        # distro Cargo must not fall back to the old unversioned rustc on PATH.
+        env["CARGO"] = self.cargo
+        env["RUSTC"] = self.rustc
+        return env
+
+
+def rust_probe_environment() -> dict[str, str]:
+    env = os.environ.copy()
+    env.pop("MESHCORE_ADMIN_PASSWORD", None)
+    env["RUSTUP_AUTO_INSTALL"] = "0"
+    return env
+
+
+def rust_host_output(command: list[str]) -> str:
+    """Bounded, read-only probes; never invoke rustup install/update/default."""
+    try:
+        result = subprocess.run(
+            command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE, text=True, check=False,
+            timeout=RUST_TOOL_PROBE_TIMEOUT_SECONDS, env=rust_probe_environment(),
+        )
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        raise OtaError(f"cannot probe {command[0]}: {exc}") from exc
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout).strip()[:1000]
+        raise OtaError(f"cannot probe {command[0]} (exit {result.returncode}): {detail}")
+    return result.stdout.strip()
+
+
+def rust_tool_version(command: str, tool: str) -> tuple[int, int, int]:
+    output = rust_host_output([command, "--version"])
+    match = re.match(rf"^{tool} (\d+)\.(\d+)\.(\d+)(?=\s|$)", output)
+    if match is None:
+        raise OtaError(f"cannot determine a stable {tool} version from {command}: {output[:200]}")
+    version = tuple(int(part) for part in match.groups())
+    if version < MOTATOOL_MIN_RUST_VERSION:
+        needed = ".".join(map(str, MOTATOOL_MIN_RUST_VERSION))
+        raise OtaError(f"{command} reports {tool} {'.'.join(map(str, version))}; need {needed} or newer")
+    return version
+
+
+def rust_executable(command: str) -> str:
+    path = shutil.which(command)
+    if path is None:
+        raise OtaError(f"Rust tool executable not found: {command}")
+    # Keep argv[0] intact for symlink/hardlink rustup proxies.
+    return str(Path(path).absolute())
+
+
+def resolve_rustup_proxy(command: str, tool: str) -> str:
+    rustup = shutil.which("rustup")
+    if rustup is not None:
+        try:
+            is_proxy = os.path.samefile(command, rustup)
+        except OSError:
+            is_proxy = False
+        if is_proxy:
+            # Resolve in the caller's context, then retain the real binary path:
+            # changing to the build cache must not change the active toolchain.
+            return rust_executable(rust_host_output([rustup, "which", tool]))
+    return command
+
+
+def versioned_cargo_candidates() -> list[str]:
+    found: dict[str, tuple[int, ...]] = {}
+    for directory in dict.fromkeys(os.get_exec_path()):
+        if not directory:
+            continue
+        try:
+            for entry in Path(directory).iterdir():
+                match = re.fullmatch(
+                    r"cargo-(\d+\.\d+(?:\.\d+)?)(?:\.exe)?", entry.name,
+                    flags=re.IGNORECASE if os.name == "nt" else 0,
+                )
+                if match and entry.is_file() and os.access(entry, os.X_OK):
+                    found[str(entry.absolute())] = tuple(map(int, match[1].split(".")))
+        except OSError:
+            continue
+    return sorted(found, key=lambda path: found[path], reverse=True)[:16]
+
+
+def installed_rustup_cargos() -> list[str]:
+    rustup = shutil.which("rustup")
+    if rustup is None:
+        return []
+    result = []
+    for line in rust_host_output([rustup, "toolchain", "list"]).splitlines():
+        name = line.split()[0] if line.strip() else ""
+        if not re.fullmatch(r"(?:stable|\d+\.\d+\.\d+)(?:-[A-Za-z0-9_-]+)?", name):
+            continue  # Do not select nightly/beta or arbitrary custom toolchains.
+        try:
+            result.append(rust_host_output([rustup, "which", "--toolchain", name, "cargo"]))
+        except OtaError:
+            continue  # An incompletely installed toolchain is not usable.
+        if len(result) == 16:
+            break
+    return result
+
+
+def select_motatool_build_tools(args: argparse.Namespace) -> RustBuildTools:
+    explicit_cargo = getattr(args, "cargo", None)
+    explicit_rustc = (
+        getattr(args, "rustc", None) or os.environ.get("RUSTC")
+        or os.environ.get("CARGO_BUILD_RUSTC")
+    )
+    checked: set[str] = set()
+    failures: list[str] = []
+
+    def check(candidate: str) -> RustBuildTools | None:
+        try:
+            cargo = resolve_rustup_proxy(rust_executable(candidate), "cargo")
+            if cargo in checked:
+                return None
+            checked.add(cargo)
+            cargo_version = rust_tool_version(cargo, "cargo")
+            compiler_candidates = [explicit_rustc] if explicit_rustc else []
+            if not explicit_rustc:
+                name = re.sub(r"^cargo", "rustc", Path(cargo).name, count=1, flags=re.IGNORECASE)
+                compiler_candidates = [str(Path(cargo).with_name(name)), name,
+                    f"rustc-{cargo_version[0]}.{cargo_version[1]}", "rustc"]
+            compiler_errors = []
+            seen_compilers: set[str] = set()
+            for compiler in dict.fromkeys(compiler_candidates):
+                try:
+                    rustc = resolve_rustup_proxy(rust_executable(compiler), "rustc")
+                    if rustc in seen_compilers:
+                        continue
+                    seen_compilers.add(rustc)
+                    rustc_version = rust_tool_version(rustc, "rustc")
+                    if cargo_version[:2] != rustc_version[:2]:
+                        raise OtaError(
+                            f"toolchain mismatch: Cargo {'.'.join(map(str, cargo_version))} at {cargo}, "
+                            f"rustc {'.'.join(map(str, rustc_version))} at {rustc}; select matching releases"
+                        )
+                    break
+                except OtaError as exc:
+                    compiler_errors.append(str(exc))
+            else:
+                raise OtaError(" | ".join(compiler_errors))
+            tools = RustBuildTools(cargo, rustc, cargo_version, rustc_version)
+            print(f"[host] build Cargo {'.'.join(map(str, cargo_version))}: {cargo}")
+            print(f"[host] build rustc {'.'.join(map(str, rustc_version))}: {rustc}")
+            return tools
+        except OtaError as exc:
+            failures.append(str(exc))
+            print(f"[host] {exc}")
+            return None
+
+    selected = check(explicit_cargo or "cargo")
+    if selected is not None:
+        return selected
+    if not explicit_cargo:
+        for candidate in versioned_cargo_candidates():
+            selected = check(candidate)
+            if selected is not None:
+                return selected
+        try:
+            installed = installed_rustup_cargos()
+        except OtaError as exc:
+            failures.append(str(exc))
+            installed = []
+        for candidate in installed:
+            selected = check(candidate)
+            if selected is not None:
+                return selected
+    needed = ".".join(map(str, MOTATOOL_MIN_RUST_VERSION))
+    raise OtaError(
+        f"motatool repair needs matching stable Cargo and rustc {needed} or newer "
+        "for its pinned dependencies, plus a native linker. "
+        "Select an installed pair with --cargo cargo-1.91 --rustc rustc-1.91, "
+        "or install a current stable user toolchain from https://rustup.rs "
+        "(with rustup already installed: rustup toolchain install stable --profile minimal). "
+        "Root access and nightly are not required for this version issue. "
+        "Shell aliases do not affect this script; explicit --cargo/--rustc and compiler "
+        "environment overrides are not silently replaced. No build or radio changes "
+        "were made. " + " | ".join(failures[-4:])
+    )
+
+
 def motatool_repair_command(root: Path, cargo: str = "cargo") -> list[str]:
     return [
         cargo, "install", "--git", MOTATOOL_REPAIR_REPOSITORY,
@@ -7520,9 +7789,14 @@ def offer_motatool_repair(
         print(f"[host] using previously installed, rechecked motatool: {binary}")
         return
 
-    cargo = shutil.which("cargo")
-    command = motatool_repair_command(root, cargo or "cargo")
+    build_tools = None if cached else select_motatool_build_tools(args)
+    command = motatool_repair_command(root, build_tools.cargo if build_tools else "cargo")
     manual = display_host_command(command)
+    if build_tools is not None:
+        if os.name == "nt":
+            manual = "$env:RUSTC = '" + build_tools.rustc.replace("'", "''") + "'; " + manual
+        else:
+            manual = "RUSTC=" + shlex.quote(build_tools.rustc) + " " + manual
     selection = display_host_command(["--motatool", str(binary)])
     # Display an argument, not a PowerShell call operator for an option.
     if os.name == "nt":
@@ -7533,12 +7807,6 @@ def offer_motatool_repair(
     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 "
@@ -7565,8 +7833,8 @@ def offer_motatool_repair(
         # 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)
+        assert build_tools is not None
+        build_env = build_tools.environment()
         try:
             result = subprocess.run(
                 command, cwd=str(root), stdin=subprocess.DEVNULL,

tools/lora_ota/test_lora_ota_bootloader.py

@@ -304,6 +304,9 @@ class BootloaderStagingTests(unittest.TestCase):
                 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, "select_motatool_build_tools", return_value=ota.RustBuildTools(
+                    "cargo", "rustc", (1, 91, 0), (1, 91, 0),
+                )),
                 mock.patch.object(ota.sys.stdin, "isatty", return_value=False),
                 self.assertRaisesRegex(ota.OtaError, "separate interactive approval"),
             ):
@@ -350,7 +353,7 @@ class BootloaderStagingTests(unittest.TestCase):
                     "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_public_key_bounded", return_value=source_key))
                 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(
@@ -428,6 +431,84 @@ class StationSelectionTests(unittest.TestCase):
         with contextlib.redirect_stdout(io.StringIO()), self.assertRaisesRegex(ota.OtaError, "different radios"):
             ota.bind_contact_selectors(self.controller(), args)
 
+    def test_source_binds_by_local_key_despite_stale_or_emoji_name(self):
+        source_key = "b2" * 32
+        controller = self.controller()
+        controller._run.return_value[0][source_key] = {
+            "public_key": source_key, "adv_name": "Old source name \U0001f4e1",
+        }
+        args = self.args(self.name)
+        args.source_shares_controller = False
+        args.source_cli_tcp = "source:5002"
+        with (
+            mock.patch.object(ota, "read_source_public_key_bounded", return_value=source_key.upper()) as read_key,
+            mock.patch.object(ota, "read_source_name_bounded") as read_name,
+            contextlib.redirect_stdout(io.StringIO()),
+        ):
+            ota.bind_contact_selectors(controller, args)
+        self.assertEqual(args.source_contact_value, source_key)
+        read_key.assert_called_once_with(args)
+        read_name.assert_not_called()
+        controller.remote_command.assert_not_called()
+
+    def test_missing_source_key_does_not_bind_same_named_other_radio(self):
+        args = self.args(self.name)
+        args.source_shares_controller = False
+        args.source_serial = "source"
+        source_key = "b2" * 32
+        controller = self.controller()
+        controller._run.return_value[0]["c3" * 32] = {
+            "public_key": "c3" * 32, "adv_name": "W4JEC MC OTA2",
+        }
+        with (
+            mock.patch.object(ota, "read_source_public_key_bounded", return_value=source_key),
+            contextlib.redirect_stdout(io.StringIO()),
+            self.assertRaisesRegex(ota.OtaError, "connected OTA source public key.*not in") as error,
+        ):
+            ota.bind_contact_selectors(controller, args)
+        self.assertIn(source_key, str(error.exception))
+        self.assertIn("normal channel", str(error.exception))
+        controller.remote_command.assert_not_called()
+
+    def test_explicit_source_contact_must_match_connected_source_key(self):
+        args = self.args(self.name)
+        args.source_shares_controller = False
+        args.source_serial = "source"
+        args.source_contact_value = "Saved source"
+        controller = self.controller()
+        controller._run.return_value[0]["c3" * 32] = {
+            "public_key": "c3" * 32, "adv_name": "Saved source",
+        }
+        with (
+            mock.patch.object(ota, "read_source_public_key_bounded", return_value="b2" * 32),
+            contextlib.redirect_stdout(io.StringIO()),
+            self.assertRaisesRegex(ota.OtaError, "--source-contact identifies.*connected OTA source"),
+        ):
+            ota.bind_contact_selectors(controller, args)
+        controller.remote_command.assert_not_called()
+
+    def test_auto_source_cannot_also_be_destination(self):
+        args = self.args(self.name)
+        args.source_shares_controller = False
+        args.source_serial = "source"
+        with (
+            mock.patch.object(ota, "read_source_public_key_bounded", return_value=self.key),
+            contextlib.redirect_stdout(io.StringIO()),
+            self.assertRaisesRegex(ota.OtaError, "different radios"),
+        ):
+            ota.bind_contact_selectors(self.controller(), args)
+
+    def test_shared_source_does_not_require_its_own_contact_or_repeater_cli(self):
+        args = self.args(self.name)
+        args.source_cli_tcp = "shared:5002"
+        with (
+            mock.patch.object(ota, "read_source_public_key_bounded") as read_key,
+            contextlib.redirect_stdout(io.StringIO()),
+        ):
+            ota.bind_contact_selectors(self.controller(), args)
+        read_key.assert_not_called()
+        self.assertIsNone(args.source_contact_value)
+
     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)

tools/lora_ota/test_lora_ota_rust_tools.py

@@ -0,0 +1,257 @@
+"""Toolchain preflight regressions; fake executables are never actually run."""
+
+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
+
+
+class RustToolSelectionTests(unittest.TestCase):
+    def setUp(self):
+        directory = tempfile.TemporaryDirectory()
+        self.addCleanup(directory.cleanup)
+        self.root = Path(directory.name)
+        self.outputs = {}
+        self.args = argparse.Namespace(cargo=None, rustc=None)
+        stack = contextlib.ExitStack()
+        self.addCleanup(stack.close)
+        stack.enter_context(mock.patch.dict(os.environ, {
+            "PATH": str(self.root), "PATHEXT": ".com;.exe;.bat;.cmd",
+        }, clear=True))
+        self.probe = stack.enter_context(mock.patch.object(ota.subprocess, "run", side_effect=self.run_probe))
+        self.rustup = stack.enter_context(mock.patch.object(ota, "installed_rustup_cargos", return_value=[]))
+        stack.enter_context(mock.patch.object(ota, "resolve_rustup_proxy", side_effect=lambda path, _: path))
+        self.output = stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
+
+    def executable(self, name, version):
+        suffix = ".exe" if os.name == "nt" else ""
+        path = self.root / (name + suffix)
+        path.write_bytes(b"test placeholder: must never execute")
+        path.chmod(0o755)
+        tool = "cargo" if name.startswith("cargo") else "rustc"
+        self.outputs[str(path)] = f"{tool} {version} (test-only build)"
+        return str(path)
+
+    def pair(self, version, suffix=""):
+        return self.executable("cargo" + suffix, version), self.executable("rustc" + suffix, version)
+
+    def run_probe(self, command, **kwargs):
+        self.assertEqual(command[1:], ["--version"])
+        self.assertIn(command[0], self.outputs)
+        return subprocess.CompletedProcess(command, 0, self.outputs[command[0]], "")
+
+    def select(self):
+        return ota.select_motatool_build_tools(self.args)
+
+    def test_compatible_default_pair_is_preserved(self):
+        cargo, rustc = self.pair("1.91.0")
+        self.pair("1.98.0", "-1.98")
+        selected = self.select()
+        self.assertEqual((selected.cargo, selected.rustc), (cargo, rustc))
+        self.rustup.assert_not_called()
+        self.assertEqual(self.probe.call_count, 2)
+
+    def test_reported_ubuntu_old_default_selects_versioned_pair(self):
+        old_cargo, _ = self.pair("1.75.0")
+        cargo, rustc = self.pair("1.91.0", "-1.91")
+        selected = self.select()
+        self.assertEqual((selected.cargo, selected.rustc), (cargo, rustc))
+        self.assertEqual(self.probe.call_args_list[0].args[0][0], old_cargo)
+        self.assertIn("1.75.0", self.output.getvalue())
+        self.assertEqual(selected.environment()["RUSTC"], rustc)
+        self.assertEqual(selected.environment()["CARGO"], cargo)
+
+    def test_new_cargo_with_old_rustc_is_not_accepted(self):
+        self.executable("cargo", "1.91.0")
+        self.executable("rustc", "1.75.0")
+        with self.assertRaisesRegex(ota.OtaError, "rustc 1.75.0"):
+            self.select()
+
+    def test_edition_2024_alone_is_not_enough_for_icu_dependencies(self):
+        self.pair("1.85.0")
+        with self.assertRaisesRegex(ota.OtaError, "1.86.0 or newer"):
+            self.select()
+
+    def test_minimum_supported_pair_is_accepted(self):
+        self.pair("1.86.0")
+        self.assertEqual(self.select().rustc_version, (1, 86, 0))
+
+    def test_incompatible_major_minor_pair_is_rejected(self):
+        self.executable("cargo", "1.91.0")
+        self.executable("rustc", "1.90.0")
+        with self.assertRaisesRegex(ota.OtaError, "toolchain mismatch"):
+            self.select()
+
+    def test_patch_release_difference_is_allowed(self):
+        self.executable("cargo", "1.91.1")
+        self.executable("rustc", "1.91.0")
+        self.assertEqual(self.select().cargo_version, (1, 91, 1))
+
+    def test_missing_unversioned_cargo_can_use_versioned_pair(self):
+        cargo, _ = self.pair("1.91.0", "-1.91")
+        self.assertEqual(self.select().cargo, cargo)
+
+    def test_versioned_cargo_does_not_fall_back_to_old_unversioned_rustc(self):
+        self.executable("cargo-1.91", "1.91.0")
+        self.executable("rustc", "1.75.0")
+        with self.assertRaisesRegex(ota.OtaError, "rustc-1.91"):
+            self.select()
+
+    def test_versioned_candidates_are_ranked_numerically(self):
+        self.pair("1.9.0", "-1.9")
+        self.pair("1.91.0", "-1.91")
+        cargo, _ = self.pair("1.100.0", "-1.100")
+        self.assertEqual(self.select().cargo, cargo)
+
+    def test_versioned_cargo_can_use_compatible_unversioned_compiler(self):
+        self.executable("cargo-1.91", "1.91.0")
+        rustc = self.executable("rustc", "1.91.0")
+        self.assertEqual(self.select().rustc, rustc)
+
+    def test_new_default_cargo_can_find_matching_versioned_compiler(self):
+        self.executable("cargo", "1.91.0")
+        self.executable("rustc", "1.75.0")
+        rustc = self.executable("rustc-1.91", "1.91.0")
+        self.assertEqual(self.select().rustc, rustc)
+
+    def test_explicit_cargo_is_not_replaced(self):
+        self.args.cargo, _ = self.pair("1.75.0")
+        self.pair("1.91.0", "-1.91")
+        with self.assertRaisesRegex(ota.OtaError, "not silently replaced"):
+            self.select()
+        self.assertEqual(self.probe.call_count, 1)
+        self.rustup.assert_not_called()
+
+    def test_explicit_versioned_pair_is_used(self):
+        cargo, rustc = self.pair("1.91.0", "-1.91")
+        self.args.cargo = "cargo-1.91"
+        self.args.rustc = "rustc-1.91"
+        self.assertEqual(self.select().rustc, rustc)
+        self.assertEqual(self.select().cargo, cargo)
+
+    def test_explicit_missing_cargo_never_falls_back(self):
+        self.pair("1.91.0")
+        self.args.cargo = "missing-cargo"
+        with self.assertRaisesRegex(ota.OtaError, "missing-cargo"):
+            self.select()
+        self.probe.assert_not_called()
+
+    def test_explicit_old_compiler_environment_is_not_silently_overridden(self):
+        self.pair("1.91.0")
+        compiler = self.executable("rustc-1.75", "1.75.0")
+        for variable in ("RUSTC", "CARGO_BUILD_RUSTC"):
+            with self.subTest(variable=variable), mock.patch.dict(os.environ, {variable: compiler}):
+                with self.assertRaisesRegex(ota.OtaError, "rustc 1.75.0"):
+                    self.select()
+
+    def test_cli_compiler_takes_precedence_over_environment(self):
+        self.pair("1.91.0")
+        old = self.executable("rustc-1.75", "1.75.0")
+        self.args.rustc = "rustc"
+        with mock.patch.dict(os.environ, {"RUSTC": old}):
+            self.assertEqual(self.select().rustc_version, (1, 91, 0))
+
+    def test_installed_rustup_toolchain_is_fallback(self):
+        self.pair("1.75.0")
+        tools = self.root / "installed-stable" / "bin"
+        tools.mkdir(parents=True)
+        cargo, rustc = self.pair("1.91.0", "-1.91")
+        actual_cargo = tools / Path(cargo).name.replace("-1.91", "")
+        actual_rustc = tools / Path(rustc).name.replace("-1.91", "")
+        Path(cargo).rename(actual_cargo)
+        Path(rustc).rename(actual_rustc)
+        self.outputs[str(actual_cargo)] = self.outputs[cargo]
+        self.outputs[str(actual_rustc)] = self.outputs[rustc]
+        self.rustup.return_value = [str(actual_cargo)]
+        self.assertEqual(self.select().rustc, str(actual_rustc))
+
+    def test_malformed_version_fails_closed(self):
+        cargo, _ = self.pair("unknown")
+        with self.assertRaisesRegex(ota.OtaError, "cannot determine a stable cargo version"):
+            self.select()
+        self.assertIn(cargo, self.output.getvalue())
+
+    def test_nightly_is_not_automatically_selected(self):
+        self.pair("1.91.0-nightly")
+        with self.assertRaisesRegex(ota.OtaError, "stable cargo version"):
+            self.select()
+
+    def test_no_toolchain_explains_non_root_stable_fix(self):
+        with self.assertRaises(ota.OtaError) as error:
+            self.select()
+        message = str(error.exception)
+        self.assertIn("--cargo cargo-1.91 --rustc rustc-1.91", message)
+        self.assertIn("Root access and nightly are not required", message)
+        self.assertIn("No build or radio changes", message)
+
+
+class RustProbeTests(unittest.TestCase):
+    def test_probe_is_bounded_noninteractive_and_strips_admin_secret(self):
+        result = subprocess.CompletedProcess([], 0, "cargo 1.91.0\n", "")
+        with (
+            mock.patch.object(ota.subprocess, "run", return_value=result) as run,
+            mock.patch.dict(os.environ, {"MESHCORE_ADMIN_PASSWORD": "secret"}),
+        ):
+            self.assertEqual(ota.rust_host_output(["cargo", "--version"]), "cargo 1.91.0")
+        options = run.call_args.kwargs
+        self.assertEqual(options["timeout"], ota.RUST_TOOL_PROBE_TIMEOUT_SECONDS)
+        self.assertEqual(options["stdin"], subprocess.DEVNULL)
+        self.assertEqual(options["env"]["RUSTUP_AUTO_INSTALL"], "0")
+        self.assertNotIn("MESHCORE_ADMIN_PASSWORD", options["env"])
+
+    def test_probe_timeout_and_missing_executable_are_actionable(self):
+        for error in (FileNotFoundError("not installed"), subprocess.TimeoutExpired("cargo", 10)):
+            with mock.patch.object(ota.subprocess, "run", side_effect=error):
+                with self.assertRaisesRegex(ota.OtaError, "cannot probe cargo"):
+                    ota.rust_host_output(["cargo", "--version"])
+
+    def test_failed_probe_preserves_diagnostic(self):
+        result = subprocess.CompletedProcess([], 1, "", "toolchain not installed")
+        with mock.patch.object(ota.subprocess, "run", return_value=result):
+            with self.assertRaisesRegex(ota.OtaError, "toolchain not installed"):
+                ota.rust_host_output(["cargo", "--version"])
+
+    def test_rustup_discovery_only_uses_installed_stable_toolchains(self):
+        def output(command):
+            if command[1:] == ["toolchain", "list"]:
+                return "stable-host (default)\n1.91.0-host\nnightly-host\nbeta-host\ncustom\n"
+            return "/installed/" + command[3] + "/cargo"
+        with (
+            mock.patch.object(ota.shutil, "which", return_value="rustup"),
+            mock.patch.object(ota, "rust_host_output", side_effect=output) as probe,
+        ):
+            self.assertEqual(ota.installed_rustup_cargos(), [
+                "/installed/stable-host/cargo", "/installed/1.91.0-host/cargo",
+            ])
+        self.assertEqual(probe.call_count, 3)
+        self.assertTrue(all(call.args[0][1] in ("toolchain", "which") for call in probe.call_args_list))
+
+    def test_proxy_is_resolved_to_real_binary_before_build_cwd_changes(self):
+        with (
+            mock.patch.object(ota.shutil, "which", return_value="rustup"),
+            mock.patch.object(ota.os.path, "samefile", return_value=True),
+            mock.patch.object(ota, "rust_host_output", return_value="/installed/stable/bin/cargo") as probe,
+            mock.patch.object(ota, "rust_executable", side_effect=lambda path: path),
+        ):
+            self.assertEqual(ota.resolve_rustup_proxy("proxy/cargo", "cargo"), "/installed/stable/bin/cargo")
+        probe.assert_called_once_with(["rustup", "which", "cargo"])
+
+    def test_build_environment_is_private_and_pins_both_binaries(self):
+        pair = ota.RustBuildTools("/new/cargo", "/new/rustc", (1, 91, 0), (1, 91, 0))
+        with mock.patch.dict(os.environ, {"CARGO": "old-cargo", "RUSTC": "old-rustc"}):
+            env = pair.environment()
+            self.assertEqual(env["CARGO"], "/new/cargo")
+            self.assertEqual(env["RUSTC"], "/new/rustc")
+            self.assertEqual(os.environ["CARGO"], "old-cargo")
+            self.assertEqual(os.environ["RUSTC"], "old-rustc")
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/lora_ota/test_lora_ota_source_identity.py

@@ -0,0 +1,213 @@
+"""Companion USB/TCP identity compatibility tests; no radios or transmissions."""
+
+import argparse
+import contextlib
+import io
+import subprocess
+import unittest
+from unittest import mock
+
+import lora_ota as ota
+
+
+KEY = "a5" * 32
+VERSION = "Companion v1.17.1.5 (protocol 1, build test)"
+
+
+def banner(key=KEY, name="W4JEC MC OTA2", *, full=True):
+    role = "Full Companion" if full else "Companion"
+    return (
+        f"\r\n===== MeshCore {role} Terminal =====\r\n\r\n"
+        f"WELCOME  {name}\r\n{key}\r\nCompanion v1.17.1.5\r\n"
+        "  (enter 'help' for commands)\r\n\r\n> "
+    )
+
+
+def connection(greeting=None, reply=VERSION, *, chunks=None):
+    conn = mock.MagicMock()
+    conn.__enter__.return_value = conn
+    conn.recv.side_effect = chunks or [
+        (greeting if greeting is not None else banner()).encode("utf-8"),
+        (reply + "\r\n> ").encode("utf-8"),
+    ]
+    return conn
+
+
+class SourceIdentityTests(unittest.TestCase):
+    def setUp(self):
+        self.args = argparse.Namespace(
+            source_cli_serial=None, source_serial=None,
+            source_cli_tcp="192.0.2.10:5002", source_baud=115200,
+            meshcli="meshcli", source_full_companion=True,
+        )
+        stack = contextlib.ExitStack()
+        self.addCleanup(stack.close)
+        stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
+
+    def test_full_tcp_uses_fresh_banner_and_supported_ver_not_get_public_key(self):
+        conn = connection()
+        with mock.patch.object(ota.socket, "create_connection", return_value=conn):
+            self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY)
+        conn.sendall.assert_called_once_with(b"ver\r\n")
+
+    def test_tcp_fragmented_banner_and_version_reply(self):
+        greeting = banner(KEY.upper()).encode("utf-8")
+        conn = connection(chunks=[greeting[:41], greeting[41:103], greeting[103:], VERSION.encode(), b"\r\n> "])
+        with mock.patch.object(ota.socket, "create_connection", return_value=conn):
+            self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY)
+
+    def test_tcp_probe_does_not_reuse_previous_connection_identity(self):
+        first, second = connection(), connection(banner("b6" * 32))
+        with mock.patch.object(ota.socket, "create_connection", side_effect=[first, second]) as connect:
+            self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY)
+            self.assertEqual(ota.read_source_public_key_bounded(self.args), "b6" * 32)
+        self.assertEqual(connect.call_count, 2)
+
+    def test_usb_ascii_first_full_source_gets_fresh_wrapped_banner(self):
+        self.args.source_cli_tcp = None
+        self.args.source_serial = "COM42"
+        self.args.source_companion_terminal = False
+        result = subprocess.CompletedProcess([], 0, banner() + VERSION + "\r\n> \r\nBinary mode\r\n", "")
+        with mock.patch.object(ota, "run_checked", return_value=result) as run:
+            self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY)
+        command = run.call_args.args[0]
+        self.assertIn("COM42", command)
+        self.assertEqual(command[-1],
+            f"{ota.COMPANION_TERMINAL_STOP}\r{ota.COMPANION_TERMINAL_START}\rver\r{ota.COMPANION_TERMINAL_STOP}\r")
+        self.assertNotIn("get public.key", command[-1])
+        self.assertFalse(self.args.source_companion_terminal)  # No persistent mode flag change.
+
+    def test_usb_can_use_explicit_cli_port(self):
+        self.args.source_cli_tcp = None
+        self.args.source_cli_serial = "COM43"
+        self.args.source_serial = "COM42"
+        result = subprocess.CompletedProcess([], 0, banner() + VERSION + "\r\n", "")
+        with mock.patch.object(ota, "run_checked", return_value=result) as run:
+            self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY)
+        self.assertIn("COM43", run.call_args.args[0])
+        self.assertNotIn("COM42", run.call_args.args[0])
+
+    def test_usb_meshcli_trimmed_prompt_is_accepted(self):
+        self.args.source_cli_tcp = None
+        self.args.source_serial = "COM42"
+        rendered = "\n".join(line.rstrip() for line in (banner() + "\r\n" + VERSION).split("\n"))
+        result = subprocess.CompletedProcess([], 0, rendered, "")
+        with mock.patch.object(ota, "run_checked", return_value=result):
+            self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY)
+
+    def test_usb_banner_alone_is_not_a_live_version_reply(self):
+        self.args.source_cli_tcp = None
+        self.args.source_serial = "COM42"
+        result = subprocess.CompletedProcess([], 0, banner() + "Binary mode\r\n", "")
+        with mock.patch.object(ota, "run_checked", return_value=result):
+            with self.assertRaisesRegex(ota.OtaError, "live ver reply"):
+                ota.read_source_public_key_bounded(self.args)
+
+    def test_unknown_command_falls_back_only_to_fresh_companion_identity(self):
+        self.args.source_full_companion = False
+        legacy = connection(reply="ERROR: unknown command: get public.key")
+        companion = connection()
+        with mock.patch.object(ota.socket, "create_connection", side_effect=[legacy, companion]):
+            self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY)
+        legacy.sendall.assert_called_once_with(b"get public.key\r\n")
+        companion.sendall.assert_called_once_with(b"ver\r\n")
+        self.assertTrue(self.args.source_full_companion)
+
+    def test_repeater_keeps_its_supported_get_public_key_command(self):
+        self.args.source_full_companion = False
+        conn = connection("Repeater console\r\n> ", "> " + KEY)
+        with mock.patch.object(ota.socket, "create_connection", return_value=conn):
+            self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY)
+        conn.sendall.assert_called_once_with(b"get public.key\r\n")
+
+    def test_timeouts_and_denials_do_not_trigger_companion_fallback(self):
+        self.args.source_full_companion = False
+        for error in (ota.TransmissionError("timeout"), ota.OtaError("permission denied")):
+            with self.subTest(error=error), mock.patch.object(ota, "source_cli_command", side_effect=error) as command:
+                with self.assertRaises(type(error)):
+                    ota.read_source_public_key_bounded(self.args)
+                command.assert_called_once_with(self.args, "get public.key", bounded=True, deadline=None)
+
+    def test_malformed_repeater_key_does_not_trigger_fallback(self):
+        self.args.source_full_companion = False
+        with mock.patch.object(ota, "source_cli_command", return_value="> incomplete") as command:
+            with self.assertRaisesRegex(ota.OtaError, "exact public key"):
+                ota.read_source_public_key_bounded(self.args)
+        command.assert_called_once()
+
+    def test_full_tcp_does_not_accept_wrong_or_multiple_version_replies(self):
+        for reply in ("OK", "Companion not-a-version", VERSION + "\r\n" + VERSION):
+            with self.subTest(reply=reply), mock.patch.object(ota.socket, "create_connection", return_value=connection(reply=reply)):
+                with self.assertRaisesRegex(ota.OtaError, "live ver reply"):
+                    ota.read_source_public_key_bounded(self.args)
+
+    def test_missing_or_ambiguous_banner_stops_before_sending_command(self):
+        for greeting in (
+            "Old OTA console\r\n> ", banner().replace(KEY, "incomplete"),
+            banner().replace(KEY, KEY + "\r\n" + "b6" * 32),
+            banner().replace("WELCOME", "RECEIVED ADVERT"),
+        ):
+            conn = connection(greeting)
+            with self.subTest(greeting=greeting), mock.patch.object(ota.socket, "create_connection", return_value=conn):
+                with self.assertRaises(ota.OtaError):
+                    ota.read_source_public_key_bounded(self.args)
+            conn.sendall.assert_not_called()
+
+    def test_shared_tcp_expected_key_mismatch_still_stops_before_command(self):
+        self.args.shared_source_public_key = "b6" * 32
+        conn = connection()
+        with mock.patch.object(ota.socket, "create_connection", return_value=conn):
+            with self.assertRaisesRegex(ota.OtaError, "terminal identity mismatch"):
+                ota.read_source_public_key_bounded(self.args)
+        conn.sendall.assert_not_called()
+
+    def test_status_detects_full_role_and_contact_binding_uses_supported_identity(self):
+        self.args.source_full_companion = False
+        self.args.target = "destination"
+        self.args.relay_values = []
+        self.args.source_shares_controller = False
+        self.args.source_contact_value = None
+        target_key = "c7" * 32
+        controller = mock.Mock()
+        controller._run.return_value = [{
+            target_key: {"public_key": target_key, "adv_name": "destination"},
+            KEY: {"public_key": KEY, "adv_name": "Saved old source name"},
+        }]
+        status = connection(reply="OTA seeder | install:disabled | serving:0")
+        identity = connection(banner(name="W4JEC MC OTA2 \U0001f4e1"))
+        with mock.patch.object(ota.socket, "create_connection", side_effect=[status, identity]):
+            ota.preflight_source_cli(self.args)
+            self.assertTrue(self.args.source_full_companion)
+            ota.bind_contact_selectors(controller, self.args)
+        self.assertEqual(self.args.source_contact_value, KEY)
+        status.sendall.assert_called_once_with(b"ota status\r\n")
+        identity.sendall.assert_called_once_with(b"ver\r\n")
+        controller.remote_command.assert_not_called()
+
+    def test_deadline_is_forwarded_to_companion_probe(self):
+        with mock.patch.object(ota, "source_cli_command", return_value="> " + KEY) as command:
+            self.assertEqual(ota.read_source_public_key_bounded(self.args, deadline=123.0), KEY)
+        command.assert_called_once_with(self.args, "ver", bounded=True, deadline=123.0, full_companion_identity=True)
+
+    def test_identity_mode_refuses_non_read_only_commands(self):
+        with self.assertRaisesRegex(ota.OtaError, "read-only ver"):
+            ota.source_cli_command(self.args, "reboot", full_companion_identity=True)
+
+
+class TerminalBannerTests(unittest.TestCase):
+    def test_both_banner_roles_and_emoji_names(self):
+        for full in (True, False):
+            value = banner(KEY.upper(), "Name \U0001f4e1", full=full)
+            self.assertEqual(ota.parse_source_terminal_banner(value + VERSION), (KEY, VERSION))
+
+    def test_duplicate_banners_are_not_one_identity(self):
+        with self.assertRaisesRegex(ota.OtaError, "one fresh"):
+            ota.parse_source_terminal_banner(banner() + banner())
+
+    def test_banner_without_prompt_is_incomplete(self):
+        with self.assertRaisesRegex(ota.OtaError, "prompt"):
+            ota.parse_source_terminal_banner(banner().removesuffix("> "))
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/lora_ota/test_lora_ota_tool_repair.py

@@ -29,6 +29,10 @@ class MotatoolRepairTests(unittest.TestCase):
         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"))
+        self.build_tools = ota.RustBuildTools("cargo", "rustc", (1, 91, 0), (1, 91, 0))
+        self.select_tools = stack.enter_context(mock.patch.object(
+            ota, "select_motatool_build_tools", return_value=self.build_tools,
+        ))
         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"))
@@ -54,6 +58,9 @@ class MotatoolRepairTests(unittest.TestCase):
         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["env"]["RUSTC"], "rustc")
+        self.assertEqual(self.install.call_args.kwargs["env"]["CARGO"], "cargo")
+        self.assertEqual(self.install.call_args.kwargs["env"]["RUSTUP_AUTO_INSTALL"], "0")
         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)
@@ -89,6 +96,7 @@ class MotatoolRepairTests(unittest.TestCase):
         self.assertFalse(self.root.exists())
 
     def test_missing_cargo_gives_prerequisite_without_installing_anything(self):
+        self.select_tools.side_effect = ota.OtaError("install a matching toolchain from https://rustup.rs")
         with mock.patch.object(ota.shutil, "which", return_value=None):
             with self.assertRaisesRegex(ota.OtaError, "https://rustup.rs"):
                 self.repair()
@@ -96,6 +104,14 @@ class MotatoolRepairTests(unittest.TestCase):
         self.prompt.assert_not_called()
         self.assertFalse(self.root.exists())
 
+    def test_old_compiler_stops_before_prompt_build_or_cache_creation(self):
+        self.select_tools.side_effect = ota.OtaError("rustc 1.75.0; need 1.86.0 or newer")
+        with self.assertRaisesRegex(ota.OtaError, "rustc 1.75.0"):
+            self.repair()
+        self.prompt.assert_not_called()
+        self.install.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)
@@ -134,6 +150,7 @@ class MotatoolRepairTests(unittest.TestCase):
             self.repair()
         self.prompt.assert_not_called()
         self.install.assert_not_called()
+        self.select_tools.assert_not_called()
         self.check.assert_called_once_with(str(self.binary), self.probe)
         self.assertEqual(self.args.motatool, str(self.binary))
 
@@ -151,6 +168,7 @@ class MotatoolRepairTests(unittest.TestCase):
         with mock.patch.object(ota.shutil, "which", return_value=None):
             self.repair()
         self.install.assert_not_called()
+        self.select_tools.assert_not_called()
         self.assertEqual(self.check.call_count, 2)
         self.assertEqual(self.args.motatool, str(self.binary))