Offer private cryptography repair before bootloader OTA staging

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/<interpreter-id>/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

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)
 

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

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