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