diff --git a/README.md b/README.md index 235cbf4..2d3fa95 100644 --- a/README.md +++ b/README.md @@ -40,7 +40,7 @@ The following radios should work with this library: - BTech UV-Pro - RadioOddity GA-5WB -- Vero VR-N76 (untested) +- Vero VR-N76 - Vero VR-N7500 (untested) - BTech GMRS-Pro (untested) @@ -93,7 +93,8 @@ Benlink has already begun to inspire other projects! Here are some that I know of so far: - [HTCommander](https://github.com/Ylianst/HTCommander) -- [flutter\_benlink](https://github.com/SarahRoseLives/flutter_benlink) +- [flutter_benlink](https://github.com/SarahRoseLives/flutter_benlink) +- [OpenHT](https://github.com/repins267/repins267-OpenHT) If you've found benlink's documentation of the Benshi protocol helpful, or use benlink in your own project, please let me know so I can add it to this list. @@ -113,9 +114,6 @@ Things to do: - [ ] Make a higher-level interface for sending / receiving TNC data (auto retry, queue message fragments) ([issue](https://github.com/khusmann/benlink/issues/1)) -- [ ] Figure out firmware flashing process / protocol (this is key for long-term - independence from the HT app) - ([issue](https://github.com/khusmann/benlink/issues/10)) - [ ] Implement more commands and settings - [ ] Find more radios that use this protocol and test them with this library @@ -126,6 +124,13 @@ receive [@na7q](https://github.com/na7q) for early testing and feedback +[@Ylianst](https://github.com/Ylianst) for a steady stream of protocol findings +and sharp questions along the way + +[@repins267](https://github.com/repins267) for turning my notes on the firmware +protocol into a complete proof of concept for flashing, working out the gRPC +update check, and having the guts to do the first flash. + ## Disclaimer This project is an independent grassroots effort, and is **not** affiliated with diff --git a/pyproject.toml b/pyproject.toml index 40118b8..1dd2ca6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,9 +25,18 @@ classifiers = [ ] requires-python = ">=3.10" +[project.optional-dependencies] +firmware = [ + "bsdiff4 >=1.2.0", + "grpcio >=1.60.0", +] + [project.urls] "Homepage" = "https://github.com/khusmann/benlink" "Bug Tracker" = "https://github.com/khusmann/benlink/issues" [tool.setuptools.packages.find] where = ["src"] + +[tool.pytest.ini_options] +pythonpath = ["src"] diff --git a/src/benlink/__init__.py b/src/benlink/__init__.py index 636bbf7..2e5ec55 100644 --- a/src/benlink/__init__.py +++ b/src/benlink/__init__.py @@ -33,7 +33,7 @@ - BTech UV-Pro - RadioOddity GA-5WB -- Vero VR-N76 (untested) +- Vero VR-N76 - Vero VR-N7500 (untested) - BTech GMRS-Pro (untested) @@ -86,7 +86,8 @@ async def main(): of so far: - [HTCommander](https://github.com/Ylianst/HTCommander) -- [flutter\_benlink](https://github.com/SarahRoseLives/flutter_benlink) +- [flutter_benlink](https://github.com/SarahRoseLives/flutter_benlink) +- [OpenHT](https://github.com/repins267/repins267-OpenHT) If you've found benlink's documentation of the Benshi protocol helpful, or use benlink in your own project, please let me know so I can add it to this list. @@ -106,9 +107,6 @@ async def main(): - [ ] Make a higher-level interface for sending / receiving TNC data (auto retry, queue message fragments) ([issue](https://github.com/khusmann/benlink/issues/1)) -- [ ] Figure out firmware flashing process / protocol (this is key for long-term - independence from the HT app) - ([issue](https://github.com/khusmann/benlink/issues/10)) - [ ] Implement more commands and settings - [ ] Find more radios that use this protocol and test them with this library @@ -119,6 +117,13 @@ async def main(): [@na7q](https://github.com/na7q) for early testing and feedback +[@Ylianst](https://github.com/Ylianst) for a steady stream of protocol findings +and sharp questions along the way + +[@repins267](https://github.com/repins267) for turning my notes on the firmware +protocol into a complete proof of concept for flashing, working out the gRPC +update check, and having the guts to do the first flash. + # Disclaimer This project is an independent grassroots effort, and is **not** affiliated with @@ -131,5 +136,6 @@ async def main(): from . import controller from . import command from . import audio +from . import firmware -__all__ = ['controller', 'command', 'audio'] \ No newline at end of file +__all__ = ['controller', 'command', 'audio', 'firmware'] diff --git a/src/benlink/command.py b/src/benlink/command.py index 8773078..e641458 100644 --- a/src/benlink/command.py +++ b/src/benlink/command.py @@ -35,8 +35,9 @@ from __future__ import annotations import typing as t import asyncio -from pydantic import BaseModel, ConfigDict +import contextlib from . import protocol as p +from .common import ImmutableBaseModel from .link import CommandLink, BleCommandLink, RfcommCommandLink from datetime import datetime @@ -82,6 +83,36 @@ async def send_bytes(self, data: bytes) -> None: async def send_message(self, command: CommandMessage) -> None: await self._link.send(command_message_to_protocol(command)) + async def send_protocol_message(self, msg: p.Message) -> None: + """Send a raw protocol message. + + For messages with no `CommandMessage` equivalent. Not the same as + `send_bytes`: the Rfcomm link wraps what it sends in a `GaiaFrame`. + """ + await self._link.send(msg) + + @contextlib.asynccontextmanager + async def subscribe( + self, + match: t.Callable[[RadioMessage], bool] | None = None, + ) -> t.AsyncGenerator[asyncio.Queue[RadioMessage], None]: + """Collect matching messages into a queue while the context is held. + + Enter this before sending whatever provokes the replies, so that a reply + arriving faster than the next `await` isn't dropped. + """ + queue: asyncio.Queue[RadioMessage] = asyncio.Queue() + + def handler(msg: RadioMessage) -> None: + if match is None or match(msg): + queue.put_nowait(msg) + + remove_handler = self._add_message_handler(handler) + try: + yield queue + finally: + remove_handler() + async def send_message_expect_reply(self, command: CommandMessage, expect: t.Type[RadioMessageT]) -> RadioMessageT | MessageReplyError: queue: asyncio.Queue[RadioMessageT | MessageReplyError] = asyncio.Queue() @@ -241,13 +272,6 @@ async def __aexit__( await self.disconnect() -class ImmutableBaseModel(BaseModel): - """@private (A base class for immutable data objects)""" - - model_config = ConfigDict(frozen=True) - """@private""" - - def command_message_to_protocol(m: CommandMessage) -> p.Message: """@private (Protocol helper)""" match m: diff --git a/src/benlink/common.py b/src/benlink/common.py new file mode 100644 index 0000000..e3e29e0 --- /dev/null +++ b/src/benlink/common.py @@ -0,0 +1,16 @@ +""" +Shared building blocks for benlink's data objects. + +Kept separate from `benlink.command` so that modules which don't talk to a radio +(e.g. `benlink.firmware`) can use them without pulling in the Bluetooth stack. +""" + +from __future__ import annotations +from pydantic import BaseModel, ConfigDict + + +class ImmutableBaseModel(BaseModel): + """@private (A base class for immutable data objects)""" + + model_config = ConfigDict(frozen=True) + """@private""" diff --git a/src/benlink/firmware/__init__.py b/src/benlink/firmware/__init__.py new file mode 100644 index 0000000..def5771 --- /dev/null +++ b/src/benlink/firmware/__init__.py @@ -0,0 +1,140 @@ +"""Finding, downloading, assembling and flashing Benshi radio firmware. + +# THIS CAN BREAK YOUR RADIO + +**Flashing firmware can leave your radio unusable, and nothing in this library can +undo it.** + +# Disclaimer + +**Use this at your own risk. I am not responsible for bricking your radio, or for +any other damage to your equipment.** This module is not endorsed by or affiliated +with Benshi, Vero, RadioOddity, BTech, or any other company. + +# The intended flow + +Firmware ships as a shared **base image** plus a per-release **patch** in BSDIFF40 +format; assembling the two yields the image the radio expects. benlink +redistributes neither, and fetches both on request. + +One command walks the whole upgrade, prompting as it goes: + +```bash +python -m benlink.firmware update XX:XX:XX:XX:XX:XX +``` + +It reads the product id and installed version from the radio, asks the update +server for the latest release, downloads the patch and base, assembles them, and +checks the result against the server's md5. Because the server names both +artifacts, this path cannot pair a patch with the wrong base. + +Add `--rfcomm CHANNEL` for RFCOMM instead of BLE, `--keep DIR` to write somewhere +durable, `-y` to accept prompts. + +Ctrl+C safely aborts a transfer. + +# The pieces + +Each step is also available alone, for archiving old releases or working away from +the radio. Everything but `info`, `flash` and `abort` avoids the Bluetooth stack. + +```bash +# which radio is this? +python -m benlink.firmware info XX:XX:XX:XX:XX:XX + +# what is the latest release? +python -m benlink.firmware check --product UV_PRO + +# that release, downloaded and assembled, without a radio +python -m benlink.firmware fetch --product UV_PRO -o fw.bin + +# one artifact at a time, for any version +python -m benlink.firmware download-patch --version 128 --product UV_PRO -o patch.bin +python -m benlink.firmware download-base --version original -o base.zip + +# combine them offline +python -m benlink.firmware assemble --base base.zip --patch patch.bin -o fw.bin + +# put an image you already have onto the radio +python -m benlink.firmware flash XX:XX:XX:XX:XX:XX --image fw.bin + +# clear an update the radio was left partway through (if you didn't exit cleanly) +python -m benlink.firmware abort XX:XX:XX:XX:XX:XX +``` + +`--product` is a shorthand for the radios in `PRODUCTS`; `--product-id` works for +any radio, and `info` tells you yours. If yours isn't listed, please +[open an issue](https://github.com/khusmann/benlink/issues) with what `info` +reports so it can be added. + +# Verification + +A BSDIFF40 patch carries no checksum of the base it was built against, so pairing +a patch with the wrong base **succeeds silently** and produces a corrupt image of +plausible length. See `BASE_IMAGES` for the known pairings. + +The server publishes an md5 of the *assembled* image for the current release, so +`update` and `fetch` are checked end to end. Older releases have none; for those, +`assemble --expect-md5` accepts one from elsewhere, such as the `md5sum_tail` in a +packet capture of an official flash. Every command that writes an image says +whether it could be verified. + +# Notes + +The product id comes from `GET_DEV_INFO` (`DeviceInfo.product_id`) and is not +unique across vendors: the VR-N76 and GA-5WB both report 259. +`DeviceInfo.firmware_version` shares the update server's numbering, so installed +and available versions compare directly. +""" + +from ._fetch import ( + BASE_IMAGES, + PRODUCTS, + FirmwareBundle, + FirmwareInfo, + ProgressCallback, + UpdateInfo, + assemble, + check_update, + download, + download_firmware, + extract_base, + fetch_firmware, + oss_base_url, + oss_patch_url, + oss_update_info, +) +from ._flash import FlashError, FlashResult, abort_update, flash + +# Grouped by what you reach for, in the order you reach for it, rather than +# alphabetically: pdoc lays the documentation page out in exactly this order. +__all__ = [ + # Which radios and base images exist + "PRODUCTS", + "BASE_IMAGES", + + # What a release looks like + "FirmwareInfo", + "UpdateInfo", + "FirmwareBundle", + "ProgressCallback", + + # Finding one + "check_update", + "oss_update_info", + "oss_patch_url", + "oss_base_url", + + # Downloading and assembling it + "fetch_firmware", + "download_firmware", + "download", + "extract_base", + "assemble", + + # Putting it on the radio + "flash", + "abort_update", + "FlashResult", + "FlashError", +] diff --git a/src/benlink/firmware/__main__.py b/src/benlink/firmware/__main__.py new file mode 100644 index 0000000..9ccc4ee --- /dev/null +++ b/src/benlink/firmware/__main__.py @@ -0,0 +1,617 @@ +"""Command line interface for `benlink.firmware`. + +Run with `python -m benlink.firmware`. +""" + +from __future__ import annotations +import typing as t +import argparse +import asyncio +import collections +import contextlib +import hashlib +import os +import signal +import sys +import tempfile +import time + +if t.TYPE_CHECKING: + from ..command import CommandConnection, DeviceInfo + +from . import ( + BASE_IMAGES, + PRODUCTS, + FirmwareInfo, + UpdateInfo, + abort_update, + assemble, + check_update, + download, + download_firmware, + extract_base, + flash, + oss_base_url, + oss_patch_url, +) + +_REBOOT_WAIT = 20.0 +"""Seconds to let the radio reboot before trying to reach it again.""" + +_COMMIT_ATTEMPTS = 5 + +_RATE_WINDOW = 30.0 +"""Seconds of history the transfer rate is averaged over.""" + + +##################### +# Output + +def _out(message: str = "") -> None: + print(message, file=sys.stderr) + + +def _size(n: float) -> str: + if n >= 1e6: + return f"{n / 1e6:.1f}MB" + if n >= 1e3: + return f"{n / 1e3:.0f}kB" + return f"{n:.0f}B" + + +def _duration(seconds: float) -> str: + total = int(seconds) + if total >= 3600: + return f"{total // 3600}h{(total % 3600) // 60:02d}m" + return f"{total // 60}:{total % 60:02d}" + + +def _make_progress() -> t.Callable[[str, int, int], None]: + """Render concurrent transfers as one updating line. + + Flashing over BLE runs for many minutes, so a bare percentage is not enough + to tell slow from stuck. + """ + recent: t.Dict[str, t.Deque[t.Tuple[float, int]]] = {} + state: t.Dict[str, t.Tuple[int, int]] = {} + width = 0 + + def render(label: str, done: int, total: int) -> str: + if not total: + return f"{label} {_size(done)}" + out = f"{label} {100 * done // total}% of {_size(total)}" + # Rate over a trailing window, not since the start. Connection setup and + # BLE's opening connection interval are slow enough that a running + # average reads far below the rate actually being achieved, and the eta + # derived from it is wrong by minutes. + window = recent[label] + elapsed = window[-1][0] - window[0][0] + moved = done - window[0][1] + if elapsed > 1.0 and moved > 0: + rate = moved / elapsed + out += f" {_size(rate)}/s eta {_duration((total - done) / rate)}" + return out + + def progress(label: str, done: int, total: int) -> None: + nonlocal width + now = time.monotonic() + window = recent.setdefault(label, collections.deque(maxlen=512)) + window.append((now, done)) + while len(window) > 2 and now - window[0][0] > _RATE_WINDOW: + window.popleft() + state[label] = (done, total) + line = " ".join(render(k, d, n) for k, (d, n) in state.items()) + # Pad to the widest line so far, or a shrinking eta leaves debris behind. + width = max(width, len(line)) + print(f"\r {line:<{width}}", end="", file=sys.stderr, flush=True) + + return progress + + +def _print_verdict(data: bytes, expected: str | None, source: str) -> None: + """Every image this tool produces reports whether it could be checked. + + An unverified image is the failure mode that bricks a radio quietly, so the + warning is never suppressed. + """ + md5 = hashlib.md5(data).hexdigest() + if not expected: + _out(f" md5 {md5} [!] unverified (no reference md5 available)") + elif md5 == expected: + _out(f" md5 {md5} ok, matches {source}") + else: + raise RuntimeError( + f"md5 mismatch against {source}: expected {expected}, got {md5}" + ) + + +def _print_update_info(info: UpdateInfo) -> None: + def show(label: str, entry: FirmwareInfo, md5_covers: str) -> None: + # The server populates version for the patch but not for the base image. + _out(f" {label} v{entry.version}" if entry.version else f" {label}") + _out(f" url {entry.url}") + if entry.md5: + # Neither md5 describes the file at the url above, which is easy to + # assume and wrong. + _out(f" md5 {entry.md5} ({md5_covers})") + + show("patch", info.firmware, "of the assembled image") + show("base", info.base, "of the extracted .bin") + + +def _write(path: str, data: bytes, force: bool) -> None: + if os.path.exists(path) and not force: + raise RuntimeError(f"{path} already exists (use --force to overwrite)") + with open(path, "wb") as f: + f.write(data) + print(path) + + +@contextlib.asynccontextmanager +async def _graceful_interrupt() -> t.AsyncGenerator[None, None]: + """Turn Ctrl+C into a cancellation of the task doing the work. + + The default handler raises `KeyboardInterrupt` wherever the main thread + happens to be, which during a transfer is almost always inside the event + loop rather than inside the coroutine. `flash` then never sees it, and never + gets to tell the radio to abort. Cancelling the task delivers the interrupt + where the cleanup lives. + + Restoring the default handler on the way in means a second Ctrl+C quits + outright rather than waiting for the abort to be sent. + """ + loop = asyncio.get_running_loop() + task = asyncio.current_task() + + def interrupt() -> None: + _out() + _out("Interrupted. Telling the radio to abort " + "(press Ctrl+C again to quit without waiting)...") + loop.remove_signal_handler(signal.SIGINT) + if task is not None: + task.cancel() + + try: + loop.add_signal_handler(signal.SIGINT, interrupt) + except NotImplementedError: # not available on Windows + yield + return + + try: + yield + finally: + with contextlib.suppress(ValueError, RuntimeError): + loop.remove_signal_handler(signal.SIGINT) + + +def _confirm(question: str, default_yes: bool, assume_yes: bool) -> bool: + if assume_yes: + return True + suffix = "[Y/n]" if default_yes else "[y/N]" + try: + answer = input(f"{question} {suffix} ").strip().lower() + except EOFError: + return False + return default_yes if not answer else answer.startswith("y") + + +##################### +# Radio + +@contextlib.asynccontextmanager +async def _radio( + args: argparse.Namespace, +) -> t.AsyncGenerator[CommandConnection, None]: + """Connect, and tolerate the radio vanishing on the way out. + + A firmware update ends with the radio rebooting, which drops the link before + anything gets to close it. Raising from the teardown would turn a completed + transfer into a crash. + """ + conn = _connection(args) + await conn.connect() + try: + yield conn + finally: + with contextlib.suppress(Exception): + await conn.disconnect() + + +def _connection(args: argparse.Namespace) -> CommandConnection: + # Imported lazily: everything except the radio commands works without a + # Bluetooth stack. + from ..command import CommandConnection + + if args.rfcomm is not None: + channel = "auto" if args.rfcomm == "auto" else int(args.rfcomm) + _out(f"Connecting over RFCOMM to {args.uuid} (channel {channel})...") + return CommandConnection.new_rfcomm(args.uuid, channel) + + _out(f"Connecting over BLE to {args.uuid}...") + return CommandConnection.new_ble(args.uuid) + + +def _print_device_info(info: DeviceInfo) -> None: + _out(f" vendor {info.vendor_id}, product {info.product_id}") + _out(f" firmware v{info.firmware_version}" + f", hardware version {info.hardware_version}") + + +##################### +# Products + +def _resolve_product( + args: argparse.Namespace, +) -> t.Tuple[int | None, str | None]: + """Resolve `--product` into a product id and patch name, letting the explicit + `--product-id` / `--patch-name` flags override either half.""" + product_id, patch_name = None, None + + if getattr(args, "product", None): + product_id, patch_name = PRODUCTS[args.product] + + if getattr(args, "product_id", None) is not None: + product_id = args.product_id + if getattr(args, "patch_name", None) is not None: + patch_name = args.patch_name + + return product_id, patch_name + + +def _require_product_id(product_id: int | None) -> int: + if product_id is None: + raise RuntimeError( + "a product is required: pass --product " + f"({', '.join(PRODUCTS)}) or --product-id" + ) + return product_id + + +##################### +# Commands + +async def _cmd_info(args: argparse.Namespace) -> int: + async with _connection(args) as conn: + _print_device_info(await conn.get_device_info()) + return 0 + + +async def _cmd_check(args: argparse.Namespace) -> int: + product_id, _ = _resolve_product(args) + info = await check_update(_require_product_id(product_id), + args.firmware_version) + if info is None: + _out("no update available") + return 2 + _print_update_info(info) + return 0 + + +async def _cmd_fetch(args: argparse.Namespace) -> int: + product_id, _ = _resolve_product(args) + info = await check_update(_require_product_id(product_id), + args.firmware_version) + if info is None: + _out("no update available") + return 2 + + _print_update_info(info) + + bundle = await download_firmware(info, _make_progress()) + _out() + _print_verdict(bundle.data, info.firmware.md5, "the update server") + + _write(args.output, bundle.data, args.force) + return 0 + + +async def _cmd_download_patch(args: argparse.Namespace) -> int: + _, patch_name = _resolve_product(args) + assert patch_name is not None # the parser requires --product or --patch-name + url = oss_patch_url(args.version, patch_name) + _out(f" url {url}") + + data = await download(url, "patch", _make_progress()) + _out() + _print_verdict(data, None, "") + + _write(args.output, data, args.force) + return 0 + + +async def _cmd_download_base(args: argparse.Namespace) -> int: + url = oss_base_url(args.version) + _out(f" url {url}") + + data = await download(url, "base", _make_progress()) + _out() + + extracted = extract_base(data) + _out(f" extracted md5 {hashlib.md5(extracted).hexdigest()}") + + _write(args.output, extracted if args.extract else data, args.force) + return 0 + + +async def _cmd_assemble(args: argparse.Namespace) -> int: + with open(args.base, "rb") as f: + base = f.read() + with open(args.patch, "rb") as f: + patch = f.read() + + data = assemble(base, patch) + _print_verdict(data, args.expect_md5, "--expect-md5") + + _write(args.output, data, args.force) + return 0 + + +async def _cmd_update(args: argparse.Namespace) -> int: + # The connection is held for the whole flow: the radio is needed at the start + # to identify it, and again at the end to flash. + async with _radio(args) as conn: + device_info = await conn.get_device_info() + _print_device_info(device_info) + + _out() + _out("Checking for updates...") + info = await check_update(device_info.product_id) + if info is None: + _out(" no update available") + return 2 + + installed = device_info.firmware_version + latest = info.firmware.version + _out(f" latest v{latest} (you have v{installed})") + _print_update_info(info) + + _out() + if latest == installed: + question = f"Already on v{latest}. Download and assemble anyway?" + if not _confirm(question, False, args.yes): + return 0 + elif not _confirm("Download and assemble?", True, args.yes): + return 0 + + bundle = await download_firmware(info, _make_progress()) + _out() + _out(f" assembled {bundle.size} bytes") + _print_verdict(bundle.data, info.firmware.md5, "the update server") + + directory = args.keep or tempfile.mkdtemp(prefix="benlink-fw-") + os.makedirs(directory, exist_ok=True) + path = os.path.join(directory, f"firmware-v{latest}.bin") + _out() + _write(path, bundle.data, args.force) + + _out() + if not _confirm(f"Flash v{latest} to this radio?", False, args.yes): + _out(f"The assembled image has been kept at {path}") + return 0 + + _out("Do not power off the radio until this finishes. " + "(Press Ctrl+C to abort safely.)") + try: + async with _graceful_interrupt(): + result = await flash(conn, bundle.data, _make_progress()) + except asyncio.CancelledError: + _out("Cancelled. Re-run to start over, as the transfer " + "does not resume.") + _out(f"The assembled image has been kept at {path}") + return 130 + except Exception as e: + _out() + _out(f"error: {e}") + _out(f"The assembled image has been kept at {path}") + return 1 + _out() + + if result == "COMPLETE": + _out("Firmware update complete.") + return 0 + + _out(" image staged, radio is rebooting") + return await _commit_after_reboot(args, bundle.data, path) + + +async def _commit_after_reboot( + args: argparse.Namespace, image: bytes, path: str +) -> int: + """Reconnect to the rebooted radio and finish the update. + + The radio drops the connection when it reboots and comes back needing only + the commit handshake. It stays in that state until it gets one, so a failed + attempt can simply be retried. + """ + for attempt in range(1, _COMMIT_ATTEMPTS + 1): + await asyncio.sleep(_REBOOT_WAIT) + try: + async with _radio(args) as conn: + if await flash(conn, image) == "COMPLETE": + _out() + _out("Firmware update complete.") + return 0 + except Exception as e: + _out(f" attempt {attempt}/{_COMMIT_ATTEMPTS} failed: {e}") + + _out() + _out("error: the radio did not come back to finish the update.") + _out("The image is already staged, so re-running `update` will resume " + f"from here. The assembled image has been kept at {path}") + return 1 + + +async def _cmd_abort(args: argparse.Namespace) -> int: + _out("This discards whatever update the radio has in progress.") + if not _confirm("Abort it?", False, args.yes): + return 0 + + async with _radio(args) as conn: + _print_device_info(await conn.get_device_info()) + await abort_update(conn) + + _out() + _out("Aborted. The radio is still running its current firmware.") + return 0 + + +async def _cmd_flash(args: argparse.Namespace) -> int: + with open(args.image, "rb") as f: + image = f.read() + + _out(f" image {args.image} ({len(image)} bytes)") + _print_verdict(image, args.expect_md5, "--expect-md5") + + async with _radio(args) as conn: + _print_device_info(await conn.get_device_info()) + + _out() + question = f"Flash {os.path.basename(args.image)} to this radio?" + if not _confirm(question, False, args.yes): + return 0 + + _out("Do not power off the radio until this finishes. " + "(Press Ctrl+C to abort safely.)") + try: + async with _graceful_interrupt(): + result = await flash(conn, image, _make_progress()) + except asyncio.CancelledError: + _out("Cancelled. Re-run to start over, as the transfer " + "does not resume.") + return 130 + except Exception as e: + _out() + _out(f"error: {e}") + return 1 + _out() + + if result == "COMPLETE": + _out("Firmware update complete.") + return 0 + + _out(" image staged, radio is rebooting") + return await _commit_after_reboot(args, image, args.image) + + +##################### +# Parser + +def _add_radio_args(parser: argparse.ArgumentParser) -> None: + parser.add_argument("uuid", help="radio device UUID, e.g. XX:XX:XX:XX:XX:XX") + parser.add_argument("--rfcomm", nargs="?", const="auto", default=None, + metavar="CHANNEL", + help="connect over RFCOMM instead of BLE") + + +def _add_product_args(parser: argparse.ArgumentParser) -> None: + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument("--product", choices=sorted(PRODUCTS)) + group.add_argument("--product-id", type=int, + help="from `info`, for radios not listed above") + + +def _add_output_args(parser: argparse.ArgumentParser) -> None: + parser.add_argument("-o", "--output", required=True) + parser.add_argument("--force", action="store_true", + help="overwrite an existing output file") + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="python -m benlink.firmware", + description="Download and assemble Benshi radio firmware.", + ) + subparsers = parser.add_subparsers(dest="command", required=True) + + update = subparsers.add_parser( + "update", help="guided upgrade: read the radio, fetch, assemble") + _add_radio_args(update) + update.add_argument("--yes", "-y", action="store_true", + help="accept all prompts") + update.add_argument("--keep", metavar="DIR", + help="write to DIR instead of a temporary directory") + update.add_argument("--force", action="store_true") + update.set_defaults(run=_cmd_update) + + flash_cmd = subparsers.add_parser( + "flash", help="flash an already-assembled image to a radio") + _add_radio_args(flash_cmd) + flash_cmd.add_argument("--image", required=True, + help="assembled firmware image to flash") + flash_cmd.add_argument("--expect-md5", metavar="MD5", + help="verify the image against a known md5 first") + flash_cmd.add_argument("--yes", "-y", action="store_true", + help="accept all prompts") + flash_cmd.set_defaults(run=_cmd_flash) + + abort_cmd = subparsers.add_parser( + "abort", help="discard an update the radio is partway through") + _add_radio_args(abort_cmd) + abort_cmd.add_argument("--yes", "-y", action="store_true", + help="accept all prompts") + abort_cmd.set_defaults(run=_cmd_abort) + + info = subparsers.add_parser( + "info", help="read product id and versions from a radio") + _add_radio_args(info) + info.set_defaults(run=_cmd_info) + + check = subparsers.add_parser( + "check", help="ask the update server for the latest release") + _add_product_args(check) + check.add_argument("--firmware-version", type=int, default=0, + help=argparse.SUPPRESS) + check.set_defaults(run=_cmd_check) + + fetch = subparsers.add_parser( + "fetch", help="check, download and assemble the latest release") + _add_product_args(fetch) + fetch.add_argument("--firmware-version", type=int, default=0, + help=argparse.SUPPRESS) + _add_output_args(fetch) + fetch.set_defaults(run=_cmd_fetch) + + patch = subparsers.add_parser( + "download-patch", help="download one patch by version") + patch.add_argument("--version", type=int, required=True) + patch_product = patch.add_mutually_exclusive_group(required=True) + patch_product.add_argument("--product", choices=sorted(PRODUCTS)) + patch_product.add_argument("--patch-name", + help="e.g. patch_base_to_vr_n76") + _add_output_args(patch) + patch.set_defaults(run=_cmd_download_patch) + + base = subparsers.add_parser( + "download-base", help="download a base image") + # No default: a patch only applies to the base it shipped with, and picking + # the wrong one corrupts the result silently. + base.add_argument("--version", choices=sorted(BASE_IMAGES), required=True, + help="which base image; patches v120-v128 use 'original', " + "v147 uses '1'") + base.add_argument("--extract", action="store_true", + help="unwrap the zip and write the .bin") + _add_output_args(base) + base.set_defaults(run=_cmd_download_base) + + assemble_cmd = subparsers.add_parser( + "assemble", help="combine a base and a patch (offline)") + assemble_cmd.add_argument("--base", required=True, + help="base image, raw or zipped") + assemble_cmd.add_argument("--patch", required=True) + assemble_cmd.add_argument("--expect-md5", metavar="MD5", + help="verify the assembled image against a known md5") + _add_output_args(assemble_cmd) + assemble_cmd.set_defaults(run=_cmd_assemble) + + return parser + + +if __name__ == "__main__": + args = _parser().parse_args() + try: + sys.exit(asyncio.run(args.run(args))) + except KeyboardInterrupt: + sys.exit(130) + except (RuntimeError, ImportError, OSError) as e: + print(f"error: {e}", file=sys.stderr) + sys.exit(1) diff --git a/src/benlink/firmware/_fetch.py b/src/benlink/firmware/_fetch.py new file mode 100644 index 0000000..1ebfee4 --- /dev/null +++ b/src/benlink/firmware/_fetch.py @@ -0,0 +1,313 @@ +"""Finding, downloading and assembling firmware images. + +See `benlink.firmware` for an overview and for the command line interface. +""" + +from __future__ import annotations +import typing as t +import asyncio +import hashlib +import io +import urllib.request +import zipfile + +from ..common import ImmutableBaseModel +from . import _rpc + +OSS_BASE_URL = "https://pubdatas.oss-cn-shenzhen.aliyuncs.com" +"""@private""" + +RPC_HOST = "rpc.benshikj.com:800" +"""@private""" + +RPC_TIMEOUT = 10.0 +"""@private""" + +PRODUCTS: t.Dict[str, t.Tuple[int, str]] = { + "VR_N76": (259, "patch_base_to_vr_n76"), + "GA_5WB": (259, "patch_base_to_vr_n76"), + "UV_PRO": (260, "patch_base_to_vr_n76_m"), + "VR_N75": (261, "patch_base_to_vr_n75_h2"), +} +"""Known radios, as `name: (product_id, patch_name)`. + +Every patch name here was returned by the update server for the corresponding product +id. Note that 259 covers both the VR-N76 and the GA-5WB, which share a patch series, +confirmed by a GA-5WB flash capture whose `md5sum_tail` matches +`patch_base_to_vr_n76.v120` assembled against the shared base. They cannot differ: +`CheckFirmwareUpdateRequest` carries no vendor id, so the server cannot tell the +two apart. +""" + +BASE_IMAGES: t.Dict[str, str] = { + "original": "upgrade_base.bin.zip", + "1": "upgrade_base_v1.bin.zip", +} +"""The base images a patch can be built against, as `name: filename`. + +A patch carries no checksum of its source, so pairing it with the wrong base produces +a corrupt image with no error (see `assemble`). Known pairings, from flash captures and +from the update server: patch v120, v121 and v128 use `original`; v147 uses `1`. Where +the changeover happened is not known, because the server only publishes metadata for +the current release. +""" + + +def _identity(data: bytes) -> bytes: + """@private (the RPC messages are encoded by hand; see `_rpc`)""" + return data + + +def _require(module: str, package: str): + try: + return __import__(module) + except ImportError: + raise ImportError( + f"{package} is required for this operation. " + f"Install with: pip install benlink[firmware]" + ) + + +##################### +# Data + +class FirmwareInfo(ImmutableBaseModel): + """One downloadable artifact (either the patch or the base image).""" + version: int + url: str + md5: str | None + """md5 of the *assembled* image for a patch, or of the *extracted* base image. + + `None` when no reference md5 is available, which is the case for every release + but the current one.""" + + @classmethod + def from_protocol(cls, info: _rpc.FirmwareInfo) -> FirmwareInfo: + """@private (Protocol helper)""" + return cls(version=info.version, url=info.url, md5=info.md5 or None) + + +class UpdateInfo(ImmutableBaseModel): + """The patch and base image that together make up a firmware release.""" + firmware: FirmwareInfo + base: FirmwareInfo + + @classmethod + def from_protocol( + cls, result: _rpc.CheckFirmwareUpdateResult + ) -> UpdateInfo | None: + """@private (Protocol helper)""" + if not result.firmware.url or not result.base.url: + return None + return cls( + firmware=FirmwareInfo.from_protocol(result.firmware), + base=FirmwareInfo.from_protocol(result.base), + ) + + +class FirmwareBundle(ImmutableBaseModel): + """An assembled, ready-to-flash firmware image.""" + data: bytes + update_info: UpdateInfo + + @property + def md5(self) -> str: + return hashlib.md5(self.data).hexdigest() + + @property + def md5_tail(self) -> bytes: + """Last 4 bytes of the md5 digest, as sent in `UPDATE_SYNC_REQ`.""" + return bytes.fromhex(self.md5)[-4:] + + @property + def size(self) -> int: + return len(self.data) + + def save(self, path: str) -> None: + with open(path, "wb") as f: + f.write(self.data) + + +ProgressCallback = t.Callable[[str, int, int], None] +"""`progress(label, bytes_done, bytes_total)`. `bytes_total` is 0 if unknown.""" + + +##################### +# Finding an update + +async def check_update( + product_id: int, + firmware_version: int = 0, +) -> UpdateInfo | None: + """Ask the update server for the latest release for `product_id`. + + `firmware_version` is the currently installed internal version. The server + returns the latest release regardless of its value, so it has no effect in + practice. Returns `None` if the server reports no update. + + Requires `grpcio`. + """ + grpc = _require("grpc", "grpcio") + + request = _rpc.encode_check_request(product_id, firmware_version) + + credentials = grpc.ssl_channel_credentials() + async with grpc.aio.secure_channel(RPC_HOST, credentials) as channel: + call = channel.unary_unary( + _rpc.METHOD, + request_serializer=_identity, + response_deserializer=_identity, + ) + try: + response: bytes = await call(request, timeout=RPC_TIMEOUT) + except grpc.aio.AioRpcError as e: + raise RuntimeError( + f"update check failed: {e.code()} {e.details()}") + + return UpdateInfo.from_protocol(_rpc.decode_check_result(response)) + + +def oss_patch_url(version: int, patch_name: str) -> str: + """URL of a patch in the object store.""" + return f"{OSS_BASE_URL}/firmware/v{version}/{patch_name}.bin" + + +def oss_base_url(base_image: str) -> str: + """URL of a base image in the object store. `base_image` is a key of + `BASE_IMAGES`.""" + if base_image not in BASE_IMAGES: + raise RuntimeError( + f"unknown base image {base_image!r}, expected one of " + f"{', '.join(BASE_IMAGES)}" + ) + return f"{OSS_BASE_URL}/{BASE_IMAGES[base_image]}" + + +def oss_update_info( + version: int, + patch_name: str, + base_image: str, +) -> UpdateInfo: + """Construct object-store URLs for a known version, without contacting the + update server. + + No md5s are available this way, so the result cannot be verified. Since a patch + only applies to the base it shipped with, picking the wrong `base_image` yields a + corrupt image silently. See `BASE_IMAGES`. + """ + return UpdateInfo( + firmware=FirmwareInfo( + version=version, + url=oss_patch_url(version, patch_name), + md5=None, + ), + base=FirmwareInfo(version=0, url=oss_base_url(base_image), md5=None), + ) + + +##################### +# Downloading and assembling + +async def download( + url: str, + label: str = "download", + progress: ProgressCallback | None = None, +) -> bytes: + """Download a single artifact.""" + return await asyncio.to_thread(_download, url, label, progress) + + +def _download(url: str, label: str, progress: ProgressCallback | None) -> bytes: + with urllib.request.urlopen(url) as response: + total = int(response.headers.get("Content-Length", 0)) + chunks: t.List[bytes] = [] + received = 0 + while chunk := response.read(65536): + chunks.append(chunk) + received += len(chunk) + if progress: + progress(label, received, total) + return b"".join(chunks) + + +def _verify(data: bytes, expected_md5: str | None, label: str) -> None: + if not expected_md5: + return + actual = hashlib.md5(data).hexdigest() + if actual != expected_md5: + raise RuntimeError( + f"{label} md5 mismatch: expected {expected_md5}, got {actual}" + ) + + +def extract_base(base: bytes) -> bytes: + """Return the base image, unwrapping the zip it ships in if needed.""" + if base[:2] != b"PK": + return base + + with zipfile.ZipFile(io.BytesIO(base)) as zf: + names = [n for n in zf.namelist() if n.endswith(".bin")] + if not names: + raise RuntimeError("no .bin found in base zip") + return zf.read(names[0]) + + +def assemble(base: bytes, patch: bytes) -> bytes: + """Apply a BSDIFF40 patch to a base image. + + `base` may be either the raw base image or the zip it ships in. + + A patch carries no checksum of the base it was built against, so applying it to + the wrong base succeeds and silently yields a corrupt image. Patches are only + valid against the base image released alongside them. Compare the result against + `UpdateInfo.firmware.md5` whenever it is known. + """ + bsdiff4 = _require("bsdiff4", "bsdiff4") + + if patch[:8] != b"BSDIFF40": + raise RuntimeError( + f"unexpected patch magic {patch[:8]!r}, expected b'BSDIFF40'" + ) + + return bsdiff4.patch(extract_base(base), patch) + + +async def download_firmware( + update_info: UpdateInfo, + progress: ProgressCallback | None = None, +) -> FirmwareBundle: + """Download the patch and base image named by `update_info` and assemble them. + + Both are always fetched fresh. Base images are revised over time and a patch + only applies to the one released with it, so reusing a local copy risks pairing + a patch with a base it was never built against. + + Requires `bsdiff4`. + """ + patch, base = await asyncio.gather( + asyncio.to_thread(_download, update_info.firmware.url, + "patch", progress), + asyncio.to_thread(_download, update_info.base.url, "base", progress), + ) + + # The server's md5s describe the extracted base and the assembled firmware, + # neither the base zip nor the patch file as downloaded. + base = extract_base(base) + _verify(base, update_info.base.md5, "base image") + + data = await asyncio.to_thread(assemble, base, patch) + _verify(data, update_info.firmware.md5, "assembled firmware") + + return FirmwareBundle(data=data, update_info=update_info) + + +async def fetch_firmware( + product_id: int, + firmware_version: int = 0, + progress: ProgressCallback | None = None, +) -> FirmwareBundle | None: + """Check for an update and download it if one is available.""" + update_info = await check_update(product_id, firmware_version) + if update_info is None: + return None + return await download_firmware(update_info, progress) diff --git a/src/benlink/firmware/_flash.py b/src/benlink/firmware/_flash.py new file mode 100644 index 0000000..e2b7253 --- /dev/null +++ b/src/benlink/firmware/_flash.py @@ -0,0 +1,501 @@ +"""Delivering an assembled firmware image to a radio. + +**This can break your radio, and nothing in this library can undo it. +Use this at your own risk. I am not responsible for bricking your radio.** + +The transfer runs over the same command connection as everything else, in two +phases separated by a reboot: + + VM_CONNECT + REGISTER_BT_NOTIFICATION (VMU_PACKET) + UPDATE_SYNC_REQ -> UPDATE_SYNC_CFM (DATA_TRANSFER) + UPDATE_START_REQ -> UPDATE_START_CFM + UPDATE_START_DATA_REQ + UPDATE_DATA <- UPDATE_DATA_BYTES_REQ (repeats) + UPDATE_IS_VALIDATION_DONE_REQ + -> UPDATE_TRANSFER_COMPLETE_IND + UPDATE_TRANSFER_COMPLETE_RES + + [radio reboots, connection drops, reconnect] + + VM_CONNECT + REGISTER_BT_NOTIFICATION (VMU_PACKET) + UPDATE_SYNC_REQ -> UPDATE_SYNC_CFM (IN_PROGRESS) + UPDATE_START_REQ -> UPDATE_START_CFM + UPDATE_IN_PROGRESS_RES -> UPDATE_COMPLETE_IND + CANCEL_BT_NOTIFICATION (VMU_PACKET) + VM_DISCONNECT + +`flash` runs one phase per call and reports whether a reboot is pending, so the +caller owns the reconnect. Which phase runs is decided by the `UpdateState` the +radio reports in `UPDATE_SYNC_CFM` rather than tracked locally, so an update +interrupted after the image was staged resumes at the right phase rather than +sending it all again. An interrupted *transfer* is not resumable: the radio +starts it over from the beginning. +""" + +from __future__ import annotations +import asyncio +import hashlib +import typing as t + +from .. import protocol as p +from ..protocol.command.bt_notification import ( + BtEventNotificationBody, BtEventType, +) +from ..protocol.command.vm import ( + UpdateState, + VmConnectBody, + VmConnectReplyBody, + VmControlBody, + VmControlReplyBody, + VmControlMessage, + VmControlType, + VmControlUpdateAbortReq, + VmControlUpdateData, + VmControlUpdateDataBytesReq, + VmControlUpdateDataStartReq, + VmControlUpdateError, + VmControlUpdateInProgressRes, + VmControlUpdateIsValidationDoneReq, + VmControlUpdateStartReq, + VmControlUpdateSyncCfm, + VmControlUpdateSyncReq, + VmControlUpdateTransferCompleteRes, + VmDisconnectBody, + VmuPacket, + VmuPacketMessage, + VmuPacketType, +) +from ._fetch import ProgressCallback + +if t.TYPE_CHECKING: + from ..command import ( + CommandConnection, RadioMessage, UnknownProtocolMessage, + ) + +_REPLY_TIMEOUT = 15.0 +_CHUNK_TIMEOUT = 60.0 +_VALIDATION_TIMEOUT = 180.0 +_COMPLETE_TIMEOUT = 180.0 +_ABORT_TIMEOUT = 5.0 + +# The radio reboots on its own once it accepts UPDATE_TRANSFER_COMPLETE_RES, so +# this byte is not "did the transfer succeed" despite the field name: 0 proceeds +# with the reboot, 1 postpones it. All successful updates in my btsnoop +# captures send 0; the one sending 1 is the app's "cancel the restart" button, +# after which the radio sits in TRANSFER_COMPLETE until a later session sends 0. +_REBOOT_NOW = False + + +FlashResult = t.Literal["REBOOT_PENDING", "COMPLETE"] +"""What `flash` left the radio doing. + +`REBOOT_PENDING`: the image is staged, and the radio is rebooting. The connection +will drop; reconnect and call `flash` again to finish. + +`COMPLETE`: the update is committed and running. +""" + + +async def flash( + conn: CommandConnection, + image: bytes, + progress: ProgressCallback | None = None, +) -> FlashResult: + """Deliver an assembled firmware image to a connected radio. + + **This can break your radio, and nothing in this library can undo it. + Use this at your own risk. I am not responsible for bricking your radio.** + + Runs whichever phase of the update the radio says it is in, so a full update + is two calls with a reconnect in between: + + if await flash(conn, image) == "REBOOT_PENDING": + ... reconnect ... + await flash(conn, image) + + Raises `FlashError` if the radio is partway through a *different* image, so + that finishing an update never commits something the caller didn't pass. + """ + async with conn.subscribe(_is_vm_message) as inbox: + await _vm_connect(conn, inbox) + + tail = _md5_tail(image) + cfm = await _sync(conn, inbox, tail) + if cfm.md5sum_tail != tail: + # UpdateError.SYNC_IS_DIFFERENT suggests the radio rejects this + # itself, but no capture shows it doing so, and committing the wrong + # image is not something to find out about the hard way. + raise FlashError( + f"radio is partway through a different image: it reports " + f"md5 ...{cfm.md5sum_tail.hex()}, this one is ...{tail.hex()}. " + f"Flash the matching image to finish that update, or call " + f"abort_update to discard it" + ) + + state = cfm.update_state + await _start(conn, inbox) + + match state: + case UpdateState.DATA_TRANSFER | UpdateState.VALIDATION: + # Only the transfer is abortable. Once an image is staged the + # radio owns it, and UPDATE_ABORT_REQ would throw it away. + try: + if state is UpdateState.DATA_TRANSFER: + await _transfer(conn, inbox, image, progress) + # VALIDATION appears in no capture. The image is already + # delivered in that state, so ask whether the checksum + # finished rather than sending all of it again. + await _validate(conn, inbox) + except (Exception, KeyboardInterrupt, asyncio.CancelledError): + # Ctrl+C and cancellation are not `Exception`, but a radio + # left mid-transfer still deserves to be told. + await _abort_transfer(conn, inbox) + raise + await _request_reboot(conn) + return "REBOOT_PENDING" + + case UpdateState.TRANSFER_COMPLETE: + await _request_reboot(conn) + return "REBOOT_PENDING" + + case UpdateState.IN_PROGRESS: + await _finalize(conn, inbox) + return "COMPLETE" + + case UpdateState.COMMIT: + # UPDATE_COMMIT_CFM exists, but no capture shows the app in this + # state or sending it, so there is nothing to copy. Stopping + # leaves the staged image intact for the app to finish. + raise FlashError( + "radio reports the COMMIT state, which benlink has never " + "observed and does not know how to answer" + ) + + +async def abort_update(conn: CommandConnection) -> None: + """Discard whatever update the radio is partway through. + + For an update left behind by a flash that did not exit cleanly, or whose + image is no longer to hand: `flash` will not finish an update for an image it + wasn't given, so without this the radio stays stuck partway. + """ + async with conn.subscribe(_is_vm_message) as inbox: + await _vm_connect(conn, inbox) + await _send_control( + conn, VmControlType.UPDATE_ABORT_REQ, VmControlUpdateAbortReq() + ) + await _recv_vmu(inbox, VmuPacketType.UPDATE_ABORT_CFM) + await _vm_disconnect(conn) + + +##################### +# Phases + +async def _transfer( + conn: CommandConnection, + inbox: asyncio.Queue[RadioMessage], + image: bytes, + progress: ProgressCallback | None, +) -> None: + """Send the image, one device-requested chunk at a time.""" + await _send_control( + conn, VmControlType.UPDATE_START_DATA_REQ, VmControlUpdateDataStartReq() + ) + + total = len(image) + offset = 0 + + while offset < total: + req = await _recv_vmu_as( + inbox, + VmuPacketType.UPDATE_DATA_BYTES_REQ, + VmControlUpdateDataBytesReq, + timeout=_CHUNK_TIMEOUT, + ) + + # Always 0 in the captures, including across aborted transfers, which + # the radio restarts rather than resumes. Honoured in case some model + # does ask, but the relative reading is a guess with nothing to check + # it against. + offset += req.n_bytes_skip + + chunk = image[offset:offset + req.n_bytes_requested] + if not chunk: + raise FlashError( + f"radio asked for {req.n_bytes_requested} bytes at offset " + f"{offset}, past the end of a {total} byte image" + ) + + offset += len(chunk) + + await _send_control( + conn, + VmControlType.UPDATE_DATA, + VmControlUpdateData( + is_final_fragment=offset >= total, + data=chunk, + ), + ) + + if progress is not None: + progress("flash", offset, total) + + +async def _validate( + conn: CommandConnection, inbox: asyncio.Queue[RadioMessage] +) -> None: + """Wait for the radio to checksum what it received.""" + await _send_control( + conn, + VmControlType.UPDATE_IS_VALIDATION_DONE_REQ, + VmControlUpdateIsValidationDoneReq(), + ) + await _recv_vmu( + inbox, + VmuPacketType.UPDATE_TRANSFER_COMPLETE_IND, + timeout=_VALIDATION_TIMEOUT, + ) + + +async def _request_reboot(conn: CommandConnection) -> None: + """Release the staged image, which the radio reboots into on its own. + + A radio that ignores this stays in `TRANSFER_COMPLETE`, and that state + dispatches straight back here, so calling `flash` again re-sends the same + message rather than making progress. + + Confirmed on the UV-Pro (260) and the GA-5WB (259), which takes the same + image as the VR-N76. Other models are untested. + """ + await _send_control( + conn, + VmControlType.UPDATE_TRANSFER_COMPLETE_RES, + VmControlUpdateTransferCompleteRes(is_complete=_REBOOT_NOW), + ) + + +async def _finalize( + conn: CommandConnection, inbox: asyncio.Queue[RadioMessage] +) -> None: + """Commit the staged image on the rebooted radio.""" + await _send_control( + conn, VmControlType.UPDATE_IN_PROGRESS_RES, VmControlUpdateInProgressRes() + ) + await _recv_vmu( + inbox, VmuPacketType.UPDATE_COMPLETE_IND, timeout=_COMPLETE_TIMEOUT + ) + + await _vm_disconnect(conn) + + +async def _vm_connect( + conn: CommandConnection, inbox: asyncio.Queue[RadioMessage] +) -> None: + await conn.send_protocol_message( + _message(p.ExtendedCommand.VM_CONNECT, VmConnectBody()) + ) + reply = await _recv_connect_reply(inbox) + if reply.status != p.ReplyStatus.SUCCESS: + raise FlashError(f"VM_CONNECT rejected: {reply.status.name}") + + # Every reply worth having arrives as a BT_EVENT_NOTIFICATION, and the radio + # sends none until asked. Without this, VM_CONNECT succeeds and then every + # wait for a VMU packet times out. The app does not wait for the reply. + await conn.send_protocol_message(_message( + p.ExtendedCommand.REGISTER_BT_NOTIFICATION, + bytes([BtEventType.VMU_PACKET]), + )) + + +async def _vm_disconnect(conn: CommandConnection) -> None: + """The reply is not waited for: by this point the radio may be committing or + rebooting, and there is nothing left to do with the answer either way.""" + await conn.send_protocol_message(_message( + p.ExtendedCommand.CANCEL_BT_NOTIFICATION, + bytes([BtEventType.VMU_PACKET]), + )) + await conn.send_protocol_message( + _message(p.ExtendedCommand.VM_DISCONNECT, VmDisconnectBody()) + ) + + +async def _sync( + conn: CommandConnection, inbox: asyncio.Queue[RadioMessage], md5_tail: bytes +) -> VmControlUpdateSyncCfm: + await _send_control( + conn, + VmControlType.UPDATE_SYNC_REQ, + VmControlUpdateSyncReq(md5sum_tail=md5_tail), + ) + return await _recv_vmu_as( + inbox, VmuPacketType.UPDATE_SYNC_CFM, VmControlUpdateSyncCfm + ) + + +async def _start( + conn: CommandConnection, inbox: asyncio.Queue[RadioMessage] +) -> None: + await _send_control( + conn, VmControlType.UPDATE_START_REQ, VmControlUpdateStartReq() + ) + # UPDATE_START_CFM carries a cfm_code, but every capture reports OK in both + # phases, so nothing here can be keyed off it. + await _recv_vmu(inbox, VmuPacketType.UPDATE_START_CFM) + + +async def _abort_transfer( + conn: CommandConnection, inbox: asyncio.Queue[RadioMessage] +) -> None: + """Best effort: the original failure is what the caller needs to see. + + Waits for `UPDATE_ABORT_CFM` so the radio has actually processed the abort + before the caller drops the link, but on a short leash — this runs while the + caller is already unwinding, often from Ctrl+C, and must not look hung. + + Only `Exception` is swallowed, so a second Ctrl+C during the abort gets out + rather than being absorbed by the cleanup. + """ + try: + await _send_control( + conn, VmControlType.UPDATE_ABORT_REQ, VmControlUpdateAbortReq() + ) + await _recv_vmu( + inbox, VmuPacketType.UPDATE_ABORT_CFM, timeout=_ABORT_TIMEOUT + ) + except Exception: + pass + + +##################### +# Transport + +def _md5_tail(image: bytes) -> bytes: + """Last 4 bytes of the md5 digest, which is how UPDATE_SYNC_REQ names an + image.""" + return hashlib.md5(image).digest()[-4:] + + +class FlashError(RuntimeError): + """The radio rejected or abandoned the update.""" + + +def _message(command: p.ExtendedCommand, body: t.Any) -> p.Message: + return p.Message( + command_group=p.CommandGroup.EXTENDED, + is_reply=False, + command=command, + body=body, + ) + + +async def _send_control( + conn: CommandConnection, control_type: VmControlType, msg: VmControlMessage +) -> None: + await conn.send_protocol_message(_message( + p.ExtendedCommand.VM_CONTROL, + VmControlBody( + vm_control_type=control_type, + # Not msg.length(), which is None for the dynamically sized bodies. + n_bytes_payload=len(msg.to_bytes()), + msg=msg, + ), + )) + + +def _is_vm_message(msg: RadioMessage) -> bool: + from ..command import UnknownProtocolMessage + + if not isinstance(msg, UnknownProtocolMessage): + return False + body = msg.message.body + if isinstance(body, (VmConnectReplyBody, VmControlReplyBody)): + return True + return ( + isinstance(body, BtEventNotificationBody) + and body.bt_event_type == BtEventType.VMU_PACKET + ) + + +_T = t.TypeVar("_T") + + +async def _with_timeout( + receive: t.Coroutine[t.Any, t.Any, _T], timeout: float, described_as: str +) -> _T: + # asyncio.timeout would read better, but it is 3.11+ and this package + # supports 3.10. + try: + return await asyncio.wait_for(receive, timeout) + except asyncio.TimeoutError: + raise FlashError( + f"radio went quiet: no {described_as} within {timeout:g}s" + ) from None + + +async def _recv_body(inbox: asyncio.Queue[RadioMessage]) -> t.Any: + """`_is_vm_message` has already established that these are VM messages.""" + msg = t.cast("UnknownProtocolMessage", await inbox.get()) + return msg.message.body + + +async def _recv_connect_reply( + inbox: asyncio.Queue[RadioMessage], timeout: float = _REPLY_TIMEOUT +) -> VmConnectReplyBody: + async def receive() -> VmConnectReplyBody: + while True: + body = await _recv_body(inbox) + if isinstance(body, VmConnectReplyBody): + return body + + return await _with_timeout(receive(), timeout, "VM_CONNECT reply") + + +async def _recv_vmu( + inbox: asyncio.Queue[RadioMessage], + expect: VmuPacketType, + timeout: float = _REPLY_TIMEOUT, +) -> VmuPacketMessage | bytes: + """Wait for a VMU packet of `expect`. + + The `VM_CONTROL` reply that comes back first only acknowledges receipt of + the control message; the answer always follows separately as a VMU packet. + An `UPDATE_ERROR` is raised here rather than left to time out. + """ + async def receive() -> VmuPacketMessage | bytes: + while True: + body = await _recv_body(inbox) + if not isinstance(body, BtEventNotificationBody): + continue + + packet = body.bt_event + if not isinstance(packet, VmuPacket): + continue + + if isinstance(packet.msg, VmControlUpdateError): + raise FlashError( + f"radio reported {packet.msg.update_error.name} while " + f"waiting for {expect.name}" + ) + + if packet.vmu_packet_type == expect: + return packet.msg + + return await _with_timeout(receive(), timeout, expect.name) + + +_VmuT = t.TypeVar("_VmuT", bound=VmuPacketMessage) + + +async def _recv_vmu_as( + inbox: asyncio.Queue[RadioMessage], + expect: VmuPacketType, + as_type: t.Type[_VmuT], + timeout: float = _REPLY_TIMEOUT, +) -> _VmuT: + """`_recv_vmu` for the packets whose fields are actually read.""" + msg = await _recv_vmu(inbox, expect, timeout) + if not isinstance(msg, as_type): + raise FlashError(f"could not parse {expect.name}: {msg!r}") + return msg diff --git a/src/benlink/firmware/_rpc.py b/src/benlink/firmware/_rpc.py new file mode 100644 index 0000000..7221a1a --- /dev/null +++ b/src/benlink/firmware/_rpc.py @@ -0,0 +1,159 @@ +"""Wire format for the vendor's firmware update RPC. + +The update server speaks gRPC, but only one method matters and its messages are +small, so they are encoded by hand rather than through protoc. That keeps the +schema readable in source, avoids a protobuf runtime dependency, and avoids +checked-in generated code that stops working on a future protobuf major release. + +The `DeviceManagement` service has three methods (`CheckFirmwareUpdate`, +`GetRegTimes`, `SetRegTimes`); only the firmware check is modelled here. Field +numbers are the contract, and the names follow the vendor's. + + syntax = "proto3"; + + package benshikj; + + message CheckFirmwareUpdateRequest { + int32 product_id = 1; + int32 firmware_version = 2; + bool beta = 3; + int64 user_id = 4; + int32 invite_code = 5; + } + + message FirmwareInfo { + int32 version = 1; + string url = 2; + string md5 = 3; + string release_notes = 4; + string release_date = 5; + } + + message CheckFirmwareUpdateResult { + FirmwareInfo firmware = 1; + FirmwareInfo base = 2; + } + + service DeviceManagement { + rpc CheckFirmwareUpdate(CheckFirmwareUpdateRequest) + returns (CheckFirmwareUpdateResult); + } + +Note that `md5` does not describe the file at `url`: for the patch it is the md5 +of the *assembled* firmware, and for the base it is the md5 of the `.bin` inside +the zip. +""" + +from __future__ import annotations +import typing as t + +METHOD = "/benshikj.DeviceManagement/CheckFirmwareUpdate" + +WIRE_VARINT = 0 +"""@private""" + +WIRE_BYTES = 2 +"""@private""" + + +class FirmwareInfo(t.NamedTuple): + """A decoded `benshikj.FirmwareInfo`.""" + version: int = 0 + url: str = "" + md5: str = "" + + +class CheckFirmwareUpdateResult(t.NamedTuple): + """A decoded `benshikj.CheckFirmwareUpdateResult`.""" + firmware: FirmwareInfo = FirmwareInfo() + base: FirmwareInfo = FirmwareInfo() + + +def _encode_varint(value: int) -> bytes: + out = bytearray() + while value > 0x7F: + out.append((value & 0x7F) | 0x80) + value >>= 7 + out.append(value) + return bytes(out) + + +def _encode_varint_field(field: int, value: int) -> bytes: + return _encode_varint(field << 3 | WIRE_VARINT) + _encode_varint(value) + + +def _read_varint(data: bytes, pos: int) -> t.Tuple[int, int]: + value = shift = 0 + while pos < len(data): + byte = data[pos] + pos += 1 + value |= (byte & 0x7F) << shift + if not byte & 0x80: + break + shift += 7 + return value, pos + + +def _walk(data: bytes) -> t.Iterator[t.Tuple[int, int, int, bytes]]: + """Yield `(field_number, wire_type, varint_value, delimited_value)`. + + Only one of the two values is meaningful, according to the wire type. + Unrecognised wire types end the walk, since their length is unknown. + """ + pos = 0 + while pos < len(data): + tag, pos = _read_varint(data, pos) + field, wire = tag >> 3, tag & 0x7 + + if wire == WIRE_VARINT: + value, pos = _read_varint(data, pos) + yield field, wire, value, b"" + elif wire == WIRE_BYTES: + length, pos = _read_varint(data, pos) + yield field, wire, 0, data[pos:pos + length] + pos += length + elif wire == 5: + pos += 4 + elif wire == 1: + pos += 8 + else: + return + + +def encode_check_request(product_id: int, firmware_version: int = 0) -> bytes: + """Encode a `CheckFirmwareUpdateRequest`. + + proto3 omits zero-valued fields, so a request carrying only a product id asks + for the latest release. + """ + out = b"" + if product_id: + out += _encode_varint_field(1, product_id) + if firmware_version: + out += _encode_varint_field(2, firmware_version) + return out + + +def _decode_firmware_info(data: bytes) -> FirmwareInfo: + version, url, md5 = 0, "", "" + for field, wire, varint, delimited in _walk(data): + if field == 1 and wire == WIRE_VARINT: + version = varint + elif field == 2 and wire == WIRE_BYTES: + url = delimited.decode("utf-8", "replace") + elif field == 3 and wire == WIRE_BYTES: + md5 = delimited.decode("utf-8", "replace") + return FirmwareInfo(version=version, url=url, md5=md5) + + +def decode_check_result(data: bytes) -> CheckFirmwareUpdateResult: + """Decode a `CheckFirmwareUpdateResult`. Absent fields decode as empty.""" + firmware = base = FirmwareInfo() + for field, wire, _, delimited in _walk(data): + if wire != WIRE_BYTES: + continue + if field == 1: + firmware = _decode_firmware_info(delimited) + elif field == 2: + base = _decode_firmware_info(delimited) + return CheckFirmwareUpdateResult(firmware=firmware, base=base) diff --git a/src/benlink/protocol/command/bt_notification.py b/src/benlink/protocol/command/bt_notification.py new file mode 100644 index 0000000..52a1026 --- /dev/null +++ b/src/benlink/protocol/command/bt_notification.py @@ -0,0 +1,44 @@ +from __future__ import annotations +from .bitfield import Bitfield, bf_int_enum, bf_dyn, bf_bytes, bf_bitfield +from enum import IntEnum +from .vm import VmuPacket + +################################################# +# BT_EVENT_NOTIFICATION + + +class BtEventType(IntEnum): + START = 0 + RSSI_LOW_THRESHOLD = 1 + RSSI_HIGH_THRESHOLD = 2 + BATTERY_LOW_THRESHOLD = 3 + BATTERY_HIGH_THRESHOLD = 4 + DEVICE_STATE_CHANGED = 5 + PIO_CHANGED = 6 + DEBUG_MESSAGE = 7 + BATTERY_CHARGED = 8 + CHARGER_CONNECTION = 9 + CAPSENSE_UPDATE = 10 + USER_ACTION = 11 + SPEECH_RECOGNITION = 12 + AV_COMMAND = 13 + REMOTE_BATTERY_LEVEL = 14 + KEY = 15 + DFU_STATE = 16 + UART_RECEIVED_DATA = 17 + VMU_PACKET = 18 + + +def bt_event_disc(m: BtEventNotificationBody, n: int): + match m.bt_event_type: + case BtEventType.VMU_PACKET: + out = VmuPacket + case _: + return bf_bytes(n // 8) + + return bf_bitfield(out, n) + + +class BtEventNotificationBody(Bitfield): + bt_event_type: BtEventType = bf_int_enum(BtEventType, 8) + bt_event: VmuPacket | bytes = bf_dyn(bt_event_disc) diff --git a/src/benlink/protocol/command/dev_state_var.py b/src/benlink/protocol/command/dev_state_var.py deleted file mode 100644 index 5eed02d..0000000 --- a/src/benlink/protocol/command/dev_state_var.py +++ /dev/null @@ -1,28 +0,0 @@ -from enum import IntEnum - -################################################# -# GET_DEV_STATE_VAR - - -class DevStateVar(IntEnum): - START = 0 - RSSI_LOW_THRESHOLD = 1 - RSSI_HIGH_THRESHOLD = 2 - BATTERY_LOW_THRESHOLD = 3 - BATTERY_HIGH_THRESHOLD = 4 - DEVICE_STATE_CHANGED = 5 - PIO_CHANGED = 6 - DEBUG_MESSAGE = 7 - BATTERY_CHARGED = 8 - CHARGER_CONNECTION = 9 - CAPSENSE_UPDATE = 10 - USER_ACTION = 11 - SPEECH_RECOGNITION = 12 - AV_COMMAND = 13 - REMOTE_BATTERY_LEVEL = 14 - KEY = 15 - DFU_STATE = 16 - UART_RECEIVED_DATA = 17 - VMU_PACKET = 18 - -# TODO diff --git a/src/benlink/protocol/command/message.py b/src/benlink/protocol/command/message.py index 6616887..47481b8 100644 --- a/src/benlink/protocol/command/message.py +++ b/src/benlink/protocol/command/message.py @@ -29,6 +29,14 @@ ) from .phone_status import SetPhoneStatusBody, SetPhoneStatusReplyBody from .status import GetHtStatusBody, GetHtStatusReplyBody +from .vm import ( + VmControlBody, VmControlReplyBody, + VmConnectBody, VmConnectReplyBody, + VmDisconnectBody, VmDisconnectReplyBody, +) +from .bt_notification import ( + BtEventNotificationBody +) from .position import GetPositionBody, GetPositionReplyBody @@ -39,13 +47,13 @@ class CommandGroup(IntEnum): class ExtendedCommand(IntEnum): UNKNOWN = 0 + VM_CONNECT = 1600 + VM_DISCONNECT = 1601 + VM_CONTROL = 1602 GET_BT_SIGNAL = 769 - UNKNOWN_01 = 1600 - UNKNOWN_02 = 1601 - UNKNOWN_03 = 1602 - UNKNOWN_04 = 16385 - UNKNOWN_05 = 16386 - GET_DEV_STATE_VAR = 16387 + REGISTER_BT_NOTIFICATION = 16385 + CANCEL_BT_NOTIFICATION = 16386 + BT_EVENT_NOTIFICATION = 16387 DEV_REGISTRATION = 1825 @classmethod @@ -188,6 +196,18 @@ def body_disc(m: Message, n: int): return bf_bytes(n // 8) case CommandGroup.EXTENDED: match m.command: + case ExtendedCommand.VM_CONTROL: + out = VmControlReplyBody if m.is_reply else VmControlBody + case ExtendedCommand.VM_CONNECT: + out = VmConnectReplyBody if m.is_reply else VmConnectBody + case ExtendedCommand.VM_DISCONNECT: + out = VmDisconnectReplyBody if m.is_reply else VmDisconnectBody + case ExtendedCommand.BT_EVENT_NOTIFICATION: + if m.is_reply: + raise ValueError( + "BtEventNotification cannot be a reply" + ) + out = BtEventNotificationBody case _: return bf_bytes(n // 8) @@ -221,6 +241,13 @@ def body_disc(m: Message, n: int): SetPhoneStatusReplyBody, GetHtStatusBody, GetHtStatusReplyBody, + VmControlBody, + VmControlReplyBody, + VmConnectBody, + VmConnectReplyBody, + VmDisconnectBody, + VmDisconnectReplyBody, + BtEventNotificationBody, GetPositionReplyBody, GetPositionBody, ] diff --git a/src/benlink/protocol/command/phone_status.py b/src/benlink/protocol/command/phone_status.py index 3f771f5..d94a3f3 100644 --- a/src/benlink/protocol/command/phone_status.py +++ b/src/benlink/protocol/command/phone_status.py @@ -1,10 +1,10 @@ from __future__ import annotations -from .bitfield import Bitfield, bf_lit_int, bf_int_enum, bf_list, bf_bool +from .bitfield import Bitfield, bf_lit_int, bf_int_enum, bf_list, bf_bool, bf_dyn, bf_bytes import typing as t from .common import ReplyStatus -class SetPhoneStatusBody(Bitfield): +class PhoneStatus(Bitfield): is_channel_bonded_lower: t.List[bool] = bf_list(bf_bool(), 16) is_linked: bool _pad: t.Literal[0] = bf_lit_int(1, default=0) @@ -12,5 +12,18 @@ class SetPhoneStatusBody(Bitfield): _pad2: t.Literal[0] = bf_lit_int(14, default=0) +def phone_status_disc(_: SetPhoneStatusBody, n: int): + if n == PhoneStatus.length(): + return PhoneStatus + + # TODO: There's a 32 bit version of phone status that popped up in + # uv-pro 0.7.9-32 upgrade firmware. I'll need to see what it is... + return bf_bytes(n // 8) + + +class SetPhoneStatusBody(Bitfield): + phone_status: PhoneStatus | bytes = bf_dyn(phone_status_disc) + + class SetPhoneStatusReplyBody(Bitfield): reply_status: ReplyStatus = bf_int_enum(ReplyStatus, 8) diff --git a/src/benlink/protocol/command/vm.py b/src/benlink/protocol/command/vm.py new file mode 100644 index 0000000..58d44ea --- /dev/null +++ b/src/benlink/protocol/command/vm.py @@ -0,0 +1,296 @@ +from __future__ import annotations +import typing as t +from .bitfield import Bitfield, bf_int_enum, bf_int, bf_bytes, bf_dyn, bf_map, bf_bitfield, bf_lit_int +from .common import ReplyStatus +from enum import IntEnum + +##################################################################### +# Order of events in a firmware update: +# +# 1. VM_CONNECT +# 2. VM_CONTROL: +# a. UPDATE_SYNC_REQ (UPDATE_SYNC_CFM) (with last 4 bytes of firmware md5sum) +# b. UPDATE_START_REQ (UPDATE_START_CFM) +# c. UPDATE_DATA_START_REQ +# d. (UPDATE_DATA_BYTES_REQ) UPDATE_DATA (145 bytes at a time. repeat until all data is sent, except for the last fragment) +# e. UPDATE_DATA (final fragment with is_final_fragment=True) +# f. UPDATE_IS_VALIDATION_DONE_REQ (UPDATE_TRANSFER_COMPLETE_IND) +# g. UPDATE_TRANSFER_COMPLETE_RES (with is_complete=False; triggers reboot) +# +# Reboot happens here, and the connection drops. UPDATE_SYNC_CFM reports +# IN_PROGRESS afterwards, which is what tells the app to resume at step 3. +# +# 3. VM_CONNECT +# h. UPDATE_SYNC_REQ (UPDATE_SYNC_CFM) (with last 4 bytes of firmware md5sum) +# i. UPDATE_START_REQ (UPDATE_START_CFM) +# j. UPDATE_IN_PROGRESS_RES (UPDATE_COMPLETE_IND) +# 4. VM_DISCONNECT + +##################################################################### +# Order of events in an aborted firmware update: +# +# 1. VM_CONNECT +# 2. VM_CONTROL: +# a. UPDATE_SYNC_REQ (UPDATE_SYNC_CFM) +# b. UPDATE_START_REQ (UPDATE_START_CFM) +# c. UPDATE_DATA_START_REQ +# d. (UPDATE_DATA_BYTES_REQ) UPDATE_DATA +# e. UPDATE_ABORT_REQ (UPDATE_ABORT_CFM) +# 3. VM_DISCONNECT + + +class VmControlType(IntEnum): + # Command from the app to the device + + # Regular firmware update flow + UPDATE_SYNC_REQ = 19 + UPDATE_START_REQ = 1 + UPDATE_START_DATA_REQ = 21 + UPDATE_DATA = 4 + UPDATE_IS_VALIDATION_DONE_REQ = 22 + UPDATE_TRANSFER_COMPLETE_RES = 12 + UPDATE_IN_PROGRESS_RES = 14 + UPDATE_ABORT_REQ = 7 + + # This looks like a fancy way of aborting when + # you get an error code in the update process + # looks like you always just send one after the other + # with the same error code? + UPDATE_ABORT_WITH_CODE_1_REQ = 31 + UPDATE_ABORT_WITH_CODE_2_REQ = 32 + + # Not used in regular firmware update? + # It seems like there's a hidden debug firmware GUI + # in the app somewhere that can send these commands + UPDATE_COMMIT_CFM = 16 + UPDATE_ERASE_SQIF_CFM = 30 + + +class VmuPacketType(IntEnum): + # Replies to commands from the VMU_PACKET BT notifications + UPDATE_START_CFM = 2 + UPDATE_DATA_BYTES_REQ = 3 + UPDATE_ABORT_CFM = 8 + UPDATE_TRANSFER_COMPLETE_IND = 11 + UPDATE_SYNC_CFM = 20 + UPDATE_COMPLETE_IND = 18 + UPDATE_ERROR = 17 # Not seen in logs + UPDATE_IS_VALIDATION_DONE_CFM = 23 # Not seen in logs + UPDATE_COMMIT_ERASE_SQIF_RES = 29 # Not seen in logs + UPDATE_COMMIT_RES = 15 # Not seen in logs + + +class BoolTransform: + def forward(self, x: int) -> bool: + return bool(x) + + def back(self, y: bool) -> int: + return int(y) + + +bf_bool_byte = bf_map(bf_int(8), BoolTransform()) + + +class VmControlUpdateSyncReq(Bitfield): + md5sum_tail: bytes = bf_bytes(4) + + +class VmControlUpdateStartReq(Bitfield): + pass + + +class VmControlUpdateDataStartReq(Bitfield): + pass + + +class VmControlUpdateData(Bitfield): + is_final_fragment: bool = bf_bool_byte + data: bytes = bf_dyn(lambda _, n: bf_bytes(n // 8)) + + +class VmControlUpdateIsValidationDoneReq(Bitfield): + pass + + +class VmControlUpdateTransferCompleteRes(Bitfield): + # Misleading name: this is the app's answer to "reboot into the new image + # now?", not a report on the transfer. False proceeds with the reboot (every + # successful update in the logs), True postpones it and leaves the radio in + # UpdateState.TRANSFER_COMPLETE, which is what the app's "cancel restart" + # button does. + is_complete: bool = bf_bool_byte + + +class VmControlUpdateInProgressRes(Bitfield): + _pad: t.Literal[0] = bf_lit_int(8, default=0) + + +class VmControlUpdateAbortReq(Bitfield): + pass + + +class UpdateState(IntEnum): + DATA_TRANSFER = 0 + VALIDATION = 1 + TRANSFER_COMPLETE = 2 + IN_PROGRESS = 3 + COMMIT = 4 + + +class UpdateStartCfmCode(IntEnum): + OK = 0 + # Not seen in logs. Every UPDATE_START_CFM reports OK, including the + # post-reboot one, so the phase of an update has to be read off + # UPDATE_SYNC_CFM.update_state rather than from this code. + GOTO_NEXT_STATE = 9 + + +class UpdateError(IntEnum): + UNKNOWN = 0 + BATTERY_LOW = 33 + SYNC_IS_DIFFERENT = 129 + + @classmethod + def _missing_(cls, value: object): + import sys + print(f"Unknown value for {cls.__name__}: {value}", file=sys.stderr) + return cls.UNKNOWN + +# Messages from VMU_PACKET + + +class VmControlUpdateSyncCfm(Bitfield): + update_state: UpdateState = bf_int_enum(UpdateState, 8) + md5sum_tail: bytes = bf_bytes(4) + unknown: bytes = bf_bytes(1) + + +class VmControlUpdateStartCfm(Bitfield): + cfm_code: UpdateStartCfmCode = bf_int_enum(UpdateStartCfmCode, 8) + unknown: bytes = bf_bytes(2) + + +class VmControlUpdateCompleteInd(Bitfield): + pass + + +class VmControlUpdateTransferCompleteInd(Bitfield): + pass + + +class VmControlUpdateAbortCfm(Bitfield): + pass + + +class VmControlUpdateError(Bitfield): + update_error: UpdateError = bf_int_enum(UpdateError, 16) + + +class VmControlUpdateDataBytesReq(Bitfield): + # The max bytes requested that the HT app allows is 250 + n_bytes_requested: int = bf_int(32) + # Skip would allow resuming a firmware update, but it is 0 in every request + # in my logs, including the ones after an aborted transfer — those restart + # from the beginning instead. So whether it counts from the current position + # or from the start of the image is unknown. + n_bytes_skip: int = bf_int(32) + + +def vm_control_disc(m: VmControlBody): + match m.vm_control_type: + case VmControlType.UPDATE_SYNC_REQ: + out = VmControlUpdateSyncReq + case VmControlType.UPDATE_START_REQ: + out = VmControlUpdateStartReq + case VmControlType.UPDATE_START_DATA_REQ: + out = VmControlUpdateDataStartReq + case VmControlType.UPDATE_DATA: + out = VmControlUpdateData + case VmControlType.UPDATE_IS_VALIDATION_DONE_REQ: + out = VmControlUpdateIsValidationDoneReq + case VmControlType.UPDATE_TRANSFER_COMPLETE_RES: + out = VmControlUpdateTransferCompleteRes + case VmControlType.UPDATE_IN_PROGRESS_RES: + out = VmControlUpdateInProgressRes + case VmControlType.UPDATE_ABORT_REQ: + out = VmControlUpdateAbortReq + case _: + return bf_bytes(m.n_bytes_payload) + + return bf_bitfield(out, m.n_bytes_payload*8) + + +def vmu_packet_desc(m: VmuPacket): + match m.vmu_packet_type: + case VmuPacketType.UPDATE_DATA_BYTES_REQ: + out = VmControlUpdateDataBytesReq + case VmuPacketType.UPDATE_SYNC_CFM: + out = VmControlUpdateSyncCfm + case VmuPacketType.UPDATE_COMPLETE_IND: + out = VmControlUpdateCompleteInd + case VmuPacketType.UPDATE_TRANSFER_COMPLETE_IND: + out = VmControlUpdateTransferCompleteInd + case VmuPacketType.UPDATE_START_CFM: + out = VmControlUpdateStartCfm + case VmuPacketType.UPDATE_ERROR: + out = VmControlUpdateError + case VmuPacketType.UPDATE_ABORT_CFM: + out = VmControlUpdateAbortCfm + case _: + return bf_bytes(m.n_bytes_payload) + + return bf_bitfield(out, m.n_bytes_payload*8) + + +VmControlMessage = t.Union[ + VmControlUpdateSyncReq, + VmControlUpdateStartReq, + VmControlUpdateDataStartReq, + VmControlUpdateData, + VmControlUpdateIsValidationDoneReq, + VmControlUpdateTransferCompleteRes, + VmControlUpdateInProgressRes, + VmControlUpdateAbortReq, +] + +VmuPacketMessage = t.Union[ + VmControlUpdateDataBytesReq, + VmControlUpdateSyncCfm, + VmControlUpdateCompleteInd, + VmControlUpdateTransferCompleteInd, + VmControlUpdateStartCfm, + VmControlUpdateError, + VmControlUpdateAbortCfm, +] + + +class VmControlBody(Bitfield): + vm_control_type: VmControlType = bf_int_enum(VmControlType, 8) + n_bytes_payload: int = bf_int(16) + msg: VmControlMessage | bytes = bf_dyn(vm_control_disc) + + +class VmuPacket(Bitfield): + vmu_packet_type: VmuPacketType = bf_int_enum(VmuPacketType, 8) + n_bytes_payload: int = bf_int(16) + msg: VmuPacketMessage | bytes = bf_dyn(vmu_packet_desc) + + +class VmControlReplyBody(Bitfield): + status: ReplyStatus = bf_int_enum(ReplyStatus, 8) + + +class VmConnectBody(Bitfield): + pass + + +class VmConnectReplyBody(Bitfield): + status: ReplyStatus = bf_int_enum(ReplyStatus, 8) + + +class VmDisconnectBody(Bitfield): + pass + + +class VmDisconnectReplyBody(Bitfield): + status: ReplyStatus = bf_int_enum(ReplyStatus, 8) diff --git a/tests/test_firmware.py b/tests/test_firmware.py new file mode 100644 index 0000000..6938e6c --- /dev/null +++ b/tests/test_firmware.py @@ -0,0 +1,203 @@ +import io +import zipfile + +import pytest + +from benlink.firmware import _rpc +from benlink.firmware import ( + PRODUCTS, + FirmwareBundle, + FirmwareInfo, + UpdateInfo, + assemble, + extract_base, + oss_base_url, + oss_patch_url, + oss_update_info, +) + +bsdiff4 = pytest.importorskip("bsdiff4") + + +def _varint(value: int) -> bytes: + out = bytearray() + while value > 0x7F: + out.append((value & 0x7F) | 0x80) + value >>= 7 + out.append(value) + return bytes(out) + + +def _delimited(field: int, payload: bytes) -> bytes: + return _varint(field << 3 | 2) + _varint(len(payload)) + payload + + +def _firmware_info_bytes(info: _rpc.FirmwareInfo) -> bytes: + return ( + _varint(1 << 3) + _varint(info.version) + + _delimited(2, info.url.encode()) + + _delimited(3, info.md5.encode()) + ) + + +def test_encode_check_request(): + # Field numbers are the wire contract: product_id is field 1, varint. + assert _rpc.encode_check_request(259) == b"\x08\x83\x02" + assert _rpc.encode_check_request(259, 147) == b"\x08\x83\x02\x10\x93\x01" + # proto3 omits zero-valued fields + assert _rpc.encode_check_request(0) == b"" + + +def test_encode_decode_roundtrip(): + info = _rpc.FirmwareInfo(147, "https://example.invalid/p.bin", "abc") + encoded = _delimited(1, _firmware_info_bytes(info)) + decoded = _rpc.decode_check_result(encoded) + assert decoded.firmware == info + assert decoded.base == _rpc.FirmwareInfo() + + +def test_decode_stops_on_unknown_wire_type(): + # tag with wire type 7 (invalid); the walk must not loop or raise + assert _rpc.decode_check_result(b"\x0f\x01\x02") == ( + _rpc.CheckFirmwareUpdateResult() + ) + + +def test_update_info_from_protocol(): + result = _rpc.CheckFirmwareUpdateResult( + firmware=_rpc.FirmwareInfo( + version=147, + url="https://example.invalid/patch.bin", + md5="0c0d095da50bebe664822adcb244834a", + ), + base=_rpc.FirmwareInfo( + url="https://example.invalid/base.zip", + md5="74b6d097d8d2d9d2d9fac88133198a08", + ), + ) + + assert UpdateInfo.from_protocol(result) == UpdateInfo( + firmware=FirmwareInfo( + version=147, + url="https://example.invalid/patch.bin", + md5="0c0d095da50bebe664822adcb244834a", + ), + base=FirmwareInfo( + version=0, + url="https://example.invalid/base.zip", + md5="74b6d097d8d2d9d2d9fac88133198a08", + ), + ) + + +def test_update_info_from_protocol_empty_means_no_update(): + empty = _rpc.CheckFirmwareUpdateResult() + assert UpdateInfo.from_protocol(empty) is None + + +def test_oss_urls(): + assert oss_patch_url(147, "patch_base_to_vr_n76").endswith( + "/firmware/v147/patch_base_to_vr_n76.bin") + assert oss_patch_url(147, "custom").endswith("/firmware/v147/custom.bin") + assert oss_base_url("original").endswith("/upgrade_base.bin.zip") + assert oss_base_url("1").endswith("/upgrade_base_v1.bin.zip") + + +def test_oss_base_url_rejects_unknown_base(): + with pytest.raises(RuntimeError, match="unknown base image"): + oss_base_url("2") + + +def test_oss_update_info(): + info = oss_update_info(147, "patch_base_to_vr_n76", "1") + assert info.firmware.url.endswith("/firmware/v147/patch_base_to_vr_n76.bin") + assert info.base.url.endswith("/upgrade_base_v1.bin.zip") + assert info.firmware.md5 is None + + +def test_assemble_raw_base(): + base = b"the quick brown fox" * 100 + expected = b"the slow brown fox" * 100 + assert assemble(base, bsdiff4.diff(base, expected)) == expected + + +def test_assemble_zipped_base(): + base = b"the quick brown fox" * 100 + expected = b"the slow brown fox" * 100 + + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("upgrade_base.bin", base) + + assert assemble(buf.getvalue(), bsdiff4.diff(base, expected)) == expected + + +def test_extract_base(): + raw = b"not a zip" + assert extract_base(raw) == raw + + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("upgrade_base.bin", b"inner") + assert extract_base(buf.getvalue()) == b"inner" + + +def test_extract_base_rejects_zip_without_bin(): + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("readme.txt", b"nope") + with pytest.raises(RuntimeError, match="no .bin found"): + extract_base(buf.getvalue()) + + +def test_assemble_against_wrong_base_is_not_detected(): + # BSDIFF40 carries no checksum of its source, so the wrong base yields a + # plausible but corrupt image. Callers must verify the assembled result. + base = b"the quick brown fox" * 100 + other = b"a completely different base" * 100 + patch = bsdiff4.diff(base, b"target" * 100) + + assert assemble(other, patch) != b"target" * 100 + + +def test_assemble_rejects_bad_patch_magic(): + with pytest.raises(RuntimeError, match="unexpected patch magic"): + assemble(b"base", b"NOTAPATCH" + b"\x00" * 32) + + +def test_resolve_product(): + from argparse import Namespace + + from benlink.firmware.__main__ import _resolve_product + + assert _resolve_product( + Namespace(product="UV_PRO", product_id=None, patch_name=None) + ) == PRODUCTS["UV_PRO"] + + # explicit flags override either half of --product + assert _resolve_product( + Namespace(product="UV_PRO", product_id=999, patch_name=None) + ) == (999, PRODUCTS["UV_PRO"][1]) + + assert _resolve_product( + Namespace(product="UV_PRO", product_id=None, patch_name="custom") + ) == (PRODUCTS["UV_PRO"][0], "custom") + + assert _resolve_product( + Namespace(product=None, product_id=None, patch_name=None) + ) == (None, None) + + +def test_ga5wb_shares_vr_n76_patch_series(): + # Confirmed against a GA-5WB flash capture; see PRODUCTS docstring. + assert PRODUCTS["GA_5WB"] == PRODUCTS["VR_N76"] + + +def test_bundle_md5_tail(): + bundle = FirmwareBundle( + data=b"hello", + update_info=oss_update_info(1, "patch_base_to_vr_n76", "1"), + ) + assert bundle.md5 == "5d41402abc4b2a76b9719d911017c592" + assert bundle.md5_tail == bytes.fromhex("1017c592") + assert bundle.size == 5 diff --git a/tests/test_flash.py b/tests/test_flash.py new file mode 100644 index 0000000..97a2176 --- /dev/null +++ b/tests/test_flash.py @@ -0,0 +1,415 @@ +import asyncio +import hashlib +import typing as t + +import pytest + +from benlink.command import CommandConnection +from benlink.firmware import abort_update, flash +from benlink.firmware._flash import FlashError, FlashResult +import benlink.protocol as p +from benlink.protocol.command.bt_notification import ( + BtEventNotificationBody, BtEventType, +) +from benlink.protocol.command.vm import ( + UpdateError, + UpdateState, + VmConnectReplyBody, + VmControlReplyBody, + VmControlType, + VmControlUpdateData, + VmControlUpdateDataBytesReq, + VmControlUpdateError, + VmControlUpdateStartCfm, + VmControlUpdateSyncCfm, + VmControlUpdateSyncReq, + VmControlUpdateTransferCompleteRes, + VmuPacket, + VmuPacketType, + UpdateStartCfmCode, + VmControlBody, + VmControlUpdateAbortCfm, + VmControlUpdateCompleteInd, + VmControlUpdateTransferCompleteInd, +) + +CHUNK = 145 + + +class FakeRadio: + """A radio that answers the update messages the way the captures do. + + Everything is round-tripped through `to_bytes`/`from_bytes` so the test + exercises real serialization in both directions. + """ + + def __init__( + self, + state: UpdateState = UpdateState.DATA_TRANSFER, + chunk: int = CHUNK, + skip_first: int = 0, + error_after: int | None = None, + preempt_sync: bool = False, + interrupt_after: int | None = None, + interrupt_abort_too: bool = False, + staged_tail: bytes | None = None, + ): + self.state = state + self.chunk = chunk + self.skip_first = skip_first + self.error_after = error_after + self.received = bytearray() + self.sent: t.List[VmControlType] = [] + self.final_flags: t.List[bool] = [] + self.error_on_finalize = False + self.preempt_sync = preempt_sync + self.interrupt_after = interrupt_after + self.interrupt_abort_too = interrupt_abort_too + self.staged_tail = staged_tail + self.disconnected = False + self.vmu_registered = False + self.aborted = False + self._callback: t.Any = None + self._chunks_served = 0 + + # CommandLink + + def is_connected(self) -> bool: + return True + + async def connect(self, callback: t.Any) -> None: + self._callback = callback + + async def disconnect(self) -> None: + pass + + async def send_bytes(self, data: bytes) -> None: + raise AssertionError("flash should not use send_bytes") + + async def send(self, msg: p.Message) -> None: + if (self.interrupt_after is not None + and self.sent.count(VmControlType.UPDATE_DATA) + >= self.interrupt_after): + if not self.interrupt_abort_too: + # One interrupt only, so the abort that follows can land. + self.interrupt_after = None + raise KeyboardInterrupt + self._handle(p.Message.from_bytes(msg.to_bytes())) + + # Radio behaviour + + def _emit(self, command: p.ExtendedCommand, body: t.Any, is_reply: bool) -> None: + out = p.Message( + command_group=p.CommandGroup.EXTENDED, + is_reply=is_reply, + command=command, + body=body, + ) + self._callback(p.Message.from_bytes(out.to_bytes())) + + def _emit_vmu(self, packet_type: VmuPacketType, msg: t.Any) -> None: + if not self.vmu_registered: + # A real radio sends nothing until REGISTER_BT_NOTIFICATION. + return + packet = VmuPacket( + vmu_packet_type=packet_type, + n_bytes_payload=len(msg.to_bytes()), + msg=msg, + ) + self._emit( + p.ExtendedCommand.BT_EVENT_NOTIFICATION, + BtEventNotificationBody( + bt_event_type=BtEventType.VMU_PACKET, bt_event=packet + ), + is_reply=False, + ) + + def _request_bytes(self) -> None: + skip = self.skip_first if self._chunks_served == 0 else 0 + self._chunks_served += 1 + self._emit_vmu( + VmuPacketType.UPDATE_DATA_BYTES_REQ, + VmControlUpdateDataBytesReq( + n_bytes_requested=self.chunk, n_bytes_skip=skip + ), + ) + + def _emit_sync_cfm(self, md5_tail: bytes) -> None: + self._emit_vmu( + VmuPacketType.UPDATE_SYNC_CFM, + VmControlUpdateSyncCfm( + update_state=self.state, md5sum_tail=md5_tail, unknown=b"\x00" + ), + ) + + def _handle(self, msg: p.Message) -> None: + if msg.command == p.ExtendedCommand.VM_CONNECT: + self._emit( + p.ExtendedCommand.VM_CONNECT, + VmConnectReplyBody(status=p.ReplyStatus.SUCCESS), + is_reply=True, + ) + return + + if msg.command == p.ExtendedCommand.REGISTER_BT_NOTIFICATION: + assert msg.body == bytes([BtEventType.VMU_PACKET]) + self.vmu_registered = True + if self.preempt_sync: + # Answers a question that has not been asked yet. + self._emit_sync_cfm( + self.staged_tail or b"\x00\x00\x00\x00") + return + + if msg.command == p.ExtendedCommand.CANCEL_BT_NOTIFICATION: + self.vmu_registered = False + return + + if msg.command == p.ExtendedCommand.VM_DISCONNECT: + self.disconnected = True + return + + body = msg.body + assert isinstance(body, VmControlBody) + self.sent.append(body.vm_control_type) + + # Every VM_CONTROL is acknowledged before the answer arrives. + self._emit( + p.ExtendedCommand.VM_CONTROL, + VmControlReplyBody(status=p.ReplyStatus.SUCCESS), + is_reply=True, + ) + + match body.vm_control_type: + case VmControlType.UPDATE_SYNC_REQ: + assert isinstance(body.msg, VmControlUpdateSyncReq) + if not self.preempt_sync: + self._emit_sync_cfm( + self.staged_tail or body.msg.md5sum_tail) + case VmControlType.UPDATE_START_REQ: + self._emit_vmu( + VmuPacketType.UPDATE_START_CFM, + VmControlUpdateStartCfm( + cfm_code=UpdateStartCfmCode.OK, unknown=b"\x00\x00" + ), + ) + case VmControlType.UPDATE_START_DATA_REQ: + self._request_bytes() + case VmControlType.UPDATE_DATA: + assert isinstance(body.msg, VmControlUpdateData) + self.received += body.msg.data + self.final_flags.append(body.msg.is_final_fragment) + if self.error_after is not None and \ + len(self.received) >= self.error_after: + self._emit_vmu( + VmuPacketType.UPDATE_ERROR, + VmControlUpdateError( + update_error=UpdateError.BATTERY_LOW), + ) + elif not body.msg.is_final_fragment: + self._request_bytes() + case VmControlType.UPDATE_IS_VALIDATION_DONE_REQ: + self._emit_vmu( + VmuPacketType.UPDATE_TRANSFER_COMPLETE_IND, + VmControlUpdateTransferCompleteInd(), + ) + case VmControlType.UPDATE_IN_PROGRESS_RES: + if self.error_on_finalize: + self._emit_vmu( + VmuPacketType.UPDATE_ERROR, + VmControlUpdateError(update_error=UpdateError.UNKNOWN), + ) + else: + self._emit_vmu( + VmuPacketType.UPDATE_COMPLETE_IND, + VmControlUpdateCompleteInd(), + ) + case VmControlType.UPDATE_ABORT_REQ: + self.aborted = True + self._emit_vmu( + VmuPacketType.UPDATE_ABORT_CFM, VmControlUpdateAbortCfm() + ) + case VmControlType.UPDATE_TRANSFER_COMPLETE_RES: + # Acked like every control message; the reboot is the answer. + pass + case _: + raise AssertionError( + f"flash sent an unexpected {body.vm_control_type.name}" + ) + + +def _run(radio: FakeRadio, image: bytes, **kwargs: t.Any) -> FlashResult: + async def main() -> FlashResult: + conn = CommandConnection(radio) + await conn.connect() + return await flash(conn, image, **kwargs) + + return asyncio.run(main()) + + +def test_transfer_phase_sends_whole_image(): + data = bytes(range(256)) * 5 + radio = FakeRadio() + + result = _run(radio, data) + + assert result == "REBOOT_PENDING" + assert bytes(radio.received) == data + assert radio.sent[-1] == VmControlType.UPDATE_TRANSFER_COMPLETE_RES + + +def test_final_fragment_is_flagged_once_at_the_end(): + data = b"x" * (CHUNK * 3) + radio = FakeRadio() + _run(radio, data) + + # The radio stops asking for more only because the last UPDATE_DATA said so, + # and an image that divides evenly into chunks must still flag its last one. + assert radio.final_flags == [False, False, True] + + +def test_image_shorter_than_one_chunk(): + data = b"tiny" + radio = FakeRadio() + + assert _run(radio, data) == "REBOOT_PENDING" + assert bytes(radio.received) == data + + +def test_progress_reports_reach_the_total(): + data = b"y" * (CHUNK * 2 + 7) + seen: t.List[t.Tuple[str, int, int]] = [] + + def record(label: str, done: int, total: int) -> None: + seen.append((label, done, total)) + + _run(FakeRadio(), data, progress=record) + + assert [n for _, n, _ in seen] == [CHUNK, CHUNK * 2, len(data)] + assert all(total == len(data) for _, _, total in seen) + + +def test_resume_honours_n_bytes_skip(): + data = bytes(range(256)) * 4 + radio = FakeRadio(skip_first=300) + + _run(radio, data) + + # The radio already had the first 300 bytes, so they are never resent. + assert bytes(radio.received) == data[300:] + + +def test_in_progress_state_finalizes_instead_of_transferring(): + radio = FakeRadio(state=UpdateState.IN_PROGRESS) + + result = _run(radio, b"unused") + + assert result == "COMPLETE" + assert VmControlType.UPDATE_DATA not in radio.sent + assert VmControlType.UPDATE_IN_PROGRESS_RES in radio.sent + assert radio.disconnected + + +def test_transfer_complete_state_only_asks_for_the_reboot(): + radio = FakeRadio(state=UpdateState.TRANSFER_COMPLETE) + + result = _run(radio, b"unused") + + assert result == "REBOOT_PENDING" + assert VmControlType.UPDATE_DATA not in radio.sent + assert radio.sent[-1] == VmControlType.UPDATE_TRANSFER_COMPLETE_RES + + +def test_reboot_request_asks_the_radio_to_restart_now(): + """The byte is 0 on the app's success path; 1 is its "cancel restart".""" + radio = FakeRadio(state=UpdateState.TRANSFER_COMPLETE) + sent: t.List[p.Message] = [] + original = radio.send + + async def record(msg: p.Message) -> None: + sent.append(msg) + await original(msg) + + radio.send = record # type: ignore[method-assign] + _run(radio, b"unused") + + final = sent[-1] + assert isinstance(final.body, VmControlBody) + assert isinstance(final.body.msg, VmControlUpdateTransferCompleteRes) + assert final.body.msg.is_complete is False + + +def test_update_error_is_raised_not_waited_out(): + radio = FakeRadio(error_after=CHUNK) + + with pytest.raises(FlashError, match="BATTERY_LOW"): + _run(radio, b"z" * CHUNK * 10) + + assert radio.aborted + + +def test_failure_after_staging_does_not_abort(): + """Aborting here would discard an image the radio has already validated.""" + radio = FakeRadio(state=UpdateState.IN_PROGRESS) + radio.error_on_finalize = True + + with pytest.raises(FlashError): + _run(radio, b"unused") + + assert not radio.aborted + + +def test_reply_arriving_before_it_is_awaited_is_not_lost(): + """The subscription is opened before the first send and held for the whole + flash, so a radio that answers early is buffered rather than dropped.""" + image = b"unused" + radio = FakeRadio(state=UpdateState.TRANSFER_COMPLETE, preempt_sync=True, + staged_tail=hashlib.md5(image).digest()[-4:]) + + assert _run(radio, image) == "REBOOT_PENDING" + assert radio.sent[-1] == VmControlType.UPDATE_TRANSFER_COMPLETE_RES + + +def test_interrupt_mid_transfer_still_tells_the_radio(): + """Ctrl+C is not an `Exception`, but the radio is left mid-transfer.""" + radio = FakeRadio(interrupt_after=3) + + with pytest.raises(KeyboardInterrupt): + _run(radio, b"x" * CHUNK * 100) + + assert radio.aborted + assert radio.sent.count(VmControlType.UPDATE_DATA) == 3 + + +def test_second_interrupt_is_not_swallowed_by_the_abort(): + radio = FakeRadio(interrupt_after=3, interrupt_abort_too=True) + + with pytest.raises(KeyboardInterrupt): + _run(radio, b"x" * CHUNK * 100) + + assert not radio.aborted + + +def test_refuses_to_finish_someone_elses_update(): + """A radio holding a different image must not be committed by mistake.""" + radio = FakeRadio(state=UpdateState.IN_PROGRESS, staged_tail=b"\xde\xad\xbe\xef") + + with pytest.raises(FlashError, match="different image"): + _run(radio, b"unused") + + assert VmControlType.UPDATE_IN_PROGRESS_RES not in radio.sent + + +def test_abort_update_clears_a_stranded_radio(): + """The way out when the image that produced a staged update is gone.""" + radio = FakeRadio(state=UpdateState.IN_PROGRESS) + + async def main() -> None: + conn = CommandConnection(radio) + await conn.connect() + await abort_update(conn) + + asyncio.run(main()) + + assert radio.aborted + assert radio.disconnected + assert VmControlType.UPDATE_IN_PROGRESS_RES not in radio.sent diff --git a/update_readme.py b/update_readme.py index 18e3edb..321e291 100755 --- a/update_readme.py +++ b/update_readme.py @@ -25,7 +25,9 @@ raise ValueError("No content section found in README.md.") readme_content_stripped = [ - line[1:] if line.startswith("##") else line + # Backslashes are doubled because the content lands inside a docstring, where + # markdown escapes like \_ would otherwise be invalid escape sequences. + (line[1:] if line.startswith("##") else line).replace("\\", "\\\\") for line in readme_content[readme_start+1:] ] @@ -36,6 +38,6 @@ *init_content[docstring_end:] ] -init_path.write_text("\n".join(updated_content)) +init_path.write_text("\n".join(updated_content) + "\n") print(f"README content has been updated into module definition")