diff --git a/docs/ota_wizard.md b/docs/ota_wizard.md index 51794ffbc..307a4bb04 100644 --- a/docs/ota_wizard.md +++ b/docs/ota_wizard.md @@ -26,9 +26,11 @@ containing `ota.peer` first; older Repeater firmware cannot manage the remote node through USB. Check with `ota.peer status` in its USB console. The Full Companion source also needs a current build containing the wizard's -direct admin guard. An older Full Companion may serve OTA files but cannot +version 2 direct admin support. An older Full Companion may serve OTA files but cannot provide this guarded admin connection; the wizard will ask for a source firmware upgrade instead of proceeding with that connection. +Check `get ota.direct` from its USB console for `v=2`. Update replies use a +separate mailbox; ordinary queued messages and saved contact routes stay intact. For this wizard, use an ESP32 source instead of RAK4631 Full. The RAK image's current flash limit prevents adding the new direct admin helper. RAK4631 @@ -107,6 +109,9 @@ when a compatible builder and the **exact running firmware** are available and the package is smaller. A file with the same version label but a different body hash is not a usable base. Published bases are found through GitHub releases; a local base can be selected when the running image was custom. +Unsigned full `.mota` files can also supply the new image for a differential +update. Signed packages retain their signatures; prepared bundles retain their +chosen transfer format. On a computer, the differential builder can be prepared automatically with a compatible Rust toolchain. On a Pi without a builder, a full update is made by @@ -157,9 +162,16 @@ session for the selected destination before proceeding. Keep the recovery record until restoration is complete. A lost install reply is followed by a running-image check; the wizard does not blindly repeat an -install command. Temporary radio windows use the existing main-channel +install command. A cancelled download is cleared only when its package ID +matches this update. Automatic installation is restored only after installation +or receive-slot clearance is confirmed; an uncertain result leaves it off and +keeps the recovery record. + +Temporary radio windows use the existing main-channel settings. Existing temporary schedules block an update before settings are changed, including on older firmware. +The window covers both bootloader and firmware transfers, including reboot +and setup time. Status checks normally run every ten seconds. The cache is normally `~/.cache/meshcore/ota` on Linux, `~/Library/Caches/MeshCore/ota` on macOS, or `%LOCALAPPDATA%\MeshCore\ota` on diff --git a/tools/lora_ota/ota_wizard.py b/tools/lora_ota/ota_wizard.py index 0a896104e..c57e2e423 100644 --- a/tools/lora_ota/ota_wizard.py +++ b/tools/lora_ota/ota_wizard.py @@ -22,9 +22,9 @@ import lora_ota as ota from ota_wizard_catalog import GitHubCatalog, ImageCache, download_asset, load_application from ota_wizard_packages import (find_builder, find_running_image, install_builder, prepare_application, prepare_bootloader, small_host) -from ota_wizard_session import DirectSession +from ota_wizard_session import DirectSession, validate_package from ota_wizard_bundle import (load_bundle, load_ready_application, load_request, - save_bundle, save_request) + save_bundle, save_request, VerificationDependencyError) from ota_wizard_usb import open_serial_session ROOT = Path(__file__).resolve().parents[2] @@ -484,16 +484,23 @@ async def prepare_selected_firmware(args, destination, firmware, catalog): if prepared: image, application = prepared if args.full and not application.info.is_full: + if application.signer: + raise ota.OtaError("This signed delta cannot be converted without losing its signature. " + "Choose Automatic or provide a signed full package.") application = await asyncio.to_thread(prepare_application, image, destination, args.cache, full=True) else: image = await asyncio.to_thread(load_application, firmware, target_id=destination.target_id, hw_id=destination.hw_id) - application = await asyncio.to_thread(load_ready_application, firmware, image, destination) + application = await asyncio.to_thread(load_ready_application, firmware, image, destination, + validate_receiver=False) + if application and not application.signer and not args.full: + application = None # Automatic also considers deltas for unsigned full mOTA inputs. ImageCache(args.cache).remember(image) if image.fw_version <= destination.version: print("The selected firmware is already installed or older; keeping the running firmware.") return image, None if application: + validate_package(application, destination) return image, application builder = None if args.full else find_builder(args.cache, args.motatool) if not args.full and not builder and not small_host(): @@ -501,6 +508,9 @@ async def prepare_selected_firmware(args, destination, firmware, catalog): base = None if builder: base = await asyncio.to_thread(find_running_image, destination, args.cache, catalog, supplied=args.base) + if base is None and catalog is None and args.base is None: + catalog = GitHubCatalog() + base = await asyncio.to_thread(find_running_image, destination, args.cache, catalog) if base is None and getattr(args, "interactive_menu", False): base = await choose_missing_base(args, destination, catalog) if not builder and not args.full: @@ -529,7 +539,7 @@ async def prepare_interactively(args) -> int: pass # A working system verifier may already be loaded. try: image, package = await prepare_selected_firmware(args, destination, firmware, catalog) - except ImportError: + except (ImportError, VerificationDependencyError): # Unsigned full packaging stays dependency-free. Signed inputs request # their verifier only when it is actually needed; no USB SDK is loaded. await asyncio.to_thread(ensure_signature_support) @@ -703,17 +713,9 @@ def prepare_without_radios(args) -> int: if not args.request or not args.firmware or not args.output: raise ota.OtaError("Use --prepare-only --request DEVICE.json --firmware FILE --output update.ota.zip.") destination = load_request(args.request) - image = load_application(args.firmware, target_id=destination.target_id, hw_id=destination.hw_id) - package = load_ready_application(args.firmware, image, destination) + image, package = asyncio.run(prepare_selected_firmware(args, destination, args.firmware, None)) if package is None: - catalog = GitHubCatalog() - builder = find_builder(args.cache, args.motatool) - if not builder and not args.full: - builder = install_builder(args.cache) - base = None - if builder and not args.full: - base = find_running_image(destination, args.cache, catalog, supplied=args.base) - package = prepare_application(image, destination, args.cache, base=base, builder=builder, full=args.full) + raise ota.OtaError("Choose firmware newer than the running version.") save_bundle(args.output, image, package) print(f"Prepared {package.info.kind}: {args.output} ({len(package.blob):,} transfer bytes)") print("Copy this ZIP to the Pi and select it as the custom firmware file.") diff --git a/tools/lora_ota/ota_wizard_bundle.py b/tools/lora_ota/ota_wizard_bundle.py index 809a796fd..68f6e85be 100644 --- a/tools/lora_ota/ota_wizard_bundle.py +++ b/tools/lora_ota/ota_wizard_bundle.py @@ -12,8 +12,12 @@ import lora_ota as ota from ota_wizard_session import Destination, Package, validate_package +class VerificationDependencyError(ota.OtaError): + """The package needs the optional signature verifier.""" + + def load_ready_application(path: Path, image: ota.EndFInfo, - destination: Destination) -> Package | None: + destination: Destination, *, validate_receiver: bool = True) -> Package | None: """Keep a supplied full mOTA intact, including its verified signing key.""" if Path(path).suffix.lower() != ".mota": return None @@ -35,9 +39,16 @@ def load_ready_application(path: Path, image: ota.EndFInfo, parsed = lib.parse_container(blob) signer = parsed.manifest.signer_pubkey.hex().upper() if parsed.manifest.is_signed else None package = Package(blob, expected.body_hash, signer) - validate_package(package, destination) + if validate_receiver: + validate_package(package, destination) + else: + # A full unsigned container can supply the complete new image for + # a delta even when the receiver cannot install that full format. + problems = lib.verify(parsed) + if problems: + raise ota.OtaError("Application mOTA integrity failed: " + "; ".join(problems)) except ImportError as exc: - raise ota.OtaError("Install Python cryptography to verify this signed application mOTA.") from exc + raise VerificationDependencyError("Signature verification support is required for this application mOTA.") from exc except (ValueError, TypeError) as exc: raise ota.OtaError("Invalid supplied application mOTA: " + str(exc)) from exc return package @@ -148,7 +159,10 @@ def _load_bundle(path: Path, destination: Destination) -> tuple[ota.EndFInfo, Pa or info.fw_version != image.fw_version): raise ota.OtaError("The complete firmware does not match the prepared package.") lib = ota.bootloader_library() - problems = lib.verify(lib.parse_container(blob)) + try: + problems = lib.verify(lib.parse_container(blob)) + except ImportError as exc: + raise VerificationDependencyError("Signature verification support is required for this OTA bundle.") from exc if problems: raise ota.OtaError("Prepared OTA package integrity failed: " + "; ".join(problems)) parsed = lib.parse_container(blob) diff --git a/tools/lora_ota/ota_wizard_catalog.py b/tools/lora_ota/ota_wizard_catalog.py index 1a9377a45..ff862a32d 100644 --- a/tools/lora_ota/ota_wizard_catalog.py +++ b/tools/lora_ota/ota_wizard_catalog.py @@ -300,8 +300,31 @@ class GitHubCatalog: def candidates(self, target_env: str | None = None, *, target_id: int | None = None, profile: str = "full", include_embedded: bool = False) -> list[ReleaseAsset]: - assets = self.fetch_assets() if self._assets is None else self._assets - return self._compatible(assets, target_env, target_id, profile, include_embedded) + if target_env is None and target_id is None: + raise ota.OtaError("Select an exact firmware target") + if self._assets is not None and not self._pages: + # An explicitly supplied asset list has no further release pages. + return self._compatible(self._assets, target_env, target_id, profile, include_embedded) + pages = max(3, max(self._pages, default=0)) + while True: + releases = [release for release in self._releases(pages) if not release.prerelease] + mains = sorted((release for release in releases if re.match(r"^v\d", release.tag, re.I)), + key=lambda release: (_version(release.tag), release.published_at), reverse=True) + matches = [] + for main in mains: + assets = [asset for release in releases if _release_family(release.tag) == main.tag + for asset in release.assets] + matches.append(self._compatible(assets, target_env, target_id, profile, include_embedded)) + if matches and matches[0]: + return matches[0] + # A missing board in the newest family may be in a later split + # shard. Finish the bounded search before selecting an older family. + if pages < 10 and self.has_more_releases: + pages = min(10, pages + 3) + continue + if not mains: + raise ota.OtaError("No published main firmware release was found") + return next((choice for choice in matches if choice), []) def choose_asset(self, target_env: str | None = None, **kwargs) -> ReleaseAsset: choices = self.candidates(target_env, **kwargs) diff --git a/tools/lora_ota/test_ota_wizard_catalog.py b/tools/lora_ota/test_ota_wizard_catalog.py index cddac3b09..eb907ca25 100644 --- a/tools/lora_ota/test_ota_wizard_catalog.py +++ b/tools/lora_ota/test_ota_wizard_catalog.py @@ -92,6 +92,65 @@ class CatalogTest(unittest.TestCase): with self.assertRaises(ota.OtaError): source.choose_asset("another_repeater") + def test_latest_compatible_uses_previous_stable_family_when_newest_omits_board(self): + newest = "v1.18.0.2" + rows = [release(newest, [asset_row(f"another_repeater-full-ota-{newest}.zip", newest)]), + release(FAMILY, [asset_row()]), + release("v1.19.0-preview.1", [asset_row( + f"{ENV}-full-ota-v1.19.0-preview.1.zip", "v1.19.0-preview.1")])] + source = catalog.GitHubCatalog(opener=lambda req, timeout: Response(json.dumps(rows).encode(), req.full_url)) + self.assertEqual(source.choose_asset(target_id=TARGET).release_tag, FAMILY) + + def test_latest_compatible_keeps_profile_and_role_gates_across_families(self): + newest = "v1.18.0.2" + rows = [release(newest, [ + asset_row(f"{ENV}-reduced-ota-{newest}.zip", newest), + asset_row(f"{ENV}-with-bootloader-2.4.12-full-ota-{newest}.zip", newest), + asset_row(f"ikoka_stick_nrf_33dbm_room_server-full-ota-{newest}.zip", newest)]), + release(FAMILY, [asset_row()])] + source = catalog.GitHubCatalog(opener=lambda req, timeout: Response(json.dumps(rows).encode(), req.full_url)) + self.assertEqual(source.choose_asset(ENV).release_tag, FAMILY) + self.assertEqual(source.choose_asset(ENV, profile="reduced").release_tag, newest) + + def test_latest_compatible_loads_late_new_family_shard_before_using_older_match(self): + newest = "v1.18.0.2" + shard = f"lora-ota-3-{newest}" + pages = { + 1: [release(newest), release(FAMILY, [asset_row()])], + 4: [release(shard, [asset_row(f"{ENV}-full-ota-{newest}.zip", shard)])], + } + calls = [] + def opener(request, timeout): + page = int(request.full_url.rsplit("=", 1)[1]) + calls.append(page) + rows = pages.get(page, []) + if page < 4: + rows += [release(f"v0.0.0-empty-{page}-{index}") for index in range(10-len(rows))] + return Response(json.dumps(rows).encode(), request.full_url) + source = catalog.GitHubCatalog(opener=opener) + selected = source.choose_asset(target_id=TARGET) + self.assertEqual(selected.release_tag, shard) + self.assertEqual(selected.version, 0x01120002) + self.assertEqual(calls, [1, 2, 3, 4]) + self.assertEqual(source.choose_asset(target_id=TARGET), selected) + self.assertEqual(calls, [1, 2, 3, 4]) + + def test_latest_compatible_search_is_bounded_and_reuses_loaded_pages(self): + calls = [] + def opener(request, timeout): + page = int(request.full_url.rsplit("=", 1)[1]) + calls.append(page) + rows = [release("v1.18.0.2")] if page == 1 else [] + rows += [release(f"v0.0.0-empty-{page}-{index}") for index in range(10-len(rows))] + return Response(json.dumps(rows).encode(), request.full_url) + source = catalog.GitHubCatalog(opener=opener) + with self.assertRaisesRegex(ota.OtaError, "No published application"): + source.choose_asset(target_id=TARGET) + self.assertEqual(calls, list(range(1, 11))) + with self.assertRaisesRegex(ota.OtaError, "No published application"): + source.choose_asset(target_id=TARGET) + self.assertEqual(calls, list(range(1, 11))) + def test_release_menu_groups_shards_and_offers_exact_board_previews(self): shard = f"lora-ota-10-{FAMILY}" preview = "ikoka-v1.18.0.2-alpha2" diff --git a/tools/lora_ota/test_ota_wizard_prepare_regressions.py b/tools/lora_ota/test_ota_wizard_prepare_regressions.py new file mode 100644 index 000000000..e768bf63e --- /dev/null +++ b/tools/lora_ota/test_ota_wizard_prepare_regressions.py @@ -0,0 +1,156 @@ +"""Exercise container conversion and lazy verification through the real planner.""" +from __future__ import annotations + +import asyncio +import builtins +from contextlib import redirect_stdout +from dataclasses import replace +import io +from pathlib import Path +import tempfile +from types import SimpleNamespace +import unittest +from unittest import mock + +import lora_ota as ota +import ota_wizard as wizard +import ota_wizard_bundle as bundles +import ota_wizard_packages as packages +import ota_wizard_session as session +from test_lora_ota import mota_blob +from test_ota_wizard_bundle import destination, image, signed_package + + +class PreparationRegressionTests(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory(prefix="meshcore-prepare-regression-") + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name) + self.live = destination() + self.current = image(self.live.version, body=b"original application" * 16) + self.new = image(body=b"new complete application" * 320) + self.base = self.root / "running.bin" + self.base.write_bytes(self.current.image) + self.full = session.full_package(self.new, 1024) + self.firmware = self.root / "new-full.mota" + self.firmware.write_bytes(self.full.blob) + + def options(self, *extra, cache="cache"): + return wizard.arguments(["--cache", str(self.root / cache), "--base", str(self.base), *extra]) + + def test_automatic_full_container_can_become_a_smaller_delta_for_external_and_internal_nrf(self): + for external in (True, False): + with self.subTest(external_storage=external): + live = replace(self.live, external_storage=external) + options = self.options(cache="external" if external else "internal") + commands = [] + + def build(command, **_kwargs): + commands.append(command) + old = ota.parse_endf(Path(command[command.index("--base") + 1]).read_bytes()) + new = ota.parse_endf(Path(command[command.index("--fw") + 1]).read_bytes()) + self.assertEqual(old.image, self.current.image) + self.assertEqual(new.image, self.new.image) + output = Path(command[command.index("--out") + 1]) + # The builder boundary is fake; the actual manifest, Merkle + # proof, base, target, codec and new-image validation run. + output.write_bytes(mota_blob(new.image, full=False, + version=new.fw_version, target=new.target_id, hw=new.hw_id, + base_hash=old.body_hash, codec=ota.MOTA_CODEC_IN_PLACE, + payload=b"small differential container")) + return SimpleNamespace(returncode=0) + + if not external: + with self.assertRaisesRegex(ota.OtaError, "Internal-flash"): + session.validate_package(self.full, live) + with mock.patch.object(wizard, "find_builder", return_value="fake-builder") as builder, \ + mock.patch.object(wizard, "install_builder", side_effect=AssertionError("unneeded setup")), \ + mock.patch.object(packages.subprocess, "run", side_effect=build), \ + redirect_stdout(io.StringIO()): + selected, package = asyncio.run(wizard.prepare_selected_firmware( + options, live, self.firmware, None)) + builder.assert_called_once() + self.assertEqual(len(commands), 1) + self.assertEqual(commands[0][commands[0].index("--patch-type") + 1], "in-place") + self.assertEqual(selected.image, self.new.image) + self.assertFalse(package.info.is_full) + self.assertEqual(package.info.base_hash, self.live.base_hash) + self.assertEqual(package.body_hash, self.new.body_hash) + self.assertLess(len(package.blob), len(self.full.blob)) + session.validate_package(package, live) + + def test_explicit_full_keeps_the_original_unsigned_container_bytes(self): + with mock.patch.object(wizard, "find_builder", side_effect=AssertionError("must stay full")), \ + mock.patch.object(packages.subprocess, "run", side_effect=AssertionError("must not build")), \ + redirect_stdout(io.StringIO()): + selected, package = asyncio.run(wizard.prepare_selected_firmware( + self.options("--full"), self.live, self.firmware, None)) + self.assertEqual(selected.image, self.new.image) + self.assertEqual(package.blob, self.full.blob) + self.assertEqual(package.info.block_size, 1024) + + def test_automatic_preserves_a_signed_full_sd_package_without_a_builder(self): + signed = signed_package(self.new) + self.firmware.write_bytes(signed.blob) + with mock.patch.object(wizard, "find_builder", side_effect=AssertionError("signature must stay intact")), \ + mock.patch.object(wizard, "find_running_image", side_effect=AssertionError("no delta base needed")), \ + mock.patch.object(packages.subprocess, "run", side_effect=AssertionError("must not repackage")), \ + redirect_stdout(io.StringIO()): + selected, package = asyncio.run(wizard.prepare_selected_firmware( + self.options(), replace(self.live, sd_storage=True), self.firmware, None)) + self.assertEqual(selected.image, self.new.image) + self.assertEqual(package.blob, signed.blob) + self.assertEqual(package.signer, signed.signer) + session.validate_package(package, replace(self.live, sd_storage=True)) + + def test_missing_verifier_in_real_signed_mota_and_bundle_paths_is_installed_then_retried(self): + # Build/sign fixtures before simulating a fresh Python installation. + signed = signed_package(self.new) + signed_file = self.root / "signed-full.mota" + signed_file.write_bytes(signed.blob) + prepared_file = self.root / "signed-prepared.ota.zip" + bundles.save_bundle(prepared_file, self.new, signed) + original_import = builtins.__import__ + live = replace(self.live, sd_storage=True) + + for firmware in (signed_file, prepared_file): + with self.subTest(input=firmware.name): + blocked = [True] + missing_imports = [] + + def import_without_crypto(name, *args, **kwargs): + if blocked[0] and (name == "cryptography" or name.startswith("cryptography.")): + missing_imports.append(name) + raise ModuleNotFoundError("No module named cryptography", name=name) + return original_import(name, *args, **kwargs) + + def install_verifier(): + self.assertTrue(missing_imports, "the real signature verifier must have failed first") + blocked[0] = False + + options = self.options(cache=firmware.stem) + with mock.patch.object(wizard, "choose_request", return_value=live), \ + mock.patch.object(wizard, "GitHubCatalog", return_value=object()), \ + mock.patch.object(wizard, "choose_firmware", new_callable=mock.AsyncMock, + return_value=(firmware, "local")) as choose, \ + mock.patch.object(ota, "cryptography_repair_root", return_value=self.root / "fresh-python"), \ + mock.patch.object(builtins, "__import__", side_effect=import_without_crypto), \ + mock.patch.object(wizard, "ensure_signature_support", side_effect=install_verifier) as setup, \ + mock.patch.object(wizard, "ensure_dependencies", side_effect=AssertionError("no USB setup")), \ + mock.patch.object(wizard, "choose_port", side_effect=AssertionError("no radio access")), \ + mock.patch.object(wizard, "find_builder", side_effect=AssertionError("no native builder")), \ + mock.patch.object(wizard, "ask", side_effect=lambda _prompt, default=None: default), \ + redirect_stdout(io.StringIO()): + self.assertEqual(asyncio.run(wizard.prepare_interactively(options)), 0) + setup.assert_called_once_with() + choose.assert_awaited_once() + outputs = list((options.cache / "bundles").glob("*.ota.zip")) + self.assertEqual(len(outputs), 1) + restored_image, restored_package = bundles.load_bundle(outputs[0], live) + self.assertEqual(restored_image.image, self.new.image) + self.assertEqual(restored_package.blob, signed.blob) + self.assertEqual(restored_package.signer, signed.signer) + + +if __name__ == "__main__": + unittest.main()