From 3e710594d8cdb505ae1adaad2dbfb9589f88dd8a Mon Sep 17 00:00:00 2001 From: mikecarper Date: Tue, 8 Sep 2026 14:13:38 -0700 Subject: [PATCH 1/2] Check OTA build toolchains and bind sources by public key --- docs/lora_ota_automation.md | 43 +++- tools/lora_ota/lora_ota.py | 247 ++++++++++++++++++- tools/lora_ota/test_lora_ota_bootloader.py | 83 ++++++- tools/lora_ota/test_lora_ota_rust_tools.py | 257 ++++++++++++++++++++ tools/lora_ota/test_lora_ota_tool_repair.py | 18 ++ 5 files changed, 633 insertions(+), 15 deletions(-) create mode 100644 tools/lora_ota/test_lora_ota_rust_tools.py diff --git a/docs/lora_ota_automation.md b/docs/lora_ota_automation.md index 5f8d0c542..1e25b7494 100644 --- a/docs/lora_ota_automation.md +++ b/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,17 @@ 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 `get public.key` from the physical source and matches that full key in the +controller's contacts. It 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 | diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index 7b4f31969..c280c95f3 100755 --- a/tools/lora_ota/lora_ota.py +++ b/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) @@ -6938,6 +6965,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 +7513,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 +7742,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 +7760,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 +7786,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, diff --git a/tools/lora_ota/test_lora_ota_bootloader.py b/tools/lora_ota/test_lora_ota_bootloader.py index 4aed554ab..da8100a31 100644 --- a/tools/lora_ota/test_lora_ota_bootloader.py +++ b/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) diff --git a/tools/lora_ota/test_lora_ota_rust_tools.py b/tools/lora_ota/test_lora_ota_rust_tools.py new file mode 100644 index 000000000..98e0ebf2f --- /dev/null +++ b/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() diff --git a/tools/lora_ota/test_lora_ota_tool_repair.py b/tools/lora_ota/test_lora_ota_tool_repair.py index edc13b0d4..e32d6a8cd 100644 --- a/tools/lora_ota/test_lora_ota_tool_repair.py +++ b/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)) From b7d4ffdb4f9f909886e3578b9dad8ce540ea7d89 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Tue, 8 Sep 2026 14:29:02 -0700 Subject: [PATCH 2/2] Read Full Companion source identity from USB and TCP banners --- docs/lora_ota_automation.md | 19 +- tools/lora_ota/lora_ota.py | 59 ++++- .../lora_ota/test_lora_ota_source_identity.py | 213 ++++++++++++++++++ 3 files changed, 283 insertions(+), 8 deletions(-) create mode 100644 tools/lora_ota/test_lora_ota_source_identity.py diff --git a/docs/lora_ota_automation.md b/docs/lora_ota_automation.md index 1e25b7494..639583824 100644 --- a/docs/lora_ota_automation.md +++ b/docs/lora_ota_automation.md @@ -217,8 +217,23 @@ 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 `get public.key` from the physical source and matches that full key in the -controller's contacts. It does not require the source's current name to match an +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 diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index c280c95f3..880706863 100755 --- a/tools/lora_ota/lora_ota.py +++ b/tools/lora_ota/lora_ota.py @@ -2785,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, @@ -2793,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") @@ -2800,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: @@ -2810,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() @@ -2853,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)", @@ -2899,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 @@ -2932,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 @@ -3033,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: @@ -5001,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( diff --git a/tools/lora_ota/test_lora_ota_source_identity.py b/tools/lora_ota/test_lora_ota_source_identity.py new file mode 100644 index 000000000..f7ac64586 --- /dev/null +++ b/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()