mirror of
https://github.com/mikecarper/MeshCore.git
synced 2026-10-23 08:43:48 +00:00
Merge remote-tracking branch 'origin/keymindCascade' into keymindCascade
This commit is contained in:
mikecarper committed
commit
26aa8118c3
9 files changed
+1864
-49
No files matched your search
+125
-6
@@ -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.
|
||||
@@ -77,6 +77,115 @@ mOTA ownership state as nRF52. The exact `ota folder on` line transfers the USB
|
||||
port from startup ASCII or idle Binary Companion mode before binary mOTA frames
|
||||
begin; neither console mode shares that port until the folder detaches.
|
||||
|
||||
## Bootloader packages and station identifiers
|
||||
|
||||
Both wrappers use the same package-aware runner. It identifies application
|
||||
versus bootloader updates from container metadata, not filenames. A ZIP mixing
|
||||
application inputs and bootloader mOTAs requires `--zip-member`; their versions
|
||||
and target IDs are independent and must not be ranked together.
|
||||
|
||||
Signed format-3 nRF52 bootloader packages can be **staged with `--no-install`**.
|
||||
Use the same controller/source/TempRadio arguments as an application transfer,
|
||||
but provide the bootloader `.mota` (or its ZIP member) and `--no-install`.
|
||||
Without that flag, even `--yes` stops before opening a radio. Bootloader staging
|
||||
requires Python `cryptography` in the interpreter running the wrapper and a
|
||||
current `motatool` supporting format 3. The runner reuses `tools/mota/motalib.py`
|
||||
for strict geometry, embedded identity/CRC, capability, hash and signature
|
||||
validation, and still runs `motatool verify` including any `--public-key` pin.
|
||||
|
||||
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:
|
||||
|
||||
```text
|
||||
Install this bootloader-capable motatool and continue after verification? [y/N]
|
||||
```
|
||||
|
||||
Only `y` or `yes` approves installation. **`--yes` does not approve host software
|
||||
installation.** Noninteractive runs stop and print a copyable manual installation
|
||||
command plus the `--motatool` argument to use afterward.
|
||||
|
||||
The 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
|
||||
[Rust installation](https://rustup.rs/) and stops without installing anything.
|
||||
Source/dependency downloads, build time and disk use are disclosed before
|
||||
confirmation. Cargo progress is displayed, with a one-hour build timeout.
|
||||
The radio admin-password environment variable is not passed to the build.
|
||||
|
||||
The installer does not overwrite the selected/system `motatool` or alter PATH.
|
||||
It may replace only a previous copy in its displayed private install directory.
|
||||
After installation it repeats the bootloader-package capability check and uses
|
||||
the new binary for this run only if that check passes. A later run using the
|
||||
default `motatool` selection can reuse the checked cache without downloading
|
||||
again; switching from an explicit `--motatool` path still asks permission.
|
||||
Refusal, a failed build, or a failed recheck stops before opening any radio.
|
||||
Failed build files may remain in the displayed cache directory for diagnostics.
|
||||
|
||||
The install root is `meshcore-lora-ota/motatool/<revision>` under `%LOCALAPPDATA%`
|
||||
on Windows, `~/Library/Caches` on macOS, or `$XDG_CACHE_HOME` (default
|
||||
`~/.cache`) on Linux. Cargo uses an explicit
|
||||
[`--root`, `--rev`, and `--locked`](https://doc.rust-lang.org/cargo/commands/cargo-install.html)
|
||||
instead of replacing a global installation or following a moving branch.
|
||||
|
||||
Before transfer it queries the destination's `ota bootloader status`, checks its
|
||||
bootloader-specific target/hardware identity, installed ABI/codecs and matching
|
||||
internal/QSPI/SD storage profile. It does not compare an application version or
|
||||
application target ID to a bootloader package. After transfer it checks the
|
||||
destination's exact staged MID and image-hash confirmation, then prints the
|
||||
manual command; it **never sends a bootloader install command**:
|
||||
|
||||
```text
|
||||
ota bootloader status
|
||||
ota bootloader install <MID8> <HASH16>
|
||||
```
|
||||
|
||||
Review and send those commands on the **destination**, not the source. The
|
||||
device independently enforces signer authorization, safe live storage,
|
||||
continuity and upgrade-only policy at installation. `--base`,
|
||||
`--allow-non-upgrade`, and `--prepare-only` remain application-only. See the
|
||||
[bootloader prerequisites and recovery limits](ota_nrf52_bootloader_update.md).
|
||||
|
||||
`TARGET_NODE`, `--relay`, and `--source-contact` accept a contact name, full
|
||||
public key, or unique hexadecimal key prefix. The runner reads the controller's
|
||||
existing contact table and binds each selection to a full key before remote
|
||||
commands, so emoji names need not be typed and later name changes cannot
|
||||
redirect the run. Duplicate names, ambiguous prefixes, or the same radio under
|
||||
different participant aliases stop with an actionable error. Missing contacts
|
||||
must first be imported or discovered; a key alone does not create a contact.
|
||||
|
||||
## Destination requirements
|
||||
|
||||
| Destination | Package installed | One-time prerequisite | Raw ZIP handling |
|
||||
@@ -104,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
|
||||
@@ -559,7 +678,7 @@ Useful controls:
|
||||
If the version gate required RXPS off, it stays off while that topology is
|
||||
preserved; use `target-rxps-settings.json` to restore it only after sending
|
||||
`normalradio`.
|
||||
- `--allow-non-upgrade` deliberately permits the same or an older version.
|
||||
- `--allow-non-upgrade` deliberately permits the same or an older application version.
|
||||
- `--replace-active-download` deliberately discards a different update already
|
||||
downloading or staged on the target. Without it, that update is preserved.
|
||||
- `--source-shares-controller` is for a Full Companion whose USB Binary API is
|
||||
|
||||
@@ -197,6 +197,13 @@ closed instead of selecting smaller or overlapping geometry.
|
||||
|
||||
## Explicit install workflow
|
||||
|
||||
The `tools/lora_ota/lora_ota.sh` and `.ps1` runners can perform the discovery,
|
||||
transfer, verification and radio cleanup for a bootloader mOTA with
|
||||
`--no-install`. They detect format 3 automatically and check the destination's
|
||||
bootloader-specific identity/capabilities. They do not install it: the final
|
||||
MID/hash confirmation below remains an explicit operator action. A mixed
|
||||
application/bootloader ZIP requires `--zip-member`.
|
||||
|
||||
Check the installed identity and capability marker:
|
||||
|
||||
```text
|
||||
|
||||
+682
-40
@@ -17,6 +17,7 @@ import argparse
|
||||
import atexit
|
||||
import getpass
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
@@ -32,11 +33,14 @@ import stat
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import sysconfig
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import dataclass, replace
|
||||
from datetime import datetime, timezone
|
||||
from functools import lru_cache
|
||||
from types import ModuleType
|
||||
from typing import Callable, TypeVar
|
||||
import zipfile
|
||||
|
||||
@@ -138,6 +142,13 @@ RELAY_TIMING_RECOVERY_FILE = "relay-timing-settings.json"
|
||||
TARGET_RXPS_RECOVERY_FILE = "target-rxps-settings.json"
|
||||
SOURCE_RXPS_RECOVERY_FILE = "source-rxps-settings.json"
|
||||
MIN_MESHCLI_VERSION = (1, 6, 0)
|
||||
# Tested format-3 inspect/verify/serve support. Keep automatic repairs pinned;
|
||||
# neither a package nor a moving branch may choose code to install on the host.
|
||||
MOTATOOL_REPAIR_REPOSITORY = "https://github.com/mikecarper/motatool.git"
|
||||
MOTATOOL_REPAIR_REVISION = "8c38369e7d35ad50cf74261869676d52dd24adf7"
|
||||
MOTATOOL_REPAIR_TIMEOUT_SECONDS = 60 * 60
|
||||
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
|
||||
@@ -160,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."""
|
||||
|
||||
@@ -193,6 +208,7 @@ class MotaInfo:
|
||||
hw_id: str
|
||||
base_hash: bytes
|
||||
payload_offset: int
|
||||
bootloader_storage: int | None = None
|
||||
|
||||
@property
|
||||
def is_full(self) -> bool:
|
||||
@@ -200,8 +216,14 @@ class MotaInfo:
|
||||
|
||||
@property
|
||||
def kind(self) -> str:
|
||||
if self.is_bootloader:
|
||||
return "bootloader full"
|
||||
return "full" if self.is_full else "delta"
|
||||
|
||||
@property
|
||||
def is_bootloader(self) -> bool:
|
||||
return bool(self.flags & MOTA_FLAG_BOOTLOADER)
|
||||
|
||||
@property
|
||||
def version(self) -> str:
|
||||
return format_version(self.fw_version)
|
||||
@@ -233,6 +255,9 @@ class TargetInfo:
|
||||
nrf_qspi: bool = False
|
||||
# Firmware predating the live maxblk marker had 1 KiB receive buffers.
|
||||
max_block_size: int = LEGACY_TARGET_MAX_BLOCK_SIZE
|
||||
boot_target_id: int | None = None
|
||||
boot_hw_id: str | None = None
|
||||
boot_storage: int | None = None
|
||||
|
||||
@property
|
||||
def nrf_external(self) -> bool:
|
||||
@@ -810,6 +835,52 @@ def merkle_root(leaves: list[bytes]) -> bytes:
|
||||
return level[0]
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def bootloader_library() -> ModuleType:
|
||||
"""Reuse the repository's strict v3 reference validator, not a second codec."""
|
||||
path = Path(__file__).resolve().parents[1] / "mota" / "motalib.py"
|
||||
name = "_meshcore_lora_ota_motalib"
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
if spec is None or spec.loader is None:
|
||||
raise OtaError(f"bootloader validator is unavailable: {path}")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[name] = module # dataclasses resolves annotations through this registry
|
||||
try:
|
||||
spec.loader.exec_module(module)
|
||||
except (OSError, ImportError) as exc:
|
||||
sys.modules.pop(name, None)
|
||||
raise OtaError(f"cannot load bootloader validator: {exc}") from exc
|
||||
return module
|
||||
|
||||
|
||||
def parse_bootloader_mota(blob: bytes, path: Path | None) -> MotaInfo:
|
||||
library = bootloader_library()
|
||||
try:
|
||||
parsed = library.parse_container(blob)
|
||||
problems = library.verify(parsed)
|
||||
if problems:
|
||||
raise ValueError("; ".join(problems))
|
||||
except ImportError as exc:
|
||||
raise BootloaderCryptoError(
|
||||
"bootloader signature verification requires Python cryptography; "
|
||||
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
|
||||
manifest = parsed.manifest
|
||||
return MotaInfo(
|
||||
path=path, blob=blob, flags=manifest.flags,
|
||||
target_id=manifest.target_id, fw_version=manifest.fw_version,
|
||||
image_size=manifest.image_size, payload_size=manifest.payload_size,
|
||||
block_size=manifest.block_size, merkle_root=manifest.merkle_root,
|
||||
image_hash=manifest.image_hash, codec_id=manifest.codec_id,
|
||||
hw_id=manifest.hw_id.rstrip(b"\0").decode("ascii"),
|
||||
base_hash=manifest.base_hash,
|
||||
payload_offset=len(blob) - len(MOTA_TRAILER) - manifest.payload_size,
|
||||
bootloader_storage=library.bootloader_caps_storage(parsed.payload),
|
||||
)
|
||||
|
||||
|
||||
def parse_mota(blob: bytes, path: Path | None = None) -> MotaInfo:
|
||||
if len(blob) < 8 + MOTA_FIXED_MANIFEST_SIZE + len(MOTA_TRAILER):
|
||||
raise OtaError("mOTA is truncated")
|
||||
@@ -821,17 +892,14 @@ def parse_mota(blob: bytes, path: Path | None = None) -> MotaInfo:
|
||||
f"mOTA size field is {declared_size}, but the file is {len(blob)} bytes"
|
||||
)
|
||||
flags = blob[9]
|
||||
if blob[8] == MOTA_BOOT_FORMAT_VERSION and flags & MOTA_FLAG_BOOTLOADER:
|
||||
raise OtaError(
|
||||
"bootloader mOTA packages require the device's explicit `ota bootloader install` "
|
||||
"workflow; this application-update runner deliberately refuses them"
|
||||
)
|
||||
if blob[10] != 0x12:
|
||||
raise OtaError(f"unsupported mOTA hash algorithm 0x{blob[10]:02x}")
|
||||
if blob[8] == MOTA_BOOT_FORMAT_VERSION:
|
||||
return parse_bootloader_mota(blob, path)
|
||||
if blob[8] != MOTA_FORMAT_VERSION:
|
||||
raise OtaError(f"unsupported mOTA format version {blob[8]}")
|
||||
if flags & MOTA_FLAG_BOOTLOADER or flags & ~MOTA_KNOWN_FLAGS:
|
||||
raise OtaError("invalid flags in v2 application mOTA")
|
||||
if blob[10] != 0x12:
|
||||
raise OtaError(f"unsupported mOTA hash algorithm 0x{blob[10]:02x}")
|
||||
|
||||
target_id, fw_version, image_size, payload_size = struct.unpack_from("<IIII", blob, 11)
|
||||
block_size_log2 = blob[27]
|
||||
@@ -1001,6 +1069,27 @@ def read_zip_member(archive: zipfile.ZipFile, member: zipfile.ZipInfo) -> bytes:
|
||||
|
||||
|
||||
def compatible_mota(info: MotaInfo, target: TargetInfo) -> tuple[bool, str]:
|
||||
if info.is_bootloader:
|
||||
if target.platform != "nrf52" or target.boot_target_id is None:
|
||||
return False, "destination has no verified `ota bootloader status` capability"
|
||||
if info.target_id != target.boot_target_id or info.hw_id != target.boot_hw_id:
|
||||
return False, (
|
||||
f"bootloader target {info.target_id:08X} hw={info.hw_id}, need "
|
||||
f"{target.boot_target_id:08X} hw={target.boot_hw_id}"
|
||||
)
|
||||
if target.bootloader_abi is None or target.bootloader_abi < 3:
|
||||
return False, "bootloader self-update requires installed ABI 3 or newer"
|
||||
if target.bootloader_codecs != 0x5:
|
||||
return False, "bootloader self-update requires FULL|INPLACE codecs 0x5"
|
||||
expected_storage = 0x09 if target.nrf_sd else 0x0E if target.nrf_qspi else 0x0A
|
||||
if (
|
||||
target.boot_storage != expected_storage
|
||||
or info.bootloader_storage != expected_storage
|
||||
):
|
||||
return False, "bootloader package, installed capabilities, and application storage differ"
|
||||
if info.block_size > target.max_block_size:
|
||||
return False, f"bootloader block exceeds destination maxblk:{target.max_block_size}"
|
||||
return True, ""
|
||||
if info.target_id != target.target_id:
|
||||
return False, f"target {info.target_id:08X}, need {target.target_id:08X}"
|
||||
if info.hw_id and target.hw_id and info.hw_id != target.hw_id:
|
||||
@@ -1052,6 +1141,8 @@ def select_mota_from_zip(
|
||||
archive: zipfile.ZipFile,
|
||||
target: TargetInfo,
|
||||
requested_member: str | None,
|
||||
*,
|
||||
package_kind: str = "application",
|
||||
) -> tuple[MotaInfo, str] | None:
|
||||
candidates: list[tuple[MotaInfo, str]] = []
|
||||
rejected: list[str] = []
|
||||
@@ -1062,6 +1153,9 @@ def select_mota_from_zip(
|
||||
continue
|
||||
try:
|
||||
info = parse_mota(read_zip_member(archive, member))
|
||||
if info.is_bootloader != (package_kind == "bootloader"):
|
||||
rejected.append(f"{member.filename}: different update type; select it explicitly")
|
||||
continue
|
||||
good, reason = compatible_mota(info, target)
|
||||
if good:
|
||||
candidates.append((info, member.filename))
|
||||
@@ -1070,7 +1164,9 @@ def select_mota_from_zip(
|
||||
except (OtaError, zipfile.BadZipFile) as exc:
|
||||
rejected.append(f"{member.filename}: {exc}")
|
||||
if not candidates:
|
||||
if requested_member and requested_member.lower().endswith(".mota"):
|
||||
if package_kind == "bootloader" or (
|
||||
requested_member and requested_member.lower().endswith(".mota")
|
||||
):
|
||||
details = "; ".join(rejected) or "member not found"
|
||||
raise OtaError(f"requested ZIP mOTA is unusable: {details}")
|
||||
return None
|
||||
@@ -1144,8 +1240,8 @@ def load_base_image(path: Path, target: TargetInfo) -> EndFInfo:
|
||||
info = parse_mota(
|
||||
read_bounded_file(path, MAX_ARCHIVE_MEMBER_SIZE, "mOTA file"), path
|
||||
)
|
||||
if not info.is_full:
|
||||
raise OtaError("--base mOTA must be a full-image container")
|
||||
if info.is_bootloader or not info.is_full:
|
||||
raise OtaError("--base mOTA must be a full application-image container")
|
||||
identity = parse_endf(info.payload)
|
||||
elif suffix == ".zip":
|
||||
identities: list[tuple[EndFInfo, str]] = []
|
||||
@@ -1156,7 +1252,7 @@ def load_base_image(path: Path, target: TargetInfo) -> EndFInfo:
|
||||
try:
|
||||
if suffix == ".mota":
|
||||
candidate = parse_mota(read_zip_member(archive, member))
|
||||
if not candidate.is_full:
|
||||
if candidate.is_bootloader or not candidate.is_full:
|
||||
continue
|
||||
candidate_identity = parse_endf(candidate.payload)
|
||||
elif suffix in (".bin", ".hex"):
|
||||
@@ -1339,6 +1435,10 @@ def prepare_package(
|
||||
source = args.package.resolve()
|
||||
if not source.is_file():
|
||||
raise OtaError(f"package does not exist: {source}")
|
||||
package_kind = getattr(args, "package_kind", None) or inspect_package_kind(
|
||||
source, args.zip_member
|
||||
)
|
||||
require_package_action(args, package_kind == "bootloader")
|
||||
served_dir = work_dir / "served"
|
||||
served_dir.mkdir(parents=True, exist_ok=False)
|
||||
selected: MotaInfo | None = None
|
||||
@@ -1354,7 +1454,8 @@ def prepare_package(
|
||||
try:
|
||||
with zipfile.ZipFile(source) as archive:
|
||||
mota_member = select_mota_from_zip(
|
||||
archive, target, args.zip_member
|
||||
archive, target, args.zip_member,
|
||||
package_kind=package_kind,
|
||||
)
|
||||
if mota_member is not None:
|
||||
selected, member_name = mota_member
|
||||
@@ -1371,6 +1472,9 @@ def prepare_package(
|
||||
raise OtaError("PACKAGE must be a .mota or .zip file")
|
||||
|
||||
if selected is not None:
|
||||
if selected.is_bootloader != (package_kind == "bootloader"):
|
||||
raise OtaError("package type changed after preflight; restart with the intended input")
|
||||
require_package_action(args, selected.is_bootloader)
|
||||
good, reason = compatible_mota(selected, target)
|
||||
if not good:
|
||||
raise OtaError(f"package is not installable on {target.name}: {reason}")
|
||||
@@ -1437,7 +1541,7 @@ def prepare_package(
|
||||
|
||||
verify_with_motatool(args.motatool, output, args.public_key)
|
||||
expected_body_hash: bytes | None = None
|
||||
if selected.is_full:
|
||||
if selected.is_full and not selected.is_bootloader:
|
||||
try:
|
||||
expected_body_hash = parse_endf(selected.payload).body_hash
|
||||
except OtaError:
|
||||
@@ -1478,6 +1582,14 @@ def reply_matches_command(command_text: str, reply: str) -> bool:
|
||||
needs_temp = lowered.startswith("lora ota needs temp radio")
|
||||
if command == "ota status":
|
||||
return text.startswith("OTA |") or "not included" in lowered or is_unknown
|
||||
if command in ("ota bootloader", "ota bootloader status"):
|
||||
return (
|
||||
text.startswith("BL board=")
|
||||
or lowered.startswith("bootloader update unavailable:")
|
||||
or is_unknown
|
||||
or needs_temp
|
||||
or (is_error and "bootloader" in lowered)
|
||||
)
|
||||
if command == "ota self":
|
||||
return (
|
||||
lowered.startswith("self ")
|
||||
@@ -2048,9 +2160,7 @@ class Controller:
|
||||
contact_keys = [
|
||||
value["public_key"].lower()
|
||||
for value in json_objects(after)
|
||||
if value.get("adv_name") == target
|
||||
and isinstance(value.get("public_key"), str)
|
||||
and re.fullmatch(r"[0-9A-Fa-f]{64}", value["public_key"])
|
||||
if contact_matches_selector(value, target)
|
||||
]
|
||||
if contact_keys != [expected_public_key.lower()]:
|
||||
raise OtaError(
|
||||
@@ -2114,9 +2224,7 @@ class Controller:
|
||||
contact_keys = [
|
||||
value["public_key"].lower()
|
||||
for value in post_objects
|
||||
if value.get("adv_name") == target
|
||||
and isinstance(value.get("public_key"), str)
|
||||
and re.fullmatch(r"[0-9A-Fa-f]{64}", value["public_key"])
|
||||
if contact_matches_selector(value, target)
|
||||
]
|
||||
if len(contact_keys) != 1:
|
||||
if any(
|
||||
@@ -2267,16 +2375,15 @@ class Controller:
|
||||
for item in objects
|
||||
):
|
||||
raise OtaError(f"controller has no contact named {target!r}")
|
||||
target_key = None
|
||||
for item in objects:
|
||||
if (
|
||||
item.get("adv_name") == target
|
||||
and isinstance(item.get("public_key"), str)
|
||||
):
|
||||
target_key = item["public_key"].lower()
|
||||
break
|
||||
if target_key is None:
|
||||
target_keys = {
|
||||
item["public_key"].lower() for item in objects
|
||||
if contact_matches_selector(item, target)
|
||||
}
|
||||
if len(target_keys) > 1:
|
||||
raise OtaError(f"ambiguous contact identity for {target}; use a full public key")
|
||||
if not target_keys:
|
||||
raise TransmissionError(f"meshcli did not return contact identity for {target}")
|
||||
target_key = target_keys.pop()
|
||||
|
||||
messages = [
|
||||
item for item in post_objects
|
||||
@@ -2396,6 +2503,111 @@ def parse_target_max_block_size(status: str, self_status: str) -> int:
|
||||
return value
|
||||
|
||||
|
||||
def contact_matches_selector(contact: dict, selector: str) -> bool:
|
||||
key = contact.get("public_key")
|
||||
if not isinstance(key, str) or not re.fullmatch(r"[0-9A-Fa-f]{64}", key):
|
||||
return False
|
||||
name = contact.get("adv_name")
|
||||
return (
|
||||
isinstance(name, str) and name.casefold() == selector.casefold()
|
||||
) or (
|
||||
re.fullmatch(r"[0-9A-Fa-f]{1,64}", selector) is not None
|
||||
and key.lower().startswith(selector.lower())
|
||||
)
|
||||
|
||||
|
||||
def bind_contact_selectors(controller: Controller, args: argparse.Namespace) -> None:
|
||||
"""Resolve the local contact table once, before any remote command is sent.
|
||||
|
||||
meshcli already accepts keys. Pin unique selectors to complete keys so names
|
||||
with emojis, renames, duplicate names, and prefix collisions cannot redirect
|
||||
maintenance halfway through a run. Reading contacts sends no LoRa packets.
|
||||
"""
|
||||
objects = controller._run(["contacts"], "resolve OTA station identifiers")
|
||||
contacts: dict[str, dict] = {}
|
||||
for obj in objects:
|
||||
for key, value in obj.items():
|
||||
if (
|
||||
isinstance(value, dict)
|
||||
and isinstance(key, str)
|
||||
and re.fullmatch(r"[0-9A-Fa-f]{64}", key)
|
||||
and isinstance(value.get("public_key"), str)
|
||||
and value["public_key"].lower() == key.lower()
|
||||
):
|
||||
contacts[key.lower()] = value
|
||||
|
||||
def resolve(selector: str) -> str:
|
||||
matches = [
|
||||
key for key, value in contacts.items()
|
||||
if contact_matches_selector(value, selector)
|
||||
]
|
||||
if not matches:
|
||||
raise OtaError(
|
||||
f"controller has no contact matching {selector!r}; import or discover it first"
|
||||
)
|
||||
if len(matches) != 1:
|
||||
raise OtaError(
|
||||
f"ambiguous station {selector!r}: {len(matches)} contacts match; "
|
||||
"use a full public key"
|
||||
)
|
||||
key = matches[0]
|
||||
print(f"[contact] {selector} -> {contacts[key].get('adv_name', '?')} [{key}]")
|
||||
return key
|
||||
|
||||
target = resolve(args.target)
|
||||
relays = [(resolve(name), password) for name, password in args.relay_values]
|
||||
source_selector = args.source_contact_value
|
||||
if (
|
||||
not source_selector and not args.source_shares_controller
|
||||
and has_managed_source_cli(args)
|
||||
):
|
||||
source_selector = read_source_name_bounded(args)
|
||||
source = resolve(source_selector) if source_selector else None
|
||||
keys = [target, *(key for key, _password in relays)]
|
||||
if source:
|
||||
keys.append(source)
|
||||
if len(keys) != len(set(keys)):
|
||||
raise OtaError("destination, relays, and source contact must identify different radios")
|
||||
args.target = target
|
||||
args.relay_values = relays
|
||||
args.source_contact_value = source
|
||||
|
||||
|
||||
def query_bootloader_target(
|
||||
controller: Controller, target: TargetInfo,
|
||||
) -> TargetInfo:
|
||||
reply = controller.remote_command(target.name, "ota bootloader status")
|
||||
match = re.fullmatch(
|
||||
r"BL board=([0-9A-Fa-f]{8}) target=([0-9A-Fa-f]{8}) name=(\S+) "
|
||||
r"crc=([0-9A-Fa-f]{8}) abi=(\d+) caps=([0-9A-Fa-f]{2}) "
|
||||
r"\| staged:(?:ready|none) mid=(?:-|[0-9A-Fa-f]{8}) hash=(?:-|[0-9A-Fa-f]{16})",
|
||||
reply.strip(),
|
||||
)
|
||||
if match is None:
|
||||
raise OtaError(
|
||||
f"destination cannot stage a bootloader update: {reply}; "
|
||||
"requires exact-board ABI-3 self-update bootloader and capable application"
|
||||
)
|
||||
board, target_id, name, _crc, abi, caps = match.groups()
|
||||
library = bootloader_library()
|
||||
try:
|
||||
hw_id = (
|
||||
library.bootloader_hw_id(int(board, 16), name)
|
||||
.rstrip(b"\0").decode("ascii")
|
||||
)
|
||||
derived_target = library.bootloader_target_id(int(board, 16), name)
|
||||
except (ValueError, UnicodeError) as exc:
|
||||
raise OtaError(f"invalid destination bootloader identity: {exc}") from exc
|
||||
if derived_target != int(target_id, 16):
|
||||
raise OtaError("destination bootloader board/name does not match its target ID")
|
||||
if int(abi) < 3 or int(abi) != target.bootloader_abi:
|
||||
raise OtaError("destination reports inconsistent or unsupported bootloader ABI")
|
||||
return replace(
|
||||
target, boot_target_id=derived_target, boot_hw_id=hw_id,
|
||||
boot_storage=int(caps, 16),
|
||||
)
|
||||
|
||||
|
||||
def query_target(
|
||||
controller: Controller,
|
||||
args: argparse.Namespace,
|
||||
@@ -2502,7 +2714,7 @@ def query_target(
|
||||
)
|
||||
current_version = format_version(version_value)
|
||||
current_version_source = "ver"
|
||||
return TargetInfo(
|
||||
target = TargetInfo(
|
||||
name=args.target,
|
||||
target_id=target_id,
|
||||
base_hash=base_hash,
|
||||
@@ -2519,6 +2731,9 @@ def query_target(
|
||||
nrf_qspi=nrf_qspi,
|
||||
max_block_size=max_block_size,
|
||||
)
|
||||
if getattr(args, "package_kind", "application") == "bootloader":
|
||||
return query_bootloader_target(controller, target)
|
||||
return target
|
||||
|
||||
|
||||
def parse_temp_radio(value: str) -> tuple[float, float, int, int, int]:
|
||||
@@ -3942,6 +4157,7 @@ def confirm_update(
|
||||
target: TargetInfo,
|
||||
package: MotaInfo,
|
||||
) -> None:
|
||||
require_package_action(args, package.is_bootloader)
|
||||
print("\nValidated update plan:")
|
||||
print(f" destination : {target.name} ({target.target_id:08X}, {target.platform})")
|
||||
print(f" running base: {target.base_hash.hex().upper()}")
|
||||
@@ -3956,6 +4172,9 @@ def confirm_update(
|
||||
print(" bootloader : not required")
|
||||
print(f" update : {package.version} {package.kind} hw={package.hw_id or '?'}")
|
||||
print(f" mOTA id : {package.manifest_id}")
|
||||
if package.is_bootloader:
|
||||
print(f" boot target : {package.target_id:08X}; stage only, explicit install required")
|
||||
print(" boot safety : destination rechecks trust, continuity and upgrade-only policy at manual install")
|
||||
print(f" TempRadio : {args.temp_radio}")
|
||||
saved_rxps = getattr(args, "target_rxps_saved", None)
|
||||
rxps_profile = getattr(args, "target_rxps_profile", None)
|
||||
@@ -3982,7 +4201,10 @@ def confirm_update(
|
||||
current_version = (
|
||||
parse_version(target.current_version) if target.current_version else None
|
||||
)
|
||||
if current_version is not None and current_version >= package.fw_version:
|
||||
if (
|
||||
not package.is_bootloader and current_version is not None
|
||||
and current_version >= package.fw_version
|
||||
):
|
||||
print(
|
||||
f" warning : destination reports {target.current_version}; "
|
||||
f"package is {package.version}"
|
||||
@@ -6189,6 +6411,11 @@ def request_install(
|
||||
package: MotaInfo,
|
||||
) -> bool:
|
||||
"""Request install without ever blindly replaying an uncertain command."""
|
||||
if package.is_bootloader:
|
||||
raise OtaError(
|
||||
"bootloader installation requires explicit "
|
||||
"`ota bootloader install <MID8> <HASH16>`; this runner only stages it"
|
||||
)
|
||||
cycle_started = time.monotonic()
|
||||
retries = 0
|
||||
confirm_ready_to_install(controller, args, package)
|
||||
@@ -6637,7 +6864,7 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
),
|
||||
)
|
||||
parser.add_argument("package", type=Path, metavar="PACKAGE", help=".mota or .zip")
|
||||
parser.add_argument("target", metavar="TARGET_NODE", help="destination contact name")
|
||||
parser.add_argument("target", metavar="TARGET_NODE", help="destination contact name, full public key, or unique key prefix")
|
||||
controller = parser.add_mutually_exclusive_group()
|
||||
controller.add_argument("--controller-serial", metavar="PORT")
|
||||
controller.add_argument("--controller-tcp", metavar="HOST[:PORT]")
|
||||
@@ -6661,14 +6888,14 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
help="destination admin password (prefer the MESHCORE_ADMIN_PASSWORD environment variable)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--relay", action="append", default=[], metavar="NAME[=PASSWORD]",
|
||||
help="optional relay, ordered farthest-to-nearest; repeat as needed",
|
||||
"--relay", action="append", default=[], metavar="NAME_OR_KEY[=PASSWORD]",
|
||||
help="optional relay name/key/unique key prefix, farthest-to-nearest; repeat as needed",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--source-contact",
|
||||
metavar="NAME",
|
||||
metavar="NAME_OR_KEY",
|
||||
help=(
|
||||
"controller contact for a separate OTA source, used for the "
|
||||
"controller contact name/key/unique key prefix for a separate OTA source, used for the "
|
||||
"three-minute on-air proof; defaults to the source's local name "
|
||||
"(no remote-admin password is required)"
|
||||
),
|
||||
@@ -6711,7 +6938,13 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
help=argparse.SUPPRESS,
|
||||
)
|
||||
parser.add_argument("--meshcli", default="meshcli")
|
||||
parser.add_argument("--motatool", default="motatool")
|
||||
parser.add_argument(
|
||||
"--motatool", default="motatool",
|
||||
help=(
|
||||
"host packaging tool; missing/incompatible bootloader support offers "
|
||||
"a separately confirmed private install (--yes does not approve it)"
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--package-build-timeout",
|
||||
type=int,
|
||||
@@ -6761,7 +6994,7 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
"--leave-controller-radio", action="store_true",
|
||||
help="leave the controller on --temp-radio instead of restoring it",
|
||||
)
|
||||
parser.add_argument("--no-install", action="store_true", help="download and verify, but do not install")
|
||||
parser.add_argument("--no-install", action="store_true", help="download and verify, but do not install (required for bootloader mOTA)")
|
||||
parser.add_argument(
|
||||
"--replace-active-download", action="store_true",
|
||||
help="discard a different update already staged on the destination",
|
||||
@@ -7015,6 +7248,407 @@ def require_meshcli_version(command: str) -> tuple[int, int, int]:
|
||||
return version
|
||||
|
||||
|
||||
def require_package_action(args: argparse.Namespace, bootloader: bool) -> None:
|
||||
if not bootloader:
|
||||
return
|
||||
if not getattr(args, "no_install", False):
|
||||
raise OtaError(
|
||||
"detected a bootloader mOTA, not an application update; use --no-install "
|
||||
"to stage it. Installation remains the device's explicit "
|
||||
"`ota bootloader install <MID8> <HASH16>` workflow"
|
||||
)
|
||||
if getattr(args, "prepare_only", False):
|
||||
raise OtaError(
|
||||
"bootloader staging requires live destination capability checks; "
|
||||
"--prepare-only is application-only"
|
||||
)
|
||||
if getattr(args, "base", None) or getattr(args, "allow_non_upgrade", False):
|
||||
raise OtaError(
|
||||
"--base and --allow-non-upgrade are application-only; "
|
||||
"bootloader continuity and upgrade policy cannot be overridden"
|
||||
)
|
||||
|
||||
|
||||
def inspect_package_kind(source: Path, requested_member: str | None) -> str:
|
||||
"""Choose the update family before opening radios; never rank app vs BL versions."""
|
||||
if source.suffix.lower() == ".mota":
|
||||
info = parse_mota(
|
||||
read_bounded_file(source, MAX_ARCHIVE_MEMBER_SIZE, "mOTA file"), source
|
||||
)
|
||||
return "bootloader" if info.is_bootloader else "application"
|
||||
kinds: set[str] = set()
|
||||
found = False
|
||||
try:
|
||||
with zipfile.ZipFile(source) as archive:
|
||||
for member in archive.infolist():
|
||||
if member.is_dir() or (
|
||||
requested_member and member.filename != requested_member
|
||||
):
|
||||
continue
|
||||
if requested_member and found:
|
||||
raise OtaError(f"ZIP member name is duplicated: {requested_member}")
|
||||
found = True
|
||||
suffix = Path(member.filename).suffix.lower()
|
||||
if suffix == ".mota":
|
||||
# Classification is deliberately cheap and conservative. Full
|
||||
# validation (including signatures for BL) follows selection.
|
||||
# A malformed boot declaration must never fall back to an app.
|
||||
with archive.open(member) as stream:
|
||||
header = stream.read(10)
|
||||
is_boot = len(header) == 10 and header[:4] == MOTA_MAGIC and (
|
||||
header[8] == MOTA_BOOT_FORMAT_VERSION
|
||||
or header[9] & MOTA_FLAG_BOOTLOADER
|
||||
)
|
||||
kinds.add("bootloader" if is_boot else "application")
|
||||
elif suffix in (".bin", ".hex"):
|
||||
kinds.add("application")
|
||||
except (OSError, RuntimeError, zipfile.BadZipFile) as exc:
|
||||
raise OtaError(f"cannot inspect ZIP archive {source}: {exc}") from exc
|
||||
if requested_member and not found:
|
||||
raise OtaError(f"requested ZIP member not found: {requested_member}")
|
||||
if len(kinds) > 1:
|
||||
raise OtaError(
|
||||
"ZIP contains both application and bootloader inputs; choose the intended "
|
||||
"update with --zip-member (their versions and target IDs are unrelated)"
|
||||
)
|
||||
return next(iter(kinds), "application")
|
||||
|
||||
|
||||
def report_staged_update(
|
||||
controller: Controller, args: argparse.Namespace, package: MotaInfo,
|
||||
) -> None:
|
||||
if not package.is_bootloader:
|
||||
print(f"{args.target} is ready to install; leaving the verified update staged.")
|
||||
return
|
||||
reply = controller.remote_command(args.target, "ota bootloader status")
|
||||
match = re.search(
|
||||
r"\btarget=([0-9A-Fa-f]{8}) .*\| staged:ready "
|
||||
r"mid=([0-9A-Fa-f]{8}) hash=([0-9A-Fa-f]{16})$", reply.strip(),
|
||||
)
|
||||
if match is None or (
|
||||
int(match[1], 16) != package.target_id
|
||||
or match[2].upper() != package.manifest_id
|
||||
or match[3].lower() != package.image_hash[:8].hex()
|
||||
):
|
||||
raise OtaError(f"destination did not confirm the exact staged bootloader MID/hash: {reply}")
|
||||
print(f"[staged] bootloader {package.version} on {args.target}; NOT installed")
|
||||
print("[manual] Review `ota bootloader status` on the destination, then explicitly send:")
|
||||
print(f" ota bootloader install {package.manifest_id} {package.image_hash[:8].hex().upper()}")
|
||||
print("[manual] Do not use `ota install`. The device still enforces signer trust, continuity and upgrade-only checks.")
|
||||
|
||||
|
||||
def host_repair_cache() -> Path:
|
||||
if os.name == "nt":
|
||||
cache = Path(os.environ.get("LOCALAPPDATA") or Path.home() / "AppData" / "Local")
|
||||
elif sys.platform == "darwin":
|
||||
cache = Path.home() / "Library" / "Caches"
|
||||
else:
|
||||
cache = Path(os.environ.get("XDG_CACHE_HOME") or Path.home() / ".cache")
|
||||
if not cache.is_absolute():
|
||||
raise OtaError("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]:
|
||||
return [
|
||||
cargo, "install", "--git", MOTATOOL_REPAIR_REPOSITORY,
|
||||
"--rev", MOTATOOL_REPAIR_REVISION, "--locked", "--bin", "motatool",
|
||||
"--root", str(root), "--target-dir", str(root / "build"), "--force",
|
||||
]
|
||||
|
||||
|
||||
def display_host_command(command: list[str]) -> str:
|
||||
if os.name == "nt":
|
||||
return "& " + " ".join("'" + word.replace("'", "''") + "'" for word in command)
|
||||
return shlex.join(command)
|
||||
|
||||
|
||||
def check_bootloader_tool(command: str, probe: Path) -> None:
|
||||
run_checked(
|
||||
[command, "inspect", str(probe)],
|
||||
label="check motatool bootloader support", timeout=30,
|
||||
)
|
||||
|
||||
|
||||
def offer_motatool_repair(
|
||||
args: argparse.Namespace, probe: Path, failure: OtaError,
|
||||
) -> None:
|
||||
"""Install only after separate consent; never replace the selected/PATH binary."""
|
||||
root = motatool_repair_root()
|
||||
binary = root / "bin" / ("motatool.exe" if os.name == "nt" else "motatool")
|
||||
print(f"[host] {args.motatool} cannot inspect this format-3 bootloader package: {failure}")
|
||||
cached = False
|
||||
if binary.is_file():
|
||||
try:
|
||||
check_bootloader_tool(str(binary), probe)
|
||||
cached = True
|
||||
except OtaError as exc:
|
||||
print(f"[host] cached motatool also failed its capability check: {exc}")
|
||||
if cached and args.motatool == "motatool":
|
||||
args.motatool = str(binary)
|
||||
print(f"[host] using previously installed, rechecked motatool: {binary}")
|
||||
return
|
||||
|
||||
cargo = shutil.which("cargo")
|
||||
command = motatool_repair_command(root, cargo or "cargo")
|
||||
manual = display_host_command(command)
|
||||
selection = display_host_command(["--motatool", str(binary)])
|
||||
# Display an argument, not a PowerShell call operator for an option.
|
||||
if os.name == "nt":
|
||||
selection = selection.removeprefix("& ")
|
||||
print(f"[repair] source: {MOTATOOL_REPAIR_REPOSITORY}")
|
||||
print(f"[repair] pinned revision: {MOTATOOL_REPAIR_REVISION}")
|
||||
print(f"[repair] private install: {root}")
|
||||
print("[repair] Existing motatool installations and PATH will not be changed.")
|
||||
print(f"[repair] Manual install command: {manual}")
|
||||
print(f"[repair] To select this build explicitly, append: {selection}")
|
||||
if not cached and cargo is None:
|
||||
raise OtaError(
|
||||
"automatic motatool repair requires Rust/Cargo and a native linker; "
|
||||
"install the toolchain from https://rustup.rs and rerun, or select a "
|
||||
"bootloader-capable build with --motatool. No installation was attempted"
|
||||
)
|
||||
if not sys.stdin.isatty():
|
||||
raise OtaError(
|
||||
"motatool repair needs separate interactive approval (--yes does not approve "
|
||||
"software installation); rerun in a terminal or use the printed manual "
|
||||
"command and --motatool. No installation was attempted"
|
||||
)
|
||||
if cached:
|
||||
question = "Use the checked cached motatool for this run instead? [y/N] "
|
||||
else:
|
||||
print(
|
||||
"[repair] This downloads and compiles the pinned source and locked dependencies. "
|
||||
"It can take several minutes and use substantial disk space; "
|
||||
"only a previous copy in the private install folder may be replaced."
|
||||
)
|
||||
question = "Install this bootloader-capable motatool and continue after verification? [y/N] "
|
||||
try:
|
||||
approved = input(question).strip().lower() in ("y", "yes")
|
||||
except EOFError:
|
||||
approved = False
|
||||
if not approved:
|
||||
raise OtaError("motatool repair declined; no installation or radio changes were made")
|
||||
|
||||
if not cached:
|
||||
# No cache directories, downloads or builds before affirmative consent.
|
||||
root.mkdir(parents=True, exist_ok=True)
|
||||
print("[repair] Building motatool (Cargo output follows)...", flush=True)
|
||||
build_env = os.environ.copy()
|
||||
build_env.pop("MESHCORE_ADMIN_PASSWORD", None)
|
||||
try:
|
||||
result = subprocess.run(
|
||||
command, cwd=str(root), stdin=subprocess.DEVNULL,
|
||||
timeout=MOTATOOL_REPAIR_TIMEOUT_SECONDS, check=False,
|
||||
env=build_env,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired) as exc:
|
||||
raise OtaError(
|
||||
f"motatool repair failed: {exc}; existing installations are unchanged. "
|
||||
f"Build files may remain in {root}; fix the toolchain/network and rerun"
|
||||
) from exc
|
||||
if result.returncode != 0:
|
||||
raise OtaError(
|
||||
f"motatool repair failed (Cargo exit {result.returncode}); see the build "
|
||||
f"output above. Existing installations are unchanged; build files remain in {root}"
|
||||
)
|
||||
if not binary.is_file():
|
||||
raise OtaError(f"motatool installer returned success but did not produce {binary}")
|
||||
# Recheck even after Cargo success; do not silently fall back to the old binary.
|
||||
try:
|
||||
check_bootloader_tool(str(binary), probe)
|
||||
except OtaError as exc:
|
||||
raise OtaError(
|
||||
f"repaired motatool still failed bootloader verification: {exc}; "
|
||||
"stopping before radio access"
|
||||
) from exc
|
||||
args.motatool = str(binary)
|
||||
print(f"[repair] bootloader support verified; continuing with {binary}")
|
||||
|
||||
|
||||
def require_bootloader_tool_support(args: argparse.Namespace) -> None:
|
||||
"""Probe actual format support: old and new motatool builds can share a version."""
|
||||
if args.package.suffix.lower() == ".mota":
|
||||
probe = args.package
|
||||
temporary_blob = None
|
||||
else:
|
||||
temporary_blob = None
|
||||
failures: list[str] = []
|
||||
try:
|
||||
with zipfile.ZipFile(args.package) as archive:
|
||||
for member in archive.infolist():
|
||||
if not member.filename.lower().endswith(".mota") or (
|
||||
args.zip_member and member.filename != args.zip_member
|
||||
):
|
||||
continue
|
||||
try:
|
||||
info = parse_mota(read_zip_member(archive, member))
|
||||
if info.is_bootloader:
|
||||
temporary_blob = info.blob
|
||||
break
|
||||
except 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:
|
||||
raise OtaError(f"invalid ZIP archive: {args.package}") from exc
|
||||
if temporary_blob is None:
|
||||
raise OtaError("ZIP contains no valid bootloader mOTA: " + "; ".join(failures))
|
||||
|
||||
# This is a local parser probe, never a serve/pull operation. Do not require
|
||||
# every hardware variant in an archive to share the operator's signer pin;
|
||||
# the selected package is independently verified with that pin later.
|
||||
with tempfile.TemporaryDirectory(prefix="meshcore-boot-probe-") as directory:
|
||||
if temporary_blob is not None:
|
||||
probe = Path(directory) / "bootloader.mota"
|
||||
probe.write_bytes(temporary_blob)
|
||||
try:
|
||||
check_bootloader_tool(args.motatool, probe)
|
||||
except OtaError as exc:
|
||||
offer_motatool_repair(args, probe, exc)
|
||||
|
||||
|
||||
def preflight_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()}")
|
||||
@@ -7027,7 +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()}")
|
||||
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)
|
||||
|
||||
@@ -7252,6 +7893,7 @@ def main(
|
||||
)
|
||||
if controller is None:
|
||||
controller = Controller(args, password)
|
||||
bind_contact_selectors(controller, args)
|
||||
verify_shared_source_identity(controller, args)
|
||||
# These gates are intentionally before query_target() or any other
|
||||
# on-air remote operation. First prove that advancing the managed
|
||||
@@ -7519,7 +8161,7 @@ def main(
|
||||
monitor_download(controller, args, package, seeder)
|
||||
|
||||
if args.no_install:
|
||||
print(f"{args.target} is ready to install; leaving the verified update staged.")
|
||||
report_staged_update(controller, args, package)
|
||||
restore_relay_timings(controller, relay_timing_settings)
|
||||
relay_timing_settings.clear()
|
||||
if args.leave_controller_radio:
|
||||
|
||||
@@ -143,11 +143,11 @@ class FormatTests(unittest.TestCase):
|
||||
with self.assertRaisesRegex(ota.OtaError, "block hashes"):
|
||||
ota.parse_mota(bytes(blob))
|
||||
|
||||
def test_bootloader_packages_are_explicitly_refused(self) -> None:
|
||||
def test_relabelling_application_does_not_make_a_bootloader(self) -> None:
|
||||
blob = bytearray(mota_blob(firmware(b"B" * 5000, VERSION_NEW)))
|
||||
blob[8] = ota.MOTA_BOOT_FORMAT_VERSION
|
||||
blob[9] |= ota.MOTA_FLAG_SIGNED | ota.MOTA_FLAG_BOOTLOADER
|
||||
with self.assertRaisesRegex(ota.OtaError, "bootloader mOTA packages"):
|
||||
with self.assertRaisesRegex(ota.OtaError, "invalid bootloader mOTA"):
|
||||
ota.parse_mota(bytes(blob))
|
||||
|
||||
blob[8] = ota.MOTA_FORMAT_VERSION
|
||||
@@ -836,6 +836,13 @@ class DebugTests(unittest.TestCase):
|
||||
|
||||
|
||||
class SourceCliTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
# These lifecycle tests stub the controller and source transports.
|
||||
# Station resolution itself is covered by StationSelectionTests.
|
||||
patcher = mock.patch.object(ota, "bind_contact_selectors")
|
||||
patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
|
||||
@staticmethod
|
||||
def source_args() -> argparse.Namespace:
|
||||
return argparse.Namespace(
|
||||
@@ -4426,6 +4433,7 @@ class TempRadioPreflightTests(unittest.TestCase):
|
||||
]
|
||||
with (
|
||||
mock.patch.object(ota, "preflight_inputs"),
|
||||
mock.patch.object(ota, "bind_contact_selectors"),
|
||||
mock.patch.object(ota, "preflight_source_cli"),
|
||||
mock.patch.object(
|
||||
ota,
|
||||
|
||||
@@ -0,0 +1,457 @@
|
||||
"""Offline bootloader staging and station-selection regression tests. No radios."""
|
||||
|
||||
import argparse
|
||||
import contextlib
|
||||
from dataclasses import replace
|
||||
import io
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import struct
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest import mock
|
||||
import warnings
|
||||
import zipfile
|
||||
import zlib
|
||||
|
||||
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
||||
|
||||
import lora_ota as ota
|
||||
from test_lora_ota import firmware, mota_blob, prepare_args, target, VERSION_NEW
|
||||
|
||||
|
||||
def boot_blob(*, storage=0x0A, version=0x0117010D):
|
||||
"""Build a signed test-only candidate using the canonical reference builder."""
|
||||
ml = ota.bootloader_library()
|
||||
board, name, app_base = 0x239A0029, "3401_DFU", ml.NRF52_APP_BASE_S140_V6
|
||||
if storage == 0x0E:
|
||||
board, name, app_base = 0x28860044, "XIAO-SENSE-DFU", ml.NRF52_APP_BASE_S140_V7
|
||||
name = ml.XIAO_BOOT_DEVICE_NAME.rstrip(b"\0").decode("ascii")
|
||||
elif storage == 0x09:
|
||||
board, name = 0x239A0071, "TOWER_V2_OTA"
|
||||
image = bytearray(b"\xff" * ml.XIAO_BOOT_IMAGE_SIZE)
|
||||
struct.pack_into("<II", image, 0, 0x20040000, ml.XIAO_BOOT_IMAGE_START + 0x101)
|
||||
struct.pack_into("<8sHHB3x", image, 0x80, ml.XIAO_BOOT_CAPS_MAGIC, 3, 5, storage)
|
||||
offset = ml.BOOT_CANDIDATE_MANIFEST_OFFSET
|
||||
struct.pack_into(
|
||||
"<8sHHIII16sI", image, offset, ml.XIAO_BOOT_MANIFEST_MAGIC,
|
||||
ml.XIAO_BOOT_MANIFEST_VERSION, ml.XIAO_BOOT_MANIFEST_SIZE,
|
||||
ml.XIAO_BOOT_IMAGE_START, len(image), board, name.encode("ascii"), 0,
|
||||
)
|
||||
struct.pack_into(
|
||||
"<8sHHIHHIHHI", image, offset + ml.XIAO_BOOT_MANIFEST_SIZE,
|
||||
ml.BOOT_CONTINUITY_MAGIC, ml.BOOT_CONTINUITY_VERSION, ml.BOOT_CONTINUITY_SIZE,
|
||||
version, ml.BOOT_CONTINUITY_FAMILY_S140,
|
||||
0x0123 if app_base == ml.NRF52_APP_BASE_S140_V7 else 0x00B6,
|
||||
app_base, ml.BOOT_CONTINUITY_LAYOUT_ABI, 0, 0,
|
||||
)
|
||||
struct.pack_into("<I", image, offset + 40, zlib.crc32(image) & 0xFFFFFFFF)
|
||||
image = bytes(image)
|
||||
manifest = ml.build_manifest(
|
||||
target_id=ml.bootloader_target_id(board, name), fw_version=version,
|
||||
image_size=len(image), payload=image, block_size=1024,
|
||||
image_hash=ml.mh32(image), codec_id=ml.CODEC_FULL, is_full=True,
|
||||
sign_priv=Ed25519PrivateKey.from_private_bytes(b"\x01" * 32), bootloader=True,
|
||||
)
|
||||
return ml.build_container(manifest, image)
|
||||
|
||||
|
||||
def boot_target(package):
|
||||
return replace(
|
||||
target(platform="nrf52", boot_codecs=5), bootloader_abi=3,
|
||||
boot_target_id=package.target_id, boot_hw_id=package.hw_id,
|
||||
boot_storage=package.bootloader_storage,
|
||||
nrf_sd=package.bootloader_storage == 9,
|
||||
nrf_qspi=package.bootloader_storage == 14,
|
||||
)
|
||||
|
||||
|
||||
class BootloaderStagingTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.blob = boot_blob()
|
||||
cls.package = ota.parse_mota(cls.blob)
|
||||
|
||||
def test_signed_full_bootloader_is_identified(self):
|
||||
package = self.package
|
||||
self.assertTrue(package.is_bootloader)
|
||||
self.assertEqual(package.kind, "bootloader full")
|
||||
self.assertEqual(package.target_id, 0x23818A80)
|
||||
self.assertEqual(package.block_size, 1024)
|
||||
self.assertEqual(package.payload_size, 40960)
|
||||
|
||||
def test_all_supported_storage_profiles(self):
|
||||
for storage in (10, 14, 9):
|
||||
with self.subTest(storage=storage):
|
||||
package = ota.parse_mota(boot_blob(storage=storage))
|
||||
self.assertEqual(ota.compatible_mota(package, boot_target(package)), (True, ""))
|
||||
|
||||
def test_malformed_bootloaders_fail_closed(self):
|
||||
for offset in (9, 10, 11, 27, 28, 105, 137, 201, 205, 600):
|
||||
with self.subTest(offset=offset):
|
||||
bad = bytearray(self.blob)
|
||||
bad[offset] ^= 1
|
||||
with self.assertRaises(ota.OtaError):
|
||||
ota.parse_mota(bytes(bad))
|
||||
|
||||
def test_bootloader_is_not_an_application_full_exception(self):
|
||||
application = ota.parse_mota(mota_blob(firmware(b"test" * 1000, VERSION_NEW)))
|
||||
self.assertFalse(ota.compatible_mota(application, target(platform="nrf52"))[0])
|
||||
|
||||
def test_bootloader_compatibility_requires_exact_live_contract(self):
|
||||
live = boot_target(self.package)
|
||||
for change in (
|
||||
dict(platform="esp32"), dict(boot_target_id=None),
|
||||
dict(boot_target_id=self.package.target_id ^ 1), dict(boot_hw_id="wrong"),
|
||||
dict(bootloader_abi=2), dict(bootloader_codecs=4),
|
||||
dict(boot_storage=14), dict(nrf_qspi=True), dict(max_block_size=512),
|
||||
):
|
||||
with self.subTest(change=change):
|
||||
self.assertFalse(ota.compatible_mota(self.package, replace(live, **change))[0])
|
||||
|
||||
def test_stage_only_is_required_even_with_yes(self):
|
||||
args = argparse.Namespace(no_install=False, yes=True)
|
||||
with self.assertRaisesRegex(ota.OtaError, "use --no-install"):
|
||||
ota.require_package_action(args, True)
|
||||
args.no_install = True
|
||||
ota.require_package_action(args, True)
|
||||
for key, value in (("prepare_only", True), ("allow_non_upgrade", True), ("base", Path("base.bin"))):
|
||||
with self.subTest(option=key):
|
||||
with self.assertRaises(ota.OtaError):
|
||||
ota.require_package_action(argparse.Namespace(no_install=True, **{key: value}), True)
|
||||
|
||||
def test_default_run_rejects_boot_before_contacting_radios(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / "not-an-app.mota"
|
||||
path.write_bytes(self.blob)
|
||||
with (
|
||||
mock.patch.object(ota, "Controller") as controller,
|
||||
mock.patch.object(ota, "source_cli_command") as source,
|
||||
mock.patch.object(ota, "require_command") as require,
|
||||
contextlib.redirect_stdout(io.StringIO()),
|
||||
contextlib.redirect_stderr(io.StringIO()) as error,
|
||||
):
|
||||
result = ota.main([
|
||||
str(path), "remote", "--controller-serial", "controller",
|
||||
"--source-serial", "source", "--yes",
|
||||
])
|
||||
self.assertEqual(result, 2)
|
||||
self.assertIn("use --no-install", error.getvalue())
|
||||
controller.assert_not_called()
|
||||
source.assert_not_called()
|
||||
require.assert_not_called()
|
||||
|
||||
def test_prepare_boot_stage_uses_motatool_verification(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
source = root / "boot.mota"
|
||||
source.write_bytes(self.blob)
|
||||
args = prepare_args(source, "motatool")
|
||||
args.no_install = True
|
||||
with mock.patch.object(ota, "verify_with_motatool") as verify:
|
||||
path, package, body_hash = ota.prepare_package(args, boot_target(self.package), root)
|
||||
self.assertTrue(package.is_bootloader)
|
||||
self.assertIsNone(body_hash)
|
||||
self.assertEqual(path.read_bytes(), self.blob)
|
||||
verify.assert_called_once_with("motatool", path, None)
|
||||
|
||||
def test_mixed_archive_requires_explicit_choice(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / "mixed.zip"
|
||||
with zipfile.ZipFile(path, "w") as archive:
|
||||
archive.writestr("renamed.mota", self.blob)
|
||||
archive.writestr("firmware.bin", firmware(b"test" * 1000, VERSION_NEW))
|
||||
with self.assertRaisesRegex(ota.OtaError, "both application and bootloader"):
|
||||
ota.inspect_package_kind(path, None)
|
||||
self.assertEqual(ota.inspect_package_kind(path, "renamed.mota"), "bootloader")
|
||||
self.assertEqual(ota.inspect_package_kind(path, "firmware.bin"), "application")
|
||||
with self.assertRaisesRegex(ota.OtaError, "not found"):
|
||||
ota.inspect_package_kind(path, "missing.mota")
|
||||
|
||||
def test_boot_archive_selects_matching_newest_and_never_falls_back(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / "boot.zip"
|
||||
with zipfile.ZipFile(path, "w") as archive:
|
||||
archive.writestr("old.mota", self.blob)
|
||||
archive.writestr("new.mota", boot_blob(version=0x0117010E))
|
||||
archive.writestr("xiao.mota", boot_blob(storage=14))
|
||||
with zipfile.ZipFile(path) as archive:
|
||||
selected = ota.select_mota_from_zip(archive, boot_target(self.package), None, package_kind="bootloader")
|
||||
self.assertEqual(selected[1], "new.mota")
|
||||
with self.assertRaisesRegex(ota.OtaError, "unusable"):
|
||||
ota.select_mota_from_zip(archive, target(platform="esp32"), None, package_kind="bootloader")
|
||||
|
||||
def test_duplicate_zip_member_cannot_be_selected_by_name(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / "duplicate.zip"
|
||||
with warnings.catch_warnings(), zipfile.ZipFile(path, "w") as archive:
|
||||
warnings.simplefilter("ignore", UserWarning)
|
||||
archive.writestr("boot.mota", self.blob)
|
||||
archive.writestr("boot.mota", self.blob)
|
||||
with self.assertRaisesRegex(ota.OtaError, "duplicated"):
|
||||
ota.inspect_package_kind(path, "boot.mota")
|
||||
|
||||
def test_query_uses_boot_identity_not_application_id(self):
|
||||
controller = mock.Mock()
|
||||
controller.remote_command.return_value = (
|
||||
"BL board=239A0029 target=23818A80 name=3401_DFU "
|
||||
"crc=12345678 abi=3 caps=0A | staged:none mid=- hash=-"
|
||||
)
|
||||
live = replace(target(platform="nrf52", boot_codecs=5), bootloader_abi=3)
|
||||
result = ota.query_bootloader_target(controller, live)
|
||||
self.assertEqual(result.target_id, live.target_id)
|
||||
self.assertEqual(result.boot_target_id, self.package.target_id)
|
||||
self.assertEqual(result.boot_hw_id, self.package.hw_id)
|
||||
for reply in (
|
||||
"ERR this build cannot update its bootloader over LoRa",
|
||||
controller.remote_command.return_value.replace("23818A80", "00000000"),
|
||||
controller.remote_command.return_value.replace("abi=3", "abi=2"),
|
||||
):
|
||||
controller.remote_command.return_value = reply
|
||||
with self.assertRaises(ota.OtaError):
|
||||
ota.query_bootloader_target(controller, live)
|
||||
|
||||
def test_staging_prints_confirmation_but_never_installs(self):
|
||||
package = self.package
|
||||
args = argparse.Namespace(target="remote")
|
||||
controller = mock.Mock()
|
||||
reply = (
|
||||
"BL board=239A0029 target=23818A80 name=3401_DFU crc=12345678 abi=3 caps=0A "
|
||||
f"| staged:ready mid={package.manifest_id} hash={package.image_hash[:8].hex()}"
|
||||
)
|
||||
controller.remote_command.return_value = reply
|
||||
with contextlib.redirect_stdout(io.StringIO()) as output:
|
||||
ota.report_staged_update(controller, args, package)
|
||||
controller.remote_command.assert_called_once_with("remote", "ota bootloader status")
|
||||
self.assertIn("NOT installed", output.getvalue())
|
||||
self.assertIn(f"ota bootloader install {package.manifest_id}", output.getvalue())
|
||||
for bad in (reply.replace("staged:ready", "staged:none"), reply.replace(package.manifest_id, "00000000"), reply.replace(package.image_hash[:8].hex(), "0000000000000000")):
|
||||
controller.remote_command.return_value = bad
|
||||
with self.assertRaisesRegex(ota.OtaError, "exact staged"):
|
||||
ota.report_staged_update(controller, args, package)
|
||||
controller.reset_mock()
|
||||
with self.assertRaisesRegex(ota.OtaError, "explicit"):
|
||||
ota.request_install(controller, args, package)
|
||||
controller.remote_command.assert_not_called()
|
||||
|
||||
def test_target_query_only_adds_boot_probe_for_boot_packages(self):
|
||||
replies = {
|
||||
"ota status": "OTA | no download | target:1234ABCD hw=TestBoard",
|
||||
"get bootloader.ver": "> 0.11.0-OTAFIX2.4.6",
|
||||
"ota self": "self body=100 image=156 base_hash=0011223344556677 | bootloader: abi=3 codecs=0x5",
|
||||
"ota stats": "OTA | fw v1.17.1.5",
|
||||
"ota bootloader status": (
|
||||
"BL board=239A0029 target=23818A80 name=3401_DFU "
|
||||
"crc=12345678 abi=3 caps=0A | staged:none mid=- hash=-"
|
||||
),
|
||||
}
|
||||
for kind in ("application", "bootloader"):
|
||||
controller = mock.Mock()
|
||||
controller.remote_command.side_effect = lambda _target, command, **_kwargs: replies[command]
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
result = ota.query_target(controller, argparse.Namespace(target="remote", package_kind=kind))
|
||||
if kind == "bootloader":
|
||||
self.assertEqual(result.boot_target_id, self.package.target_id)
|
||||
self.assertEqual(ota.compatible_mota(self.package, result), (True, ""))
|
||||
else:
|
||||
self.assertIsNone(result.boot_target_id)
|
||||
self.assertNotIn(mock.call("remote", "ota bootloader status"), controller.remote_command.call_args_list)
|
||||
|
||||
def test_application_version_is_not_compared_to_bootloader_version(self):
|
||||
args = ota.build_parser().parse_args(["boot.mota", "remote", "--no-install", "--yes"])
|
||||
live = replace(boot_target(self.package), current_version="v99.0.0")
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
ota.confirm_update(args, live, self.package)
|
||||
|
||||
def test_boot_status_reply_filter(self):
|
||||
for reply in (
|
||||
"BL board=239A0029 target=23818A80", "Bootloader update unavailable: bad CRC",
|
||||
"ERR this build cannot update its bootloader over LoRa",
|
||||
):
|
||||
self.assertTrue(ota.reply_matches_command("ota bootloader status", reply))
|
||||
self.assertFalse(ota.reply_matches_command("ota bootloader status", "OTA | no download | target:1234ABCD"))
|
||||
|
||||
@unittest.skipUnless(os.environ.get("MOTATOOL_TEST_BIN") or shutil.which("motatool"), "motatool is not installed")
|
||||
def test_real_motatool_verifies_bootloader_and_rejects_wrong_signer_pin(self):
|
||||
tool = os.environ.get("MOTATOOL_TEST_BIN") or "motatool"
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
package = root / "boot.mota"
|
||||
package.write_bytes(self.blob)
|
||||
pin = root / "signer.pub"
|
||||
pin.write_text(self.blob[105:137].hex(), encoding="ascii")
|
||||
ota.require_bootloader_tool_support(argparse.Namespace(package=package, motatool=tool))
|
||||
archive_path = root / "boot.zip"
|
||||
with zipfile.ZipFile(archive_path, "w") as archive:
|
||||
archive.writestr("boot.mota", self.blob)
|
||||
ota.require_bootloader_tool_support(argparse.Namespace(
|
||||
package=archive_path, zip_member="boot.mota", motatool=tool,
|
||||
))
|
||||
ota.verify_with_motatool(tool, package, pin)
|
||||
pin.write_text("02" * 32, encoding="ascii")
|
||||
with self.assertRaises(ota.OtaError):
|
||||
ota.verify_with_motatool(tool, package, pin)
|
||||
|
||||
def test_old_motatool_stops_before_any_radio_access(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / "boot.mota"
|
||||
path.write_bytes(self.blob)
|
||||
args = ota.build_parser().parse_args([str(path), "remote", "--no-install"])
|
||||
with (
|
||||
mock.patch.object(ota, "require_command"),
|
||||
mock.patch.object(ota, "require_meshcli_version") as meshcli,
|
||||
mock.patch.object(ota, "run_checked", side_effect=ota.OtaError("unsupported format_ver 3")),
|
||||
mock.patch.object(ota, "motatool_repair_root", return_value=Path(directory) / "cache"),
|
||||
mock.patch.object(ota.shutil, "which", return_value="cargo"),
|
||||
mock.patch.object(ota.sys.stdin, "isatty", return_value=False),
|
||||
self.assertRaisesRegex(ota.OtaError, "separate interactive approval"),
|
||||
):
|
||||
ota.preflight_inputs(args)
|
||||
meshcli.assert_not_called()
|
||||
|
||||
def test_simulated_main_stages_bootloader_and_restores_radios(self):
|
||||
package = self.package
|
||||
destination_key, source_key = "a1" * 32, "b2" * 32
|
||||
controller = mock.Mock()
|
||||
controller._run.return_value = [{
|
||||
destination_key: {"adv_name": "remote", "public_key": destination_key},
|
||||
source_key: {"adv_name": "source", "public_key": source_key},
|
||||
}]
|
||||
normal = ota.RadioSettings(910.525, 62.5, 7, 5, False)
|
||||
controller.get_radio.return_value = normal
|
||||
|
||||
def remote_command(station, command, **_kwargs):
|
||||
self.assertEqual(station, destination_key)
|
||||
replies = {
|
||||
"ota status": "OTA | no download | target:1234ABCD",
|
||||
"ota ls": "Updates 1",
|
||||
f"ota pull {package.manifest_id} flash": f"OK pulling mid={package.manifest_id} -> flash",
|
||||
"ota bootloader status": (
|
||||
"BL board=239A0029 target=23818A80 name=3401_DFU crc=12345678 abi=3 caps=0A "
|
||||
f"| staged:ready mid={package.manifest_id} hash={package.image_hash[:8].hex()}"
|
||||
),
|
||||
}
|
||||
if command not in replies:
|
||||
self.fail(f"unexpected remote command: {command}")
|
||||
return replies[command]
|
||||
|
||||
controller.remote_command.side_effect = remote_command
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
source = root / "boot.mota"
|
||||
source.write_bytes(self.blob)
|
||||
seeder = mock.Mock()
|
||||
with contextlib.ExitStack() as stack:
|
||||
for name in (
|
||||
"require_command", "require_meshcli_version", "require_bootloader_tool_support", "preflight_source_cli",
|
||||
"ensure_source_clock_gate_safe", "ensure_controller_clock_safe",
|
||||
"verify_with_motatool", "arm_target_temp_radio",
|
||||
"switch_controller_to_temp_radio", "source_cli_command",
|
||||
):
|
||||
stack.enter_context(mock.patch.object(ota, name))
|
||||
stack.enter_context(mock.patch.object(ota, "read_source_name_bounded", return_value="source"))
|
||||
stack.enter_context(mock.patch.object(ota, "read_source_rxps", return_value=ota.RxpsSettings(False, 0, 0)))
|
||||
stack.enter_context(mock.patch.object(ota, "read_remote_rxps", return_value=ota.RxpsSettings(False, 0, 0)))
|
||||
stack.enter_context(mock.patch.object(
|
||||
ota, "query_target", return_value=replace(boot_target(package), name=destination_key),
|
||||
))
|
||||
stack.enter_context(mock.patch.object(ota, "read_lora_ota_participant_versions", return_value={}))
|
||||
rehearsal = stack.enter_context(mock.patch.object(ota, "run_temp_radio_preflight"))
|
||||
stack.enter_context(mock.patch.object(ota, "arm_source_temp_radio_once", return_value=True))
|
||||
stack.enter_context(mock.patch.object(ota, "SeederProcess", return_value=seeder))
|
||||
stack.enter_context(mock.patch.object(ota, "monitor_download"))
|
||||
cleanup = stack.enter_context(mock.patch.object(ota, "shorten_target_temp_window"))
|
||||
stack.enter_context(mock.patch.object(ota, "shorten_source_temp_window", return_value=True))
|
||||
install = stack.enter_context(mock.patch.object(ota, "request_install"))
|
||||
stack.enter_context(mock.patch.object(ota.time, "sleep"))
|
||||
output = stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
|
||||
error = stack.enter_context(contextlib.redirect_stderr(io.StringIO()))
|
||||
result = ota.main([
|
||||
str(source), destination_key[:12], "--controller-serial", "controller",
|
||||
"--source-serial", "source", "--password", "test-only",
|
||||
"--no-install", "--yes", "--work-dir", str(root / "work"),
|
||||
], controller_override=controller)
|
||||
self.assertEqual(result, 0, error.getvalue())
|
||||
self.assertIn("detected bootloader update", output.getvalue())
|
||||
self.assertIn("NOT installed", output.getvalue())
|
||||
rehearsal.assert_called_once()
|
||||
seeder.start.assert_called_once()
|
||||
cleanup.assert_called_once()
|
||||
controller.set_radio.assert_called_once_with(normal, "restore controller radio after staging")
|
||||
install.assert_not_called()
|
||||
self.assertIn(
|
||||
mock.call(destination_key, f"ota pull {package.manifest_id} flash", retry=False),
|
||||
controller.remote_command.call_args_list,
|
||||
)
|
||||
|
||||
|
||||
class StationSelectionTests(unittest.TestCase):
|
||||
key = "3ee21f453f8f" + "ab" * 26
|
||||
name = "Ashport \U0001f4e1"
|
||||
|
||||
def args(self, selector):
|
||||
return argparse.Namespace(
|
||||
target=selector, relay_values=[], source_contact_value=None,
|
||||
source_shares_controller=True,
|
||||
)
|
||||
|
||||
def controller(self):
|
||||
controller = mock.Mock()
|
||||
controller._run.return_value = [{self.key: {"public_key": self.key, "adv_name": self.name}}]
|
||||
return controller
|
||||
|
||||
def test_names_full_keys_and_prefixes_bind_to_one_full_key(self):
|
||||
for selector in (self.name, self.name.upper(), self.key, self.key.upper(), self.key[:12]):
|
||||
with self.subTest(selector=selector):
|
||||
args = self.args(selector)
|
||||
controller = self.controller()
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
ota.bind_contact_selectors(controller, args)
|
||||
self.assertEqual(args.target, self.key)
|
||||
controller._run.assert_called_once_with(["contacts"], "resolve OTA station identifiers")
|
||||
controller.remote_command.assert_not_called()
|
||||
|
||||
def test_missing_and_ambiguous_identifiers_are_refused(self):
|
||||
controller = self.controller()
|
||||
with self.assertRaisesRegex(ota.OtaError, "no contact"):
|
||||
ota.bind_contact_selectors(controller, self.args("missing"))
|
||||
second = self.key[:-2] + "cd"
|
||||
controller._run.return_value[0][second] = {"public_key": second, "adv_name": self.name}
|
||||
for selector in (self.name, self.key[:12]):
|
||||
with self.assertRaisesRegex(ota.OtaError, "ambiguous"):
|
||||
ota.bind_contact_selectors(controller, self.args(selector))
|
||||
|
||||
def test_same_radio_cannot_be_target_and_relay_under_aliases(self):
|
||||
args = self.args(self.name)
|
||||
args.relay_values = [(self.key[:12], "unused")]
|
||||
with contextlib.redirect_stdout(io.StringIO()), self.assertRaisesRegex(ota.OtaError, "different radios"):
|
||||
ota.bind_contact_selectors(self.controller(), args)
|
||||
|
||||
def test_remote_reply_accepts_key_selector_with_emoji_name(self):
|
||||
for selector in (self.name, self.key, self.key[:12]):
|
||||
controller = object.__new__(ota.Controller)
|
||||
controller.reply_timeout = 5
|
||||
controller._authenticated_targets = {selector}
|
||||
controller._run_marked = mock.Mock(return_value=(
|
||||
[{"adv_name": self.name, "public_key": self.key}],
|
||||
[{"txt_type": 1, "text": "OTA | no download | target:1234ABCD", "pubkey_prefix": self.key[:12]}],
|
||||
))
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
self.assertIn("OTA", controller._remote_command_once(selector, "ota status", "unused"))
|
||||
|
||||
def test_clock_and_ack_accept_key_selector(self):
|
||||
controller = object.__new__(ota.Controller)
|
||||
controller._execute = mock.Mock(side_effect=lambda commands, _label: argparse.Namespace(stdout=(
|
||||
commands[1] + '\n{"adv_name": "renamed", "public_key": "' + self.key + '"}\n1800000000\n'
|
||||
)))
|
||||
self.assertEqual(controller.get_contact_clock(self.key, self.key), 1800000000)
|
||||
controller._run_marked = mock.Mock(return_value=([], [
|
||||
{"adv_name": "renamed", "public_key": self.key},
|
||||
{"expected_ack": "12345678"}, {"code": "12345678"},
|
||||
]))
|
||||
controller.prove_contact_ack(self.key, self.key, "proof")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
@@ -0,0 +1,206 @@
|
||||
"""Automatic host-tool repair tests. All installer execution is mocked."""
|
||||
|
||||
import argparse
|
||||
import contextlib
|
||||
import io
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
import lora_ota as ota
|
||||
from test_lora_ota_bootloader import boot_blob
|
||||
|
||||
|
||||
class MotatoolRepairTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
directory = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(directory.cleanup)
|
||||
self.folder = Path(directory.name)
|
||||
self.root = self.folder / "private-install"
|
||||
self.binary = self.root / "bin" / ("motatool.exe" if os.name == "nt" else "motatool")
|
||||
self.probe = self.folder / "boot.mota"
|
||||
self.probe.write_bytes(boot_blob())
|
||||
self.args = argparse.Namespace(motatool="old-motatool", yes=True, package=self.probe)
|
||||
self.failure = ota.OtaError("unsupported format_ver 3")
|
||||
stack = contextlib.ExitStack()
|
||||
self.addCleanup(stack.close)
|
||||
stack.enter_context(mock.patch.object(ota, "motatool_repair_root", return_value=self.root))
|
||||
stack.enter_context(mock.patch.object(ota.shutil, "which", return_value="cargo"))
|
||||
stack.enter_context(mock.patch.object(ota.sys.stdin, "isatty", return_value=True))
|
||||
self.prompt = stack.enter_context(mock.patch("builtins.input", return_value="yes"))
|
||||
self.check = stack.enter_context(mock.patch.object(ota, "check_bootloader_tool"))
|
||||
self.install = stack.enter_context(mock.patch.object(ota.subprocess, "run", side_effect=self.fake_install))
|
||||
self.output = stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
|
||||
|
||||
def fake_install(self, command, **kwargs):
|
||||
self.binary.parent.mkdir(parents=True, exist_ok=True)
|
||||
self.binary.write_bytes(b"test-only installer output, not executable")
|
||||
return subprocess.CompletedProcess(command, 0)
|
||||
|
||||
def repair(self):
|
||||
ota.offer_motatool_repair(self.args, self.probe, self.failure)
|
||||
|
||||
def test_consent_installs_pinned_private_tool_rechecks_and_selects_it(self):
|
||||
with mock.patch.dict(os.environ, {"MESHCORE_ADMIN_PASSWORD": "never-pass-to-build"}):
|
||||
self.repair()
|
||||
self.prompt.assert_called_once()
|
||||
self.install.assert_called_once()
|
||||
command = self.install.call_args.args[0]
|
||||
self.assertEqual(command, ota.motatool_repair_command(self.root))
|
||||
self.assertIn(ota.MOTATOOL_REPAIR_REPOSITORY, command)
|
||||
self.assertIn(ota.MOTATOOL_REPAIR_REVISION, command)
|
||||
self.assertIn("--locked", command)
|
||||
self.assertNotIn("MESHCORE_ADMIN_PASSWORD", self.install.call_args.kwargs["env"])
|
||||
self.assertEqual(self.install.call_args.kwargs["stdin"], subprocess.DEVNULL)
|
||||
self.assertEqual(self.install.call_args.kwargs["timeout"], ota.MOTATOOL_REPAIR_TIMEOUT_SECONDS)
|
||||
self.check.assert_called_once_with(str(self.binary), self.probe)
|
||||
self.assertEqual(self.args.motatool, str(self.binary))
|
||||
self.assertIn("Existing motatool installations and PATH will not be changed", self.output.getvalue())
|
||||
|
||||
def test_decline_or_blank_never_creates_install_directory(self):
|
||||
for answer in ("", "no", "n", "anything else"):
|
||||
with self.subTest(answer=answer):
|
||||
self.prompt.return_value = answer
|
||||
with self.assertRaisesRegex(ota.OtaError, "declined"):
|
||||
self.repair()
|
||||
self.assertFalse(self.root.exists())
|
||||
self.install.assert_not_called()
|
||||
self.assertEqual(self.args.motatool, "old-motatool")
|
||||
|
||||
def test_eof_does_not_approve_install(self):
|
||||
self.prompt.side_effect = EOFError
|
||||
with self.assertRaisesRegex(ota.OtaError, "declined"):
|
||||
self.repair()
|
||||
self.assertFalse(self.root.exists())
|
||||
self.install.assert_not_called()
|
||||
|
||||
def test_yes_flag_never_approves_repair_in_noninteractive_run(self):
|
||||
self.assertTrue(self.args.yes)
|
||||
with mock.patch.object(ota.sys.stdin, "isatty", return_value=False):
|
||||
with self.assertRaisesRegex(ota.OtaError, "--yes does not approve"):
|
||||
self.repair()
|
||||
self.prompt.assert_not_called()
|
||||
self.install.assert_not_called()
|
||||
self.assertIn("Manual install command:", self.output.getvalue())
|
||||
self.assertIn("--motatool", self.output.getvalue())
|
||||
self.assertFalse(self.root.exists())
|
||||
|
||||
def test_missing_cargo_gives_prerequisite_without_installing_anything(self):
|
||||
with mock.patch.object(ota.shutil, "which", return_value=None):
|
||||
with self.assertRaisesRegex(ota.OtaError, "https://rustup.rs"):
|
||||
self.repair()
|
||||
self.install.assert_not_called()
|
||||
self.prompt.assert_not_called()
|
||||
self.assertFalse(self.root.exists())
|
||||
|
||||
def test_cargo_failure_stops_and_preserves_original_selection(self):
|
||||
self.install.side_effect = None
|
||||
self.install.return_value = subprocess.CompletedProcess([], 101)
|
||||
with self.assertRaisesRegex(ota.OtaError, "Cargo exit 101"):
|
||||
self.repair()
|
||||
self.check.assert_not_called()
|
||||
self.assertEqual(self.args.motatool, "old-motatool")
|
||||
|
||||
def test_cargo_timeout_or_launch_failure_stops(self):
|
||||
for error in (OSError("cannot start compiler"), subprocess.TimeoutExpired("cargo", 3600)):
|
||||
with self.subTest(error=error):
|
||||
self.install.side_effect = error
|
||||
with self.assertRaisesRegex(ota.OtaError, "repair failed"):
|
||||
self.repair()
|
||||
self.check.assert_not_called()
|
||||
self.assertEqual(self.args.motatool, "old-motatool")
|
||||
|
||||
def test_installer_success_without_binary_is_not_success(self):
|
||||
self.install.side_effect = None
|
||||
self.install.return_value = subprocess.CompletedProcess([], 0)
|
||||
with self.assertRaisesRegex(ota.OtaError, "did not produce"):
|
||||
self.repair()
|
||||
self.check.assert_not_called()
|
||||
self.assertEqual(self.args.motatool, "old-motatool")
|
||||
|
||||
def test_failed_postinstall_check_does_not_select_bad_tool(self):
|
||||
self.check.side_effect = ota.OtaError("still unsupported")
|
||||
with self.assertRaisesRegex(ota.OtaError, "still failed bootloader verification"):
|
||||
self.repair()
|
||||
self.assertEqual(self.args.motatool, "old-motatool")
|
||||
|
||||
def test_default_tool_reuses_rechecked_cache_without_download_or_prompt(self):
|
||||
self.args.motatool = "motatool"
|
||||
self.fake_install([])
|
||||
with mock.patch.object(ota.sys.stdin, "isatty", return_value=False):
|
||||
self.repair()
|
||||
self.prompt.assert_not_called()
|
||||
self.install.assert_not_called()
|
||||
self.check.assert_called_once_with(str(self.binary), self.probe)
|
||||
self.assertEqual(self.args.motatool, str(self.binary))
|
||||
|
||||
def test_custom_tool_is_not_silently_replaced_even_with_valid_cache(self):
|
||||
self.fake_install([])
|
||||
self.prompt.return_value = "no"
|
||||
with self.assertRaisesRegex(ota.OtaError, "declined"):
|
||||
self.repair()
|
||||
self.assertIn("cached motatool", self.prompt.call_args.args[0])
|
||||
self.assertEqual(self.args.motatool, "old-motatool")
|
||||
self.install.assert_not_called()
|
||||
|
||||
def test_approved_cached_tool_needs_no_cargo(self):
|
||||
self.fake_install([])
|
||||
with mock.patch.object(ota.shutil, "which", return_value=None):
|
||||
self.repair()
|
||||
self.install.assert_not_called()
|
||||
self.assertEqual(self.check.call_count, 2)
|
||||
self.assertEqual(self.args.motatool, str(self.binary))
|
||||
|
||||
def test_bad_cache_requires_new_consent_and_recheck(self):
|
||||
self.args.motatool = "motatool"
|
||||
self.fake_install([])
|
||||
self.check.side_effect = [ota.OtaError("old cached tool"), None]
|
||||
self.repair()
|
||||
self.prompt.assert_called_once()
|
||||
self.install.assert_called_once()
|
||||
self.assertEqual(self.check.call_count, 2)
|
||||
|
||||
def test_working_tool_does_not_offer_or_install(self):
|
||||
with mock.patch.object(ota, "offer_motatool_repair") as offer:
|
||||
ota.require_bootloader_tool_support(self.args)
|
||||
offer.assert_not_called()
|
||||
self.install.assert_not_called()
|
||||
|
||||
def test_missing_or_incompatible_tool_reaches_the_offer(self):
|
||||
self.check.side_effect = self.failure
|
||||
with mock.patch.object(ota, "offer_motatool_repair") as offer:
|
||||
ota.require_bootloader_tool_support(self.args)
|
||||
offer.assert_called_once_with(self.args, self.probe, self.failure)
|
||||
|
||||
def test_repair_failure_in_main_never_contacts_radios(self):
|
||||
self.check.side_effect = self.failure
|
||||
self.prompt.return_value = "no"
|
||||
with (
|
||||
mock.patch.object(ota, "Controller") as controller,
|
||||
mock.patch.object(ota, "source_cli_command") as source,
|
||||
contextlib.redirect_stderr(io.StringIO()),
|
||||
):
|
||||
status = ota.main([
|
||||
str(self.probe), "remote", "--controller-serial", "controller",
|
||||
"--source-serial", "source", "--no-install", "--yes",
|
||||
])
|
||||
self.assertEqual(status, 2)
|
||||
self.prompt.assert_called_once()
|
||||
self.install.assert_not_called()
|
||||
controller.assert_not_called()
|
||||
source.assert_not_called()
|
||||
|
||||
def test_repaired_binary_is_selected_before_meshcli_preflight(self):
|
||||
self.check.side_effect = [self.failure, None]
|
||||
args = ota.build_parser().parse_args([str(self.probe), "remote", "--no-install"])
|
||||
with mock.patch.object(ota, "require_meshcli_version") as meshcli:
|
||||
ota.preflight_inputs(args)
|
||||
self.assertEqual(args.motatool, str(self.binary))
|
||||
meshcli.assert_called_once()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -82,4 +82,6 @@ from the generated application target table.
|
||||
This library does not authorize a device update. A capable node will only arm
|
||||
such a v3 package through the exact manual confirmation described in
|
||||
[`docs/ota_nrf52_bootloader_update.md`](../../docs/ota_nrf52_bootloader_update.md).
|
||||
The ordinary `tools/lora_ota` runner rejects bootloader packages.
|
||||
The `tools/lora_ota` runner accepts verified bootloader packages only with
|
||||
`--no-install`, checks the destination's bootloader-specific identity and
|
||||
capabilities, and leaves installation to that explicit manual confirmation.
|
||||
Reference in new issue
Block a user