From 386d03bdd8392ff57f4d0b5b9964b51e49fce959 Mon Sep 17 00:00:00 2001 From: ArthurKoba <61914203+ArthurKoba@users.noreply.github.com> Date: Thu, 17 Sep 2026 01:19:03 +0300 Subject: [PATCH 1/4] install: unlock SPI NOR before persistent writes Issue `sf lock 0` after a successful NOR probe and before any selected persistent NOR stage. This matches the established OpenIPC flashing sequence and handles boards where erase/write is rejected with a locked DMA/write area. Keep older U-Boot variants compatible by warning and continuing when the lock subcommand is unavailable; explicit unlock failures still stop the install before destructive writes. Add regression coverage for successful unlock, unsupported legacy `sf` syntax, and hard unlock failure ordering. Hardware verified on a GARUS Hi3516EV200 board with 8 MiB SPI NOR: both a kernel-only install and a full install run executed the unlock successfully; U-Boot, kernel, and rootfs writes passed CRC readback and the factory MAC was preserved. --- src/defib/install/orchestrator.py | 24 +++++ tests/test_install_nor_unlock.py | 165 ++++++++++++++++++++++++++++++ 2 files changed, 189 insertions(+) create mode 100644 tests/test_install_nor_unlock.py diff --git a/src/defib/install/orchestrator.py b/src/defib/install/orchestrator.py index a2439c9..63401be 100644 --- a/src/defib/install/orchestrator.py +++ b/src/defib/install/orchestrator.py @@ -658,6 +658,30 @@ async def _set_env_verified_or_fail(key: str, value: str) -> None: if len(rootfs_data) > r_sz: await close_and_fail(f"Rootfs too large: {len(rootfs_data)} > {r_sz}") + persistent_nor_stages = {"uboot", "kernel", "rootfs", "rootfs-data", "env"} + if stage_set & persistent_nor_stages: + unlock_resp = await _cmd("sf lock 0", timeout=5.0) + unlock_text = unlock_resp.lower() + unlock_unsupported = ( + "unknown command" in unlock_text + or "usage:" in unlock_text + or "not supported" in unlock_text + or "unsupported" in unlock_text + ) + if unlock_unsupported: + warn( + "U-Boot does not expose a usable `sf lock` command; " + "continuing and relying on erase/write result checks." + ) + else: + unlock_error = uboot_flash_command_error(unlock_resp) + if unlock_error: + await close_and_fail( + f"SPI NOR unlock failed ({unlock_error}): {unlock_resp.strip()}" + ) + if output == "human": + console.print(" [green]SPI NOR write protection cleared[/green]") + if output == "human": console.print( f" [green]SPI flash detected[/green]: {nor_size} MiB ({nor_source}), " diff --git a/tests/test_install_nor_unlock.py b/tests/test_install_nor_unlock.py new file mode 100644 index 0000000..ed8ee00 --- /dev/null +++ b/tests/test_install_nor_unlock.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import io +import tarfile + +import pytest +import typer + +from defib.install import InstallRequest +from defib.recovery.events import RecoveryResult +from defib.transport.base import Transport, TransportTimeout + + +class ShellTransport(Transport): + def __init__(self) -> None: + self.rx = bytearray() + self.closed = False + + async def read(self, size: int, timeout: float | None = None) -> bytes: + if not self.rx: + raise TransportTimeout("no data") + data = bytes(self.rx[:size]) + del self.rx[:size] + return data + + async def write(self, data: bytes) -> None: + if b"\x03" in data: + self.rx.extend(b"OpenIPC # ") + + async def flush_input(self) -> None: + self.rx.clear() + + async def flush_output(self) -> None: + return None + + async def bytes_waiting(self) -> int: + return len(self.rx) + + async def close(self) -> None: + self.closed = True + + +async def _run_env_only_install(monkeypatch, tmp_path, unlock_response: str): + import defib.flashdump + import defib.recovery.session + import defib.transport.serial_platform + from defib.install.orchestrator import run_install + + firmware_tar = tmp_path / "firmware.tgz" + with tarfile.open(firmware_tar, "w:gz") as archive: + for name, payload in ( + ("uImage.hi3516ev200", b"K" * 1024), + ("rootfs.squashfs.hi3516ev200", b"R" * 2048), + ): + info = tarfile.TarInfo(name) + info.size = len(payload) + archive.addfile(info, io.BytesIO(payload)) + + uboot = tmp_path / "u-boot.bin" + uboot.write_bytes(b"U" * 1024) + + transport = ShellTransport() + commands: list[str] = [] + + class FakeRecoverySession: + def __init__(self, *args, **kwargs) -> None: + pass + + async def run(self, transport_obj, **kwargs): + assert transport_obj is transport + return RecoveryResult(success=True) + + async def fake_create_transport(port: str): + assert port == "COM15" + return transport + + async def fake_send_command( + transport_obj, command: str, timeout: float = 0.0, **kwargs + ) -> str: + assert transport_obj is transport + commands.append(command) + if command == "sf probe 0": + return 'Spi(cs1): Block:64KB Chip:8MB Name:"XT25F64B"\nOpenIPC # ' + if command == "sf lock 0": + return unlock_response + if command == "printenv ethaddr": + return "ethaddr=00:12:41:8e:c6:0e\nOpenIPC # " + if command == "saveenv": + return "Saving Environment to SPI Flash... done\nOpenIPC # " + return "OpenIPC # " + + monkeypatch.setattr(defib.flashdump, "send_command", fake_send_command) + monkeypatch.setattr( + defib.recovery.session, + "RecoverySession", + FakeRecoverySession, + ) + monkeypatch.setattr( + defib.transport.serial_platform, + "create_transport", + fake_create_transport, + ) + monkeypatch.setattr( + defib.transport.serial_platform, + "normalize_port_name", + lambda port: port, + ) + + request = InstallRequest( + chip="hi3516ev200", + firmware_path=str(firmware_tar), + uboot_path=str(uboot), + port="COM15", + nor_size=8, + stages=("env",), + output="json", + ) + return run_install, request, commands, transport + + +@pytest.mark.asyncio +async def test_nor_unlock_runs_before_persistent_env_write(monkeypatch, tmp_path): + run_install, request, commands, transport = await _run_env_only_install( + monkeypatch, + tmp_path, + "OpenIPC # ", + ) + + await run_install(request) + + assert commands.index("sf probe 0") < commands.index("sf lock 0") + assert commands.index("sf lock 0") < commands.index("saveenv") + assert transport.closed is True + + +@pytest.mark.asyncio +async def test_nor_unlock_unsupported_is_compatible(monkeypatch, tmp_path): + run_install, request, commands, transport = await _run_env_only_install( + monkeypatch, + tmp_path, + "Usage:\nsf probe [[bus:]cs] [hz] [mode]\nsf read addr offset len\nOpenIPC # ", + ) + + await run_install(request) + + assert "sf lock 0" in commands + assert "saveenv" in commands + assert transport.closed is True + + +@pytest.mark.asyncio +async def test_nor_unlock_failure_stops_before_persistent_write(monkeypatch, tmp_path): + run_install, request, commands, transport = await _run_env_only_install( + monkeypatch, + tmp_path, + "ERROR: SPI NOR unlock failed\nOpenIPC # ", + ) + + with pytest.raises(typer.Exit) as exc_info: + await run_install(request) + + assert exc_info.value.exit_code == 1 + assert "sf lock 0" in commands + assert "saveenv" not in commands + assert transport.closed is True From d21753ec73e619de729444e7135567ce15101c64 Mon Sep 17 00:00:00 2001 From: ArthurKoba <61914203+ArthurKoba@users.noreply.github.com> Date: Fri, 18 Sep 2026 02:11:58 +0300 Subject: [PATCH 2/4] install: make NOR flashing fail closed and retry corrupt TFTP RAM Hardware testing exposed two independent reliability failures around persistent installs: incomplete U-Boot command responses could advance the state machine, and a completed TFTP transfer could occasionally leave corrupt RAM contents. Require prompt completion for ordinary shell commands, preserve the explicit download-command result, reject a failed sf probe, and keep reset on its own promptless control-flow path. Verify every TFTP-staged payload before persistent writes by checking the reported transfer size and RAM CRC. Keep retry policy in the named TFTP_RAM_VERIFY_RETRIES constant, defaulting to one retry. When verification fails and another attempt is available, emit an operator-facing warning naming the next attempt and the exact TFTP file being fetched again. Host TFTP switches future RFC2348 negotiation to 512-byte blocks after the first failed verification; pod TFTP retries without host-side blocksize control. Apply the same verified staging helper to U-Boot, kernel, NOR rootfs, and the extracted UBIFS payload on NAND/UBI installs. Phase-1 boot-ROM/SPL/U-Boot upload is a separate transport and is intentionally outside this TFTP policy. Regression coverage deterministically corrupts the first RAM CRC and proves that the same file is fetched a second time, the warning reports Attempt 2, no flash erase happens before the second CRC succeeds, and the host fallback caps TFTP blocks at 512 bytes. Persistent corruption still fails closed before any flash write. Hardware verification remains on this test branch before folding the result into the existing PR. --- src/defib/flashdump.py | 9 +- src/defib/install/orchestrator.py | 232 +++++++++++++++++++++------ src/defib/network/tftp_server.py | 20 ++- tests/test_ds_i203_final_contract.py | 70 +++++++- tests/test_install_nor_unlock.py | 128 ++++++++++++++- tests/test_network.py | 21 +++ tests/test_uart_command_integrity.py | 21 +++ 7 files changed, 443 insertions(+), 58 deletions(-) diff --git a/src/defib/flashdump.py b/src/defib/flashdump.py index 6908c60..c5107f4 100644 --- a/src/defib/flashdump.py +++ b/src/defib/flashdump.py @@ -316,7 +316,14 @@ async def send_command( return buf.decode("ascii", errors="replace") continue - return buf.decode("ascii", errors="replace") + response = buf.decode("ascii", errors="replace") + if wait_for: + partial = response.strip()[-200:] or "" + raise TransportTimeout( + f"Timed out waiting for {wait_for!r} after U-Boot command {cmd!r}; " + f"partial response: {partial}" + ) + return response async def tftp_to_ram( diff --git a/src/defib/install/orchestrator.py b/src/defib/install/orchestrator.py index 63401be..538f3c6 100644 --- a/src/defib/install/orchestrator.py +++ b/src/defib/install/orchestrator.py @@ -32,6 +32,9 @@ from defib.install.model import InstallRequest, resolve_install_stages +TFTP_RAM_VERIFY_RETRIES = 1 + + async def run_install(request: InstallRequest) -> None: import json as json_mod import logging @@ -50,7 +53,7 @@ async def run_install(request: InstallRequest) -> None: ) from defib.flashdump import get_ram_staging_addr, send_command from defib.network.ip_manager import list_interfaces_async, temporary_ip - from defib.network.tftp_server import start_tftp_server + from defib.network.tftp_server import DEFAULT_BLOCKSIZE, start_tftp_server from defib.profiles.loader import recovery_mode from defib.recovery.events import LogEvent, ProgressEvent from defib.recovery.session import RecoverySession @@ -555,22 +558,46 @@ async def _wait_for_openipc_shell_after_reset(timeout: float = 20.0) -> None: from defib.protocol.download_cmd import DownloadCommandClient dl_client = DownloadCommandClient(transport) - async def _cmd(cmd: str, timeout: float = 60.0, **kw: object) -> str: + async def _cmd_result( + cmd: str, + timeout: float = 60.0, + *, + allow_failure: bool = False, + **kw: object, + ) -> tuple[bool, str]: try: ok, out = await dl_client.send_command(cmd, timeout=timeout) except TransportError as exc: - await close_and_fail(f"U-Boot transport failed while running {cmd!r}: {exc}") - if not ok and output == "human": - console.print(f" [yellow]Warning: {cmd} → ERROR[/yellow]") + await close_and_fail( + f"U-Boot transport failed while running {cmd!r}: {exc}" + ) + if not ok and not allow_failure: + detail = out.strip()[-200:] or "" + await close_and_fail( + f"U-Boot command failed or timed out while running {cmd!r}: " + f"{detail}" + ) + return ok, out + + async def _cmd(cmd: str, timeout: float = 60.0, **kw: object) -> str: + del kw + _, out = await _cmd_result(cmd, timeout=timeout) return out else: if output == "human": console.print(" [cyan]U-Boot shell mode[/cyan]") - async def _cmd(cmd: str, timeout: float = 60.0, **kw: object) -> str: + async def _cmd_result( + cmd: str, + timeout: float = 60.0, + *, + allow_failure: bool = False, + **kw: object, + ) -> tuple[bool, str]: # Stock U-Boot consoles on older HiSilicon boards can corrupt bytes while - # a line is being entered. For those boards, require U-Boot to echo - # every character before Enter is sent. + # a line is being entered. For those boards, require U-Boot to echo + # every character before Enter is sent. send_command also requires the + # prompt to return, so a partial response cannot advance the install. for attempt in range(2): try: out = await send_command( @@ -585,11 +612,37 @@ async def _cmd(cmd: str, timeout: float = 60.0, **kw: object) -> str: f"U-Boot transport failed while running {cmd!r}: {exc}" ) if "unknown command" not in out.lower() or attempt == 1: - return out + return True, out await transport.write(b"\x03\r") await _aio.sleep(0.05) + return True, out + + async def _cmd(cmd: str, timeout: float = 60.0, **kw: object) -> str: + del kw + _, out = await _cmd_result(cmd, timeout=timeout) return out + async def _reset_command(timeout: float) -> None: + """Issue reset without requiring the old U-Boot prompt to return.""" + if download_mode: + # download_process may reset before it can emit [EOT](OK). Preserve + # the historical behavior: once reset is issued, lack of a protocol + # completion marker is not itself an install failure. + await _cmd_result("reset", timeout=timeout, allow_failure=True) + return + try: + await send_command( + transport, + "reset", + timeout=timeout, + wait_for=None, + verify_echo=verify_shell_echo, + ) + except TransportError as exc: + await close_and_fail( + f"U-Boot transport failed while running 'reset': {exc}" + ) + async def _set_env_verified_or_fail(key: str, value: str) -> None: try: await set_uboot_env_verified(_cmd, key, value) @@ -629,8 +682,11 @@ async def _set_env_verified_or_fail(key: str, value: str) -> None: console.print(" [green]NAND flash detected[/green]") else: resp = await _cmd("sf probe 0", timeout=5.0) - if "error" in resp.lower() or "fail" in resp.lower(): - await close_and_fail(f"sf probe failed: {resp.strip()}") + probe_error = uboot_flash_command_error(resp) + if probe_error: + await close_and_fail( + f"sf probe failed ({probe_error}): {resp.strip()}" + ) block_match = re_mod.search(r"Block:\s*(\d+)\s*KB", resp, re_mod.IGNORECASE) if block_match: nor_erase_block = int(block_match.group(1)) * 1024 @@ -660,7 +716,11 @@ async def _set_env_verified_or_fail(key: str, value: str) -> None: persistent_nor_stages = {"uboot", "kernel", "rootfs", "rootfs-data", "env"} if stage_set & persistent_nor_stages: - unlock_resp = await _cmd("sf lock 0", timeout=5.0) + unlock_ok, unlock_resp = await _cmd_result( + "sf lock 0", + timeout=5.0, + allow_failure=True, + ) unlock_text = unlock_resp.lower() unlock_unsupported = ( "unknown command" in unlock_text @@ -673,6 +733,11 @@ async def _set_env_verified_or_fail(key: str, value: str) -> None: "U-Boot does not expose a usable `sf lock` command; " "continuing and relying on erase/write result checks." ) + elif not unlock_ok: + detail = unlock_resp.strip()[-200:] or "" + await close_and_fail( + f"SPI NOR unlock command failed or timed out: {detail}" + ) else: unlock_error = uboot_flash_command_error(unlock_resp) if unlock_error: @@ -855,13 +920,102 @@ async def _tftp_to_ram(filename: str, timeout: float = 120.0) -> str: ram_addr, use_loadaddr=has_stock_uboot, ) - resp = await _cmd(tftpboot_cmd, timeout=timeout) + ok, resp = await _cmd_result( + tftpboot_cmd, + timeout=timeout, + allow_failure=True, + ) if "unknown command" in resp.lower(): - resp = await _cmd(tftp_cmd, timeout=timeout) + ok, resp = await _cmd_result( + tftp_cmd, + timeout=timeout, + allow_failure=True, + ) + if not ok: + detail = resp.strip()[-200:] or "" + raise RuntimeError(f"TFTP command failed or timed out: {detail}") if "done" not in resp.lower() and "bytes transferred" not in resp.lower(): raise RuntimeError(f"TFTP download failed: {resp.strip()[-200:]}") return resp + async def _verify_tftp_ram( + name: str, + tftp_name: str, + orig_data: bytes, + expected_crc: int, + ) -> int: + """Verify TFTP staging before any persistent write.""" + attempts = TFTP_RAM_VERIFY_RETRIES + 1 + for attempt in range(attempts): + failure: str + try: + tftp_resp = await _tftp_to_ram(tftp_name, timeout=120.0) + except RuntimeError as exc: + failure = str(exc) + else: + size_match = re_mod.search( + r"bytes transferred\s*=\s*(\d+)", + tftp_resp, + re_mod.IGNORECASE, + ) + reported_size = ( + int(size_match.group(1)) + if size_match is not None + else None + ) + if ( + reported_size is not None + and reported_size != len(orig_data) + ): + failure = ( + f"reported {reported_size} bytes, " + f"expected {len(orig_data)}" + ) + else: + crc_resp = await _cmd( + f"crc32 0x{ram_addr:x} 0x{len(orig_data):x}", + timeout=10.0, + ) + ram_crc = parse_uboot_crc32(crc_resp) + if ram_crc is None: + failure = ( + "CRC response did not contain a complete " + f"checksum: {crc_resp.strip()[-120:]}" + ) + elif ram_crc == expected_crc: + return ram_crc + else: + failure = ( + f"CRC expected={expected_crc:08X} " + f"got={ram_crc:08X}" + ) + + if attempt < TFTP_RAM_VERIFY_RETRIES: + next_attempt = attempt + 2 + if tftp_protocol is not None: + tftp_protocol.set_max_blocksize(DEFAULT_BLOCKSIZE) + warn( + f"Attempt {next_attempt}: fetching TFTP file " + f"{tftp_name!r} again for {name} after RAM " + f"verification failed ({failure}); using " + f"{DEFAULT_BLOCKSIZE}-byte blocks." + ) + else: + warn( + f"Attempt {next_attempt}: fetching TFTP file " + f"{tftp_name!r} again for {name} after RAM " + f"verification failed ({failure})." + ) + continue + + console.print( + f"[red]{name} TFTP RAM verification failed after " + f"{attempts} attempt(s):[/red] {failure}" + ) + raise typer.Exit(1) + + raise AssertionError("unreachable TFTP verification state") + async def tftp_and_flash( name: str, tftp_name: str, orig_data: bytes, flash_off: int, erase_sz: int, @@ -873,31 +1027,15 @@ async def tftp_and_flash( f"({len(orig_data)} bytes)" ) - try: - resp = await _tftp_to_ram(tftp_name, timeout=120.0) - except RuntimeError as e: - console.print(f"[red]TFTP failed for {name}:[/red] {e}") - raise typer.Exit(1) - - # Verify TFTP transfer in RAM before writing to flash + # Verify TFTP transfer in RAM before writing to flash. A completed + # transfer with bad RAM contents gets one conservative retry. expected_crc = zlib.crc32(orig_data) & 0xFFFFFFFF - resp = await _cmd( - f"crc32 0x{ram_addr:x} 0x{len(orig_data):x}", - timeout=10.0, + ram_crc = await _verify_tftp_ram( + name, + tftp_name, + orig_data, + expected_crc, ) - ram_crc = parse_uboot_crc32(resp) - if ram_crc is None: - console.print( - f"[red]{name} CRC check after TFTP returned no checksum:[/red] " - f"{resp.strip()[-200:]}" - ) - raise typer.Exit(1) - if ram_crc != expected_crc: - console.print( - f"[red]{name} CRC mismatch after TFTP![/red] " - f"expected={expected_crc:08X} got={ram_crc:08X}" - ) - raise typer.Exit(1) if output == "human": console.print(f" TFTP CRC verified: {ram_crc:08X}") @@ -1011,13 +1149,17 @@ async def tftp_and_flash( ) await replace_in_tftp(tftp_alias["rootfs"], ubifs_data) - try: - await _tftp_to_ram(tftp_alias["rootfs"], timeout=120.0) - except RuntimeError as exc: - console.print(f"[red]TFTP failed for rootfs:[/red] {exc}") - raise typer.Exit(1) + ubifs_crc = zlib.crc32(ubifs_data) & 0xFFFFFFFF + verified_ubifs_crc = await _verify_tftp_ram( + "rootfs (UBI)", + tftp_alias["rootfs"], + ubifs_data, + ubifs_crc, + ) if output == "human": - console.print(" TFTP OK") + console.print( + f" TFTP CRC verified: {verified_ubifs_crc:08X}" + ) await _cmd(f"nand erase 0x{r_off:x} 0x{r_sz:x}", timeout=120.0) nand_name = "hinand" @@ -1156,7 +1298,7 @@ async def tftp_and_flash( f"expected={expected_env_crc:08X} got={env_crc:08X}" ) - await _cmd("reset", timeout=1.0) + await _reset_command(timeout=1.0) await _wait_for_openipc_shell_after_reset() if output == "human": console.print(" [green]OpenIPC U-Boot defaults loaded[/green]") @@ -1272,7 +1414,7 @@ async def tftp_and_flash( if "reset" in stage_set: if output == "human": console.print("\n [bold]Resetting device...[/bold]") - await _cmd("reset", timeout=3.0) + await _reset_command(timeout=3.0) elif output == "human" and "env" in stage_set: console.print( "\n [yellow]Final reset skipped; device left at U-Boot prompt.[/yellow]" diff --git a/src/defib/network/tftp_server.py b/src/defib/network/tftp_server.py index 219ffa5..99f7868 100644 --- a/src/defib/network/tftp_server.py +++ b/src/defib/network/tftp_server.py @@ -52,6 +52,7 @@ class TFTPBindError(Exception): ERR_NO_SUCH_USER = 7 DEFAULT_BLOCKSIZE = 512 +MAX_BLOCKSIZE = 65464 DEFAULT_PORT = 69 DEFAULT_TIMEOUT = 5.0 MAX_RETRIES = 5 @@ -103,6 +104,11 @@ def __init__( self.stats = TFTPServerStats() self._done_count = done_count self._done_event = asyncio.Event() + self._max_blocksize = MAX_BLOCKSIZE + + def set_max_blocksize(self, blocksize: int) -> None: + """Cap future RFC2348 negotiations without disturbing active transfers.""" + self._max_blocksize = max(8, min(blocksize, MAX_BLOCKSIZE)) def connection_made(self, transport: asyncio.BaseTransport) -> None: self._transport = transport # type: ignore[assignment] @@ -153,18 +159,26 @@ def _handle_rrq(self, data: bytes, addr: tuple[str, int]) -> None: options[opt_name] = opt_value i += 2 + blksize_negotiated = False if "blksize" in options: try: requested_bs = int(options["blksize"]) - blocksize = max(8, min(requested_bs, 65464)) + blocksize = max(8, min(requested_bs, self._max_blocksize)) + blksize_negotiated = True + logger.debug( + "TFTP blksize requested=%d negotiated=%d", + requested_bs, + blocksize, + ) except ValueError: pass transfer = TFTPTransfer(addr=addr, data=serve_data, blocksize=blocksize) self._transfers[addr] = transfer - # Send OACK if options were negotiated - if options and blocksize != DEFAULT_BLOCKSIZE: + # Send OACK whenever the blksize option was accepted, including an + # explicit fallback to the RFC1350 default of 512 bytes. + if blksize_negotiated: oack = struct.pack("!H", OPCODE_OACK) oack += b"blksize\x00" + str(blocksize).encode() + b"\x00" self._send(addr, oack) diff --git a/tests/test_ds_i203_final_contract.py b/tests/test_ds_i203_final_contract.py index 7f09991..38e06f1 100644 --- a/tests/test_ds_i203_final_contract.py +++ b/tests/test_ds_i203_final_contract.py @@ -7,6 +7,7 @@ from __future__ import annotations import io +import json import sys import tarfile from dataclasses import dataclass @@ -648,9 +649,12 @@ async def fail_env_verify(*args, **kwargs): @pytest.mark.asyncio -@pytest.mark.parametrize("crc_failure", [None, "tftp", "readback", "env"]) +@pytest.mark.parametrize( + "crc_failure", + [None, "tftp", "tftp-once", "readback", "env"], +) async def test_ds_i203_stock_install_persists_detected_layout_but_not_camera_policy( - monkeypatch, tmp_path, crc_failure + monkeypatch, tmp_path, crc_failure, capsys ): """Exercise the complete stock->OpenIPC NOR install contract. @@ -666,6 +670,8 @@ async def test_ds_i203_stock_install_persists_detected_layout_but_not_camera_pol from defib.install import orchestrator from defib.install.layout import nor_mtdparts + + assert orchestrator.TFTP_RAM_VERIFY_RETRIES == 1 from defib.recovery.events import RecoveryResult from defib.vendors.base import UBootBootstrapResult @@ -752,15 +758,22 @@ def close(self) -> None: class FakeTFTPProtocol: def __init__(self, files): self._files = dict(files) + self.blocksize_caps: list[int] = [] + + def set_max_blocksize(self, blocksize: int) -> None: + self.blocksize_caps.append(blocksize) tftp_files: dict[str, bytes] = {} + tftp_protocol_obj: FakeTFTPProtocol | None = None async def fake_start_tftp_server(*, files, bind_addr, port, done_count): + nonlocal tftp_protocol_obj assert bind_addr == "192.168.1.11" assert done_count == 3 tftp_files.clear() tftp_files.update(files) - return FakeTFTPTransport(), FakeTFTPProtocol(files) + tftp_protocol_obj = FakeTFTPProtocol(files) + return FakeTFTPTransport(), tftp_protocol_obj env: dict[str, str] = {"ethaddr": FACTORY_MAC} saved_env: dict[str, str] = {} @@ -842,8 +855,10 @@ async def fake_send_command( if command.startswith("crc32 "): crc_calls += 1 - if crc_failure == "tftp" and crc_calls == 1: - return "CRC32 command timed out\nOpenIPC # " + if crc_failure == "tftp" and crc_calls <= 2: + return "==> 00000000\nOpenIPC # " + if crc_failure == "tftp-once" and crc_calls == 1: + return "==> 00000000\nOpenIPC # " if crc_failure == "readback" and crc_calls == 2: return "CRC32 output truncated\nOpenIPC # " parts = command.split() @@ -922,7 +937,7 @@ async def fake_send_command( output="json", ) - if crc_failure is None: + if crc_failure in (None, "tftp-once"): await orchestrator.run_install(request) else: with pytest.raises(typer.Exit) as exc_info: @@ -940,6 +955,49 @@ async def fake_send_command( assert any(cmd.startswith("cmp.l 0x82000000 ") for cmd in commands) return + if crc_failure == "tftp-once": + assert commands.count("tftpboot u") == 2 + assert tftp_protocol_obj is not None + assert tftp_protocol_obj.blocksize_caps == [512] + + warning_lines = [ + line + for line in capsys.readouterr().out.splitlines() + if '"event": "warning"' in line + ] + assert len(warning_lines) == 1 + warning = json.loads(warning_lines[0]) + assert warning["message"].startswith( + "Attempt 2: fetching TFTP file 'u' again for U-Boot " + ) + assert "CRC expected=" in warning["message"] + assert "using 512-byte blocks." in warning["message"] + + uboot_tftp = [ + index + for index, command in enumerate(commands) + if command == "tftpboot u" + ] + uboot_crc = [ + index + for index, command in enumerate(commands) + if command.startswith("crc32 0x82000000 0x40000") + ] + first_erase = next( + index + for index, command in enumerate(commands) + if command.startswith("sf erase ") + ) + assert len(uboot_tftp) == 2 + assert len(uboot_crc) >= 2 + assert ( + uboot_tftp[0] + < uboot_crc[0] + < uboot_tftp[1] + < uboot_crc[1] + < first_erase + ) + expected_mtdparts = nor_mtdparts(16) # Installer-only PHY override must be established before the first TFTP. diff --git a/tests/test_install_nor_unlock.py b/tests/test_install_nor_unlock.py index ed8ee00..f8ea177 100644 --- a/tests/test_install_nor_unlock.py +++ b/tests/test_install_nor_unlock.py @@ -40,7 +40,15 @@ async def close(self) -> None: self.closed = True -async def _run_env_only_install(monkeypatch, tmp_path, unlock_response: str): +async def _run_env_only_install( + monkeypatch, + tmp_path, + unlock_response: str, + *, + probe_response: str = 'Spi(cs1): Block:64KB Chip:8MB Name:"XT25F64B"\nOpenIPC # ', + download_mode: bool = False, + download_unlock_ok: bool = True, +): import defib.flashdump import defib.recovery.session import defib.transport.serial_platform @@ -59,7 +67,12 @@ async def _run_env_only_install(monkeypatch, tmp_path, unlock_response: str): uboot = tmp_path / "u-boot.bin" uboot.write_bytes(b"U" * 1024) - transport = ShellTransport() + class DownloadModeTransport(ShellTransport): + async def write(self, data: bytes) -> None: + if b"\x03" in data: + self.rx.extend(b"start download process.\n") + + transport = DownloadModeTransport() if download_mode else ShellTransport() commands: list[str] = [] class FakeRecoverySession: @@ -80,7 +93,7 @@ async def fake_send_command( assert transport_obj is transport commands.append(command) if command == "sf probe 0": - return 'Spi(cs1): Block:64KB Chip:8MB Name:"XT25F64B"\nOpenIPC # ' + return probe_response if command == "sf lock 0": return unlock_response if command == "printenv ethaddr": @@ -90,6 +103,32 @@ async def fake_send_command( return "OpenIPC # " monkeypatch.setattr(defib.flashdump, "send_command", fake_send_command) + + import defib.protocol.download_cmd + + class FakeDownloadCommandClient: + def __init__(self, transport_obj) -> None: + assert transport_obj is transport + + async def send_command( + self, command: str, timeout: float = 0.0 + ) -> tuple[bool, str]: + commands.append(command) + if command == "sf probe 0": + return True, probe_response + if command == "sf lock 0": + return download_unlock_ok, unlock_response + if command == "printenv ethaddr": + return True, "ethaddr=00:12:41:8e:c6:0e\n" + if command == "saveenv": + return True, "Saving Environment to SPI Flash... done\n" + return True, "" + + monkeypatch.setattr( + defib.protocol.download_cmd, + "DownloadCommandClient", + FakeDownloadCommandClient, + ) monkeypatch.setattr( defib.recovery.session, "RecoverySession", @@ -163,3 +202,86 @@ async def test_nor_unlock_failure_stops_before_persistent_write(monkeypatch, tmp assert "sf lock 0" in commands assert "saveenv" not in commands assert transport.closed is True + + +@pytest.mark.asyncio +async def test_nor_probe_failure_stops_before_unlock_with_size_override( + monkeypatch, tmp_path +): + run_install, request, commands, transport = await _run_env_only_install( + monkeypatch, + tmp_path, + "OpenIPC # ", + probe_response="No SPI flash selected. Please run `sf probe'\nOpenIPC # ", + ) + + with pytest.raises(typer.Exit) as exc_info: + await run_install(request) + + assert exc_info.value.exit_code == 1 + assert commands == ["sf probe 0"] + assert transport.closed is True + + +@pytest.mark.asyncio +async def test_download_command_failure_stops_before_persistent_write( + monkeypatch, tmp_path +): + run_install, request, commands, transport = await _run_env_only_install( + monkeypatch, + tmp_path, + "", + download_mode=True, + download_unlock_ok=False, + ) + + with pytest.raises(typer.Exit) as exc_info: + await run_install(request) + + assert exc_info.value.exit_code == 1 + assert "sf probe 0" in commands + assert "sf lock 0" in commands + assert "saveenv" not in commands + assert transport.closed is True + + + +@pytest.mark.asyncio +async def test_final_reset_does_not_require_prompt(monkeypatch, tmp_path): + from dataclasses import replace + + import defib.flashdump + + run_install, request, commands, transport = await _run_env_only_install( + monkeypatch, + tmp_path, + "OpenIPC # ", + ) + + async def prompt_sensitive_send_command( + transport_obj, + command: str, + timeout: float = 0.0, + wait_for: str | None = None, + **kwargs, + ) -> str: + assert transport_obj is transport + commands.append(command) + if command == "sf probe 0": + return 'Spi(cs1): Block:64KB Chip:8MB Name:"XT25F64B"\nOpenIPC # ' + if command == "reset": + assert wait_for is None + return "resetting...\n" + return "OpenIPC # " + + monkeypatch.setattr( + defib.flashdump, + "send_command", + prompt_sensitive_send_command, + ) + reset_request = replace(request, stages=("reset",)) + + await run_install(reset_request) + + assert commands == ["sf probe 0", "reset"] + assert transport.closed is True diff --git a/tests/test_network.py b/tests/test_network.py index f238b04..6ba3fc4 100644 --- a/tests/test_network.py +++ b/tests/test_network.py @@ -130,6 +130,27 @@ def test_blocksize_option(self): pkt = transport.sent[1][0] assert len(pkt[4:]) == 1024 # Negotiated blocksize + def test_blocksize_cap_is_explicitly_negotiated(self): + """A runtime cap should OACK the smaller block size requested by policy.""" + file_data = b"\x00" * 2048 + protocol = TFTPServerProtocol(file_data) + protocol.set_max_blocksize(DEFAULT_BLOCKSIZE) + transport = MockDatagramTransport() + protocol.connection_made(transport) + addr = ("10.0.0.1", 1234) + + rrq = struct.pack("!H", OPCODE_RRQ) + rrq += b"fw.bin\x00octet\x00blksize\x001468\x00" + protocol.datagram_received(rrq, addr) + + oack = transport.sent[0][0] + assert struct.unpack("!H", oack[:2])[0] == 6 + assert b"blksize\x00512\x00" in oack + + protocol.datagram_received(struct.pack("!HH", OPCODE_ACK, 0), addr) + data = transport.sent[1][0] + assert len(data[4:]) == DEFAULT_BLOCKSIZE + def test_error_response(self): """Client error should be handled gracefully.""" protocol = TFTPServerProtocol(b"data") diff --git a/tests/test_uart_command_integrity.py b/tests/test_uart_command_integrity.py index 52ba7ef..d62f54d 100644 --- a/tests/test_uart_command_integrity.py +++ b/tests/test_uart_command_integrity.py @@ -87,3 +87,24 @@ async def test_echo_verification_refuses_unacknowledged_command() -> None: wait_for="# ", verify_echo=True, ) + + + +class PartialResponseNoPromptTransport(MockTransport): + async def write(self, data: bytes) -> None: + await super().write(data) + if b"\r" in data: + self.enqueue_rx(b"Bytes transferred = 1558411\n") + + +@pytest.mark.asyncio +async def test_wait_for_prompt_rejects_partial_command_response() -> None: + transport = PartialResponseNoPromptTransport(flush_clears_buffer=False) + + with pytest.raises(TransportTimeout, match="waiting for '# '"): + await send_command( + transport, + "tftpboot 0x82000000 k", + timeout=0.05, + wait_for="# ", + ) From a1702a510bb9ac9b9eaa2698fecfe44071c03bbe Mon Sep 17 00:00:00 2001 From: ArthurKoba <61914203+ArthurKoba@users.noreply.github.com> Date: Fri, 18 Sep 2026 02:15:38 +0300 Subject: [PATCH 3/4] install: add explicit rootfs_data wipe option Generic NOR installs currently preserve rootfs_data, while registered stock U-Boot migrations erase it as part of their migration flow. Keep both existing defaults unchanged. Add --wipe-rootfs-data as an explicit destructive option for operators who want a clean persistent overlay during an otherwise normal install. The flag is independent of --stage selection: requesting it always performs the NOR-tail erase and the existing readback/CRC verification before the install continues. Reject the option on NAND, where this NOR rootfs_data layout does not apply, and reject the contradictory combination with --skip-stage rootfs-data before any device access. Treat an explicit wipe as a persistent operation in the vendor-chainload partial-install guard and in the SPI NOR unlock gate, so a wipe cannot bypass the protection clearing added by the preceding install hardening. Document the preserved defaults and the opt-in wipe, and add focused coverage for the legacy generic rootfs-data stage behavior, explicit wipe execution, unlock/erase/verification ordering, NAND rejection, and conflicting CLI intent. Whether a full generic install should eventually wipe rootfs_data by default is a separate policy decision because changing that default would destroy existing overlay data. This commit deliberately does not make that behavior change. --- CLAUDE.md | 4 + README.md | 11 ++ src/defib/cli/app.py | 10 ++ src/defib/install/model.py | 1 + src/defib/install/orchestrator.py | 27 +++- tests/test_cli.py | 1 + tests/test_install_rootfs_data.py | 205 ++++++++++++++++++++++++++++++ 7 files changed, 257 insertions(+), 2 deletions(-) create mode 100644 tests/test_install_rootfs_data.py diff --git a/CLAUDE.md b/CLAUDE.md index 0bbb57a..bb4f0b1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -221,6 +221,10 @@ Safety properties already built in — do not re-derive or undo them: stage. Generic wipes can lose `ethaddr`; registered stock-U-Boot migrations capture and restore the factory MAC instead of falling back to OpenIPC's compiled-in `00:00:23:34:45:66`. +- Generic NOR installs preserve the persistent `rootfs_data` overlay unless + `--wipe-rootfs-data` is explicitly requested. The wipe is destructive and + must be CRC-verified after erase. Registered stock-U-Boot migrations retain + their existing cleanup behavior; do not silently change either default. - `install` defaults to the complete production stage plan. Development runs may use repeated `--stage` for an exact subset or repeated `--skip-stage` to subtract stages. The stage names are `uboot`, `kernel`, `rootfs`, diff --git a/README.md b/README.md index 6a05e10..9e3a2c4 100644 --- a/README.md +++ b/README.md @@ -96,6 +96,12 @@ The `env` stage on a stock-U-Boot NOR migration requires explicit and restores the captured factory `ethaddr`. Other selected stages can be run without wiping the environment. +Generic NOR installs preserve the existing `rootfs_data` overlay by default. +Add `--wipe-rootfs-data` when a clean persistent overlay is desired; Defib +erases the region and verifies the erased contents by CRC before continuing. +Registered stock-U-Boot migrations retain their existing `rootfs_data` +cleanup behavior. + The release U-Boot owns boot-critical hardware initialization such as DDR cold-init and RAM probing limits. Defib owns the layout it actually flashes: it detects NOR capacity, selects the standard OpenIPC 8/16/32 MiB layout, and @@ -126,6 +132,11 @@ defib install -c hi3518ev100:hiwatch-ds-i203 \ --uboot u-boot-hi3518ev100-ddr3-256m-universal.bin \ --wipe-env --stage uboot --stage env -p /dev/ttyUSB0 -d +# Generic NOR install with a clean persistent overlay. +defib install -c hi3516ev200 \ + --firmware openipc.hi3516ev200-nor-lite.tgz \ + --wipe-rootfs-data -p /dev/ttyUSB0 + # Full production plan except kernel/rootfs writes. defib install -c hi3518ev100:hiwatch-ds-i203 \ --firmware hi3518ev100_lite_hiwatch-ds-i203-nor.tgz \ diff --git a/src/defib/cli/app.py b/src/defib/cli/app.py index 25d39ae..643fa57 100644 --- a/src/defib/cli/app.py +++ b/src/defib/cli/app.py @@ -2179,6 +2179,15 @@ def install( "their captured factory ethaddr is restored." ), ), + wipe_rootfs_data: bool = typer.Option( + False, + "--wipe-rootfs-data", + help=( + "Erase and CRC-verify the persistent rootfs_data overlay on NOR. " + "Generic installs preserve it unless this flag is given; registered " + "stock-U-Boot migrations keep their existing migration behavior." + ), + ), final_reset: bool = typer.Option( True, "--final-reset/--no-final-reset", @@ -2234,6 +2243,7 @@ def install( nor_size=nor_size, nand=nand, wipe_env=wipe_env, + wipe_rootfs_data=wipe_rootfs_data, final_reset=final_reset, stages=tuple(stage or ()), skip_stages=tuple(skip_stage or ()), diff --git a/src/defib/install/model.py b/src/defib/install/model.py index 2e8a277..25c2894 100644 --- a/src/defib/install/model.py +++ b/src/defib/install/model.py @@ -75,6 +75,7 @@ class InstallRequest: nor_size: int = 0 nand: bool = False wipe_env: bool = False + wipe_rootfs_data: bool = False final_reset: bool = True stages: tuple[str, ...] = () skip_stages: tuple[str, ...] = () diff --git a/src/defib/install/orchestrator.py b/src/defib/install/orchestrator.py index 538f3c6..08ff05f 100644 --- a/src/defib/install/orchestrator.py +++ b/src/defib/install/orchestrator.py @@ -77,6 +77,7 @@ async def run_install(request: InstallRequest) -> None: nor_size = request.nor_size nand = request.nand wipe_env = request.wipe_env + wipe_rootfs_data = request.wipe_rootfs_data final_reset = request.final_reset tftp_via = request.tftp_via output = request.output @@ -112,6 +113,19 @@ def warn(message: str) -> None: if stage_error is not None: fail(stage_error, exit_code=2) + skipped_stage_set = { + stage.strip().lower() + for stage in request.skip_stages + if stage.strip() + } + if wipe_rootfs_data and "rootfs-data" in skipped_stage_set: + fail( + "--wipe-rootfs-data conflicts with --skip-stage rootfs-data", + exit_code=2, + ) + if wipe_rootfs_data and nand: + fail("--wipe-rootfs-data is only supported for NOR installs", exit_code=2) + stage_set = set(stages) needs_tftp = bool(stage_set & {"uboot", "kernel", "rootfs"}) @@ -454,6 +468,8 @@ def _bootstrap_progress(event: ProgressEvent) -> None: if vendor_chainloaded: partial_persistent = stage_set & {"kernel", "rootfs", "rootfs-data", "env"} + if wipe_rootfs_data: + partial_persistent.add("rootfs-data") if partial_persistent and "uboot" not in stage_set: await transport.close() if power_controller: @@ -715,7 +731,7 @@ async def _set_env_verified_or_fail(key: str, value: str) -> None: await close_and_fail(f"Rootfs too large: {len(rootfs_data)} > {r_sz}") persistent_nor_stages = {"uboot", "kernel", "rootfs", "rootfs-data", "env"} - if stage_set & persistent_nor_stages: + if stage_set & persistent_nor_stages or wipe_rootfs_data: unlock_ok, unlock_resp = await _cmd_result( "sf lock 0", timeout=5.0, @@ -1186,7 +1202,14 @@ async def tftp_and_flash( "rootfs", tftp_alias["rootfs"], rootfs_data, r_off, r_sz ) - if "rootfs-data" in stage_set and has_stock_uboot and not nand: + erase_rootfs_data = ( + not nand + and ( + wipe_rootfs_data + or ("rootfs-data" in stage_set and has_stock_uboot) + ) + ) + if erase_rootfs_data: data_offset = r_off + r_sz data_size = nor_size * 1024 * 1024 - data_offset if data_size <= 0: diff --git a/tests/test_cli.py b/tests/test_cli.py index 4110d89..1a4c66a 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -103,6 +103,7 @@ def test_install_help_documents_stage_selection(self): assert "--stage" in out assert "--skip-stage" in out assert "rootfs-data" in out + assert "--wipe-rootfs-data" in out def test_restore_help_documents_poe_port(self): result = runner.invoke(app, ["restore", "--help"]) diff --git a/tests/test_install_rootfs_data.py b/tests/test_install_rootfs_data.py new file mode 100644 index 0000000..854ab42 --- /dev/null +++ b/tests/test_install_rootfs_data.py @@ -0,0 +1,205 @@ +from __future__ import annotations + +import io +import tarfile + +import pytest +import typer + +from defib.install import InstallRequest +from defib.install.layout import erased_region_crc +from defib.recovery.events import RecoveryResult +from defib.transport.base import Transport, TransportTimeout + + +class ShellTransport(Transport): + def __init__(self) -> None: + self.rx = bytearray() + self.closed = False + + async def read(self, size: int, timeout: float | None = None) -> bytes: + if not self.rx: + raise TransportTimeout("no data") + data = bytes(self.rx[:size]) + del self.rx[:size] + return data + + async def write(self, data: bytes) -> None: + if b"\x03" in data: + self.rx.extend(b"OpenIPC # ") + + async def flush_input(self) -> None: + self.rx.clear() + + async def flush_output(self) -> None: + return None + + async def bytes_waiting(self) -> int: + return len(self.rx) + + async def close(self) -> None: + self.closed = True + + +async def _prepare_generic_install(monkeypatch, tmp_path): + import defib.flashdump + import defib.recovery.session + import defib.transport.serial_platform + from defib.install.orchestrator import run_install + + firmware_tar = tmp_path / "firmware.tgz" + with tarfile.open(firmware_tar, "w:gz") as archive: + for name, payload in ( + ("uImage.hi3516ev200", b"K" * 1024), + ("rootfs.squashfs.hi3516ev200", b"R" * 2048), + ): + info = tarfile.TarInfo(name) + info.size = len(payload) + archive.addfile(info, io.BytesIO(payload)) + + uboot = tmp_path / "u-boot.bin" + uboot.write_bytes(b"U" * 1024) + + transport = ShellTransport() + commands: list[str] = [] + expected_crc = erased_region_crc(0xB0000) + + class FakeRecoverySession: + def __init__(self, *args, **kwargs) -> None: + pass + + async def run(self, transport_obj, **kwargs): + assert transport_obj is transport + return RecoveryResult(success=True) + + async def fake_create_transport(port: str): + assert port == "COM15" + return transport + + async def fake_send_command( + transport_obj, command: str, timeout: float = 0.0, **kwargs + ) -> str: + assert transport_obj is transport + commands.append(command) + if command == "sf probe 0": + return 'Spi(cs1): Block:64KB Chip:8MB Name:"XT25F64B"\nOpenIPC # ' + if command == "sf lock 0": + return "OpenIPC # " + if command == "sf erase 0x750000 0xb0000": + return "Erasing at 0x800000 -- 100% complete.\nOpenIPC # " + if command == ( + "sf read 0x42000000 0x750000 0xb0000; " + "crc32 0x42000000 0xb0000" + ): + return f"==> {expected_crc:08X}\nOpenIPC # " + if command == "reset": + return "resetting..." + return "OpenIPC # " + + monkeypatch.setattr(defib.flashdump, "send_command", fake_send_command) + monkeypatch.setattr( + defib.recovery.session, + "RecoverySession", + FakeRecoverySession, + ) + monkeypatch.setattr( + defib.transport.serial_platform, + "create_transport", + fake_create_transport, + ) + monkeypatch.setattr( + defib.transport.serial_platform, + "normalize_port_name", + lambda port: port, + ) + + def request(**kwargs) -> InstallRequest: + return InstallRequest( + chip="hi3516ev200", + firmware_path=str(firmware_tar), + uboot_path=str(uboot), + port="COM15", + nor_size=8, + output="json", + **kwargs, + ) + + return run_install, request, commands, transport + + +@pytest.mark.asyncio +async def test_generic_rootfs_data_stage_keeps_existing_noop_behavior( + monkeypatch, tmp_path +): + run_install, request, commands, transport = await _prepare_generic_install( + monkeypatch, tmp_path + ) + + await run_install(request(stages=("rootfs-data",))) + + assert "sf erase 0x750000 0xb0000" not in commands + assert not any(" 0x750000 0xb0000" in command for command in commands) + assert transport.closed is True + + +@pytest.mark.asyncio +async def test_wipe_rootfs_data_flag_erases_and_verifies_without_stage_dependency( + monkeypatch, tmp_path +): + run_install, request, commands, transport = await _prepare_generic_install( + monkeypatch, tmp_path + ) + + await run_install( + request( + stages=("reset",), + wipe_rootfs_data=True, + ) + ) + + unlock = "sf lock 0" + erase = "sf erase 0x750000 0xb0000" + verify = "sf read 0x42000000 0x750000 0xb0000; crc32 0x42000000 0xb0000" + assert unlock in commands + assert erase in commands + assert verify in commands + assert commands.index(unlock) < commands.index(erase) < commands.index(verify) + assert commands.index(verify) < commands.index("reset") + assert transport.closed is True + + +@pytest.mark.asyncio +async def test_wipe_rootfs_data_rejects_nand_before_transport(monkeypatch, tmp_path): + run_install, request, commands, _transport = await _prepare_generic_install( + monkeypatch, tmp_path + ) + + with pytest.raises(typer.Exit) as exc_info: + await run_install( + request( + stages=("reset",), + wipe_rootfs_data=True, + nand=True, + ) + ) + + assert exc_info.value.exit_code == 2 + assert commands == [] + + +@pytest.mark.asyncio +async def test_wipe_rootfs_data_conflicts_with_skip_stage(monkeypatch, tmp_path): + run_install, request, commands, _transport = await _prepare_generic_install( + monkeypatch, tmp_path + ) + + with pytest.raises(typer.Exit) as exc_info: + await run_install( + request( + skip_stages=("rootfs-data",), + wipe_rootfs_data=True, + ) + ) + + assert exc_info.value.exit_code == 2 + assert commands == [] From 1e11c83820785d10bde780900cf397a543074003 Mon Sep 17 00:00:00 2001 From: ArthurKoba <61914203+ArthurKoba@users.noreply.github.com> Date: Tue, 22 Sep 2026 10:35:28 +0300 Subject: [PATCH 4/4] install: address NOR install review findings Follow up on review of the existing three-commit NOR install series without rewriting its history. Keep flashdump.send_command backward-compatible by making strict prompt completion opt-in to install. Restore and dump-flash therefore retain their lenient prompt semantics, while install preserves explicit shell/download completion status and fails closed where persistent writes depend on it. Reject real SPI probe failures such as "Failed to initialize SPI flash", centralize sf lock result classification, and re-probe/re-unlock after the in-session U-Boot reset before saveenv writes. Treat optional printenv misses as semantic absence rather than transport failure, preserving the generic rescue-MAC path and explicit environment verification in download-command mode. Share one U-Boot TFTP command sequencer between lenient and strict callers without sharing transport policy. Scope the 512-byte host fallback to the retry that needs it, restore normal negotiation after success, scale CRC timeout with payload size, and retry CRC timeout/incomplete output. NAND targets whose older U-Boot genuinely lacks crc32 retain compatibility with a single warning and TFTP completion/size checks; timeouts, malformed checksums, and mismatches remain fatal. Make --wipe-rootfs-data consistent with exact --stage plans and cover its stock-U-Boot chainload guard. Remove unreachable verification code and compute the expected RAM CRC inside the verifier. Add regression coverage for the review findings: strict-vs-legacy prompt behavior, promptless reset, dump CRC capability probing, real sf probe failure, shell/download result contracts, optional environment reads, shared TFTP fallback classification and scope, CRC retry modes, NAND without crc32, post-reset SPI reinitialization, and rootfs_data destructive-path guards. --- CLAUDE.md | 4 +- README.md | 4 +- src/defib/cli/app.py | 5 +- src/defib/flashdump.py | 58 ++++-- src/defib/install/firmware.py | 21 +- src/defib/install/layout.py | 31 ++- src/defib/install/orchestrator.py | 293 +++++++++++++++++++-------- src/defib/uboot_tftp.py | 69 +++++++ tests/test_ds_i203_final_contract.py | 47 ++++- tests/test_flashdump.py | 12 ++ tests/test_install_flash_helpers.py | 86 ++++++++ tests/test_install_nor_unlock.py | 103 +++++++++- tests/test_install_rootfs_data.py | 79 +++++++- tests/test_nand_install.py | 168 +++++++++++++++ tests/test_uart_command_integrity.py | 35 ++++ 15 files changed, 870 insertions(+), 145 deletions(-) create mode 100644 src/defib/uboot_tftp.py diff --git a/CLAUDE.md b/CLAUDE.md index bb4f0b1..3c6ebf9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -223,7 +223,9 @@ Safety properties already built in — do not re-derive or undo them: compiled-in `00:00:23:34:45:66`. - Generic NOR installs preserve the persistent `rootfs_data` overlay unless `--wipe-rootfs-data` is explicitly requested. The wipe is destructive and - must be CRC-verified after erase. Registered stock-U-Boot migrations retain + must be CRC-verified after erase. Exact `--stage` plans must include + `rootfs-data` when the wipe flag is used, and `--skip-stage rootfs-data` + conflicts with an explicit wipe. Registered stock-U-Boot migrations retain their existing cleanup behavior; do not silently change either default. - `install` defaults to the complete production stage plan. Development runs may use repeated `--stage` for an exact subset or repeated `--skip-stage` to diff --git a/README.md b/README.md index 9e3a2c4..82f4ad2 100644 --- a/README.md +++ b/README.md @@ -99,7 +99,9 @@ without wiping the environment. Generic NOR installs preserve the existing `rootfs_data` overlay by default. Add `--wipe-rootfs-data` when a clean persistent overlay is desired; Defib erases the region and verifies the erased contents by CRC before continuing. -Registered stock-U-Boot migrations retain their existing `rootfs_data` +When an exact `--stage` plan is used, that plan must include +`--stage rootfs-data`; `--skip-stage rootfs-data` conflicts with an explicit +wipe. Registered stock-U-Boot migrations retain their existing `rootfs_data` cleanup behavior. The release U-Boot owns boot-critical hardware initialization such as DDR diff --git a/src/defib/cli/app.py b/src/defib/cli/app.py index 643fa57..5fb6300 100644 --- a/src/defib/cli/app.py +++ b/src/defib/cli/app.py @@ -2184,8 +2184,9 @@ def install( "--wipe-rootfs-data", help=( "Erase and CRC-verify the persistent rootfs_data overlay on NOR. " - "Generic installs preserve it unless this flag is given; registered " - "stock-U-Boot migrations keep their existing migration behavior." + "Generic installs preserve it unless this flag is given; with exact " + "--stage selection include --stage rootfs-data. Registered stock-U-Boot " + "migrations keep their existing migration behavior." ), ), final_reset: bool = typer.Option( diff --git a/src/defib/flashdump.py b/src/defib/flashdump.py index c5107f4..1299887 100644 --- a/src/defib/flashdump.py +++ b/src/defib/flashdump.py @@ -23,6 +23,7 @@ from typing import Callable from defib.transport.base import Transport, TransportTimeout +from defib.uboot_tftp import run_uboot_tftp logger = logging.getLogger(__name__) @@ -269,6 +270,7 @@ async def send_command( timeout: float = 5.0, wait_for: str | None = None, verify_echo: bool = False, + require_prompt: bool = False, ) -> str: """Send a command to U-Boot and collect the response. @@ -276,6 +278,11 @@ async def send_command( the command is entered one character at a time and U-Boot must echo every byte correctly before the final carriage return is sent. This prevents a corrupted ``sf erase/write/read`` line from ever being executed. + + Historically callers using ``wait_for`` received the partial buffer when the + deadline expired. Keep that behavior by default because dump/restore probes + use timeout as capability information. Installer writes opt into strict prompt + completion with ``require_prompt=True``. """ # Clear any pending input. try: @@ -317,7 +324,7 @@ async def send_command( continue response = buf.decode("ascii", errors="replace") - if wait_for: + if wait_for and require_prompt: partial = response.strip()[-200:] or "" raise TransportTimeout( f"Timed out waiting for {wait_for!r} after U-Boot command {cmd!r}; " @@ -332,22 +339,24 @@ async def tftp_to_ram( filename: str, timeout: float = 120.0, ) -> str: - """Download a file via TFTP into RAM. - - Tries ``tftpboot`` first; falls back to ``tftp`` if the U-Boot - build doesn't have the ``tftpboot`` alias. - - Returns the command response text. Raises RuntimeError on failure. - """ - cmd = f"tftpboot 0x{addr:x} {filename}" - resp = await send_command(transport, cmd, timeout=timeout, wait_for="# ") - if "unknown command" in resp.lower(): - logger.debug("tftpboot not available, falling back to tftp") - cmd = f"tftp 0x{addr:x} {filename}" - resp = await send_command(transport, cmd, timeout=timeout, wait_for="# ") - if "done" not in resp.lower() and "bytes transferred" not in resp.lower(): - raise RuntimeError(f"TFTP download failed: {resp.strip()[-200:]}") - return resp + """Download a file via TFTP into RAM using legacy lenient prompt handling.""" + + async def run_command(command: str, command_timeout: float) -> tuple[bool, str]: + response = await send_command( + transport, + command, + timeout=command_timeout, + wait_for="# ", + ) + return True, response + + return await run_uboot_tftp( + run_command, + filename, + addr, + use_loadaddr=False, + timeout=timeout, + ) def detect_flash_from_text(text: str) -> int | None: @@ -428,11 +437,18 @@ async def detect_flash( async def _detect_crc32(transport: Transport) -> bool: - """Check if U-Boot has the crc32 command.""" + """Check if U-Boot has the crc32 command without making backup depend on it.""" resp = await send_command(transport, "crc32 0 0", timeout=3.0, wait_for="# ") - # If crc32 exists, it will output a CRC value or usage. - # If not, it will say "Unknown command" - return "unknown command" not in resp.lower() + text = resp.lower() + if "unknown command" in text: + return False + # A supported crc32 command prints either a checksum or usage/help text. + # No/partial response is inconclusive, so keep dump-flash usable without + # per-block CRC verification rather than treating silence as support. + return bool( + re.search(r"==>\s*[0-9a-f]{8}|crc32 for", text) + or ("usage:" in text and "crc32" in text) + ) async def _get_device_crc32( diff --git a/src/defib/install/firmware.py b/src/defib/install/firmware.py index 389e26d..1395f1b 100644 --- a/src/defib/install/firmware.py +++ b/src/defib/install/firmware.py @@ -7,6 +7,8 @@ from dataclasses import dataclass from pathlib import Path +from defib.uboot_tftp import uboot_tftp_commands as uboot_tftp_commands + @dataclass(frozen=True) class FirmwareBundle: @@ -18,25 +20,6 @@ class FirmwareBundle: rootfs: bytes -def uboot_tftp_commands( - filename: str, - ram_addr: int, - *, - use_loadaddr: bool, -) -> tuple[str, str]: - """Return primary/fallback U-Boot TFTP commands for one staged file. - - Vendor-U-Boot migrations deliberately use ``loadaddr`` so the command - line stays short on fragile legacy UART consoles. Generic boot-ROM and - download-command installs retain the historical explicit RAM address and - therefore do not depend on environment read-back formatting. - """ - if use_loadaddr: - return f"tftpboot {filename}", f"tftp {filename}" - address = f"0x{ram_addr:x}" - return f"tftpboot {address} {filename}", f"tftp {address} {filename}" - - def load_firmware_bundle(path: str | Path) -> FirmwareBundle: """Read kernel/rootfs and verify any matching md5sum entries in one pass.""" kernel_name = "" diff --git a/src/defib/install/layout.py b/src/defib/install/layout.py index 9b974ec..b6a4a80 100644 --- a/src/defib/install/layout.py +++ b/src/defib/install/layout.py @@ -88,7 +88,9 @@ def uboot_flash_command_error(response: str) -> str | None: if not line: continue text = line.lower() - if text.startswith(("error:", "unknown command", "usage: sf", "usage: nand")): + if text.startswith(("error:", "unknown command", "usage:")): + return line + if re.match(r"failed\b", text): return line if "no spi flash selected" in text: return line @@ -101,6 +103,33 @@ def uboot_flash_command_error(response: str) -> str | None: return None +def uboot_sf_lock_unsupported(response: str) -> bool: + """Return True only when the U-Boot build lacks a usable sf lock command.""" + + lines = [line.strip().lower() for line in response.splitlines() if line.strip()] + for line in lines: + mentions_sf_lock = "sf" in line or "lock" in line + if mentions_sf_lock and "unknown command" in line: + return True + if mentions_sf_lock and ("not supported" in line or "unsupported" in line): + return True + + # Older U-Boot variants often print the whole sf usage table for an unknown + # subcommand. That is compatibility information only when the table itself + # contains no sf lock entry. If sf lock is listed, a Usage response means our + # invocation was rejected and must remain a hard failure. + usage_seen = any(line.startswith("usage:") for line in lines) + sf_usage: list[str] = [] + for line in lines: + candidate = line.removeprefix("usage:").strip() if line.startswith("usage:") else line + if re.match(r"^sf(?:\s|$)", candidate): + sf_usage.append(candidate) + if usage_seen and sf_usage: + return not any(re.match(r"^sf\s+lock(?:\s|$)", line) for line in sf_usage) + + return False + + async def set_uboot_env_verified( cmd: Command, key: str, diff --git a/src/defib/install/orchestrator.py b/src/defib/install/orchestrator.py index 08ff05f..9ad2fa0 100644 --- a/src/defib/install/orchestrator.py +++ b/src/defib/install/orchestrator.py @@ -13,7 +13,7 @@ import typer -from defib.install.firmware import load_firmware_bundle, uboot_tftp_commands +from defib.install.firmware import load_firmware_bundle from defib.install.layout import ( NAND_LAYOUT, NOR8M_LAYOUT, @@ -27,9 +27,11 @@ select_nor_size_mb, set_uboot_env_verified, uboot_flash_command_error, + uboot_sf_lock_unsupported, verify_spi_environment_crc, ) from defib.install.model import InstallRequest, resolve_install_stages +from defib.uboot_tftp import run_uboot_tftp TFTP_RAM_VERIFY_RETRIES = 1 @@ -53,7 +55,11 @@ async def run_install(request: InstallRequest) -> None: ) from defib.flashdump import get_ram_staging_addr, send_command from defib.network.ip_manager import list_interfaces_async, temporary_ip - from defib.network.tftp_server import DEFAULT_BLOCKSIZE, start_tftp_server + from defib.network.tftp_server import ( + DEFAULT_BLOCKSIZE, + MAX_BLOCKSIZE, + start_tftp_server, + ) from defib.profiles.loader import recovery_mode from defib.recovery.events import LogEvent, ProgressEvent from defib.recovery.session import RecoverySession @@ -113,11 +119,21 @@ def warn(message: str) -> None: if stage_error is not None: fail(stage_error, exit_code=2) + selected_stage_set = { + stage.strip().lower() + for stage in request.stages + if stage.strip() + } skipped_stage_set = { stage.strip().lower() for stage in request.skip_stages if stage.strip() } + if wipe_rootfs_data and selected_stage_set and "rootfs-data" not in selected_stage_set: + fail( + "--wipe-rootfs-data with --stage requires --stage rootfs-data", + exit_code=2, + ) if wipe_rootfs_data and "rootfs-data" in skipped_stage_set: fail( "--wipe-rootfs-data conflicts with --skip-stage rootfs-data", @@ -584,6 +600,8 @@ async def _cmd_result( try: ok, out = await dl_client.send_command(cmd, timeout=timeout) except TransportError as exc: + if allow_failure: + return False, str(exc) await close_and_fail( f"U-Boot transport failed while running {cmd!r}: {exc}" ) @@ -622,12 +640,19 @@ async def _cmd_result( timeout=timeout, wait_for="# ", verify_echo=verify_shell_echo, + require_prompt=True, ) except TransportError as exc: + if allow_failure: + return False, str(exc) await close_and_fail( f"U-Boot transport failed while running {cmd!r}: {exc}" ) - if "unknown command" not in out.lower() or attempt == 1: + if ( + allow_failure + or "unknown command" not in out.lower() + or attempt == 1 + ): return True, out await transport.write(b"\x03\r") await _aio.sleep(0.05) @@ -638,6 +663,60 @@ async def _cmd(cmd: str, timeout: float = 60.0, **kw: object) -> str: _, out = await _cmd_result(cmd, timeout=timeout) return out + async def _optional_printenv(key: str, timeout: float = 5.0) -> str: + """Read an optional env key without treating "not defined" as fatal.""" + + ok, out = await _cmd_result( + f"printenv {key}", + timeout=timeout, + allow_failure=True, + ) + if ok: + return out + text = out.lower() + if key.lower() in text and "not defined" in text: + return out + detail = out.strip()[-200:] or "" + await close_and_fail( + f"U-Boot command failed or timed out while reading optional {key!r}: " + f"{detail}" + ) + + async def _unlock_nor_or_fail() -> None: + """Clear SPI NOR write protection without guessing command completion.""" + + unlock_ok, unlock_resp = await _cmd_result( + "sf lock 0", + timeout=5.0, + allow_failure=True, + ) + if not unlock_ok: + if download_mode and uboot_sf_lock_unsupported(unlock_resp): + warn( + "U-Boot does not expose a usable `sf lock` command; " + "continuing and relying on erase/write result checks." + ) + return + detail = unlock_resp.strip()[-200:] or "" + await close_and_fail( + f"SPI NOR unlock command failed or timed out: {detail}" + ) + + if uboot_sf_lock_unsupported(unlock_resp): + warn( + "U-Boot does not expose a usable `sf lock` command; " + "continuing and relying on erase/write result checks." + ) + return + + unlock_error = uboot_flash_command_error(unlock_resp) + if unlock_error: + await close_and_fail( + f"SPI NOR unlock failed ({unlock_error}): {unlock_resp.strip()}" + ) + if output == "human": + console.print(" [green]SPI NOR write protection cleared[/green]") + async def _reset_command(timeout: float) -> None: """Issue reset without requiring the old U-Boot prompt to return.""" if download_mode: @@ -732,36 +811,7 @@ async def _set_env_verified_or_fail(key: str, value: str) -> None: persistent_nor_stages = {"uboot", "kernel", "rootfs", "rootfs-data", "env"} if stage_set & persistent_nor_stages or wipe_rootfs_data: - unlock_ok, unlock_resp = await _cmd_result( - "sf lock 0", - timeout=5.0, - allow_failure=True, - ) - unlock_text = unlock_resp.lower() - unlock_unsupported = ( - "unknown command" in unlock_text - or "usage:" in unlock_text - or "not supported" in unlock_text - or "unsupported" in unlock_text - ) - if unlock_unsupported: - warn( - "U-Boot does not expose a usable `sf lock` command; " - "continuing and relying on erase/write result checks." - ) - elif not unlock_ok: - detail = unlock_resp.strip()[-200:] or "" - await close_and_fail( - f"SPI NOR unlock command failed or timed out: {detail}" - ) - else: - unlock_error = uboot_flash_command_error(unlock_resp) - if unlock_error: - await close_and_fail( - f"SPI NOR unlock failed ({unlock_error}): {unlock_resp.strip()}" - ) - if output == "human": - console.print(" [green]SPI NOR write protection cleared[/green]") + await _unlock_nor_or_fail() if output == "human": console.print( @@ -930,39 +980,54 @@ async def replace_in_tftp(name: str, data: bytes) -> None: ) async def _tftp_to_ram(filename: str, timeout: float = 120.0) -> str: - """TFTP download, preserving the generic explicit-address path.""" - tftpboot_cmd, tftp_cmd = uboot_tftp_commands( + """TFTP through the installer's strict status-preserving runner.""" + + async def run_command( + command: str, + command_timeout: float, + ) -> tuple[bool, str]: + return await _cmd_result( + command, + timeout=command_timeout, + allow_failure=True, + ) + + return await run_uboot_tftp( + run_command, filename, ram_addr, use_loadaddr=has_stock_uboot, - ) - ok, resp = await _cmd_result( - tftpboot_cmd, timeout=timeout, - allow_failure=True, ) - if "unknown command" in resp.lower(): - ok, resp = await _cmd_result( - tftp_cmd, - timeout=timeout, - allow_failure=True, - ) - if not ok: - detail = resp.strip()[-200:] or "" - raise RuntimeError(f"TFTP command failed or timed out: {detail}") - if "done" not in resp.lower() and "bytes transferred" not in resp.lower(): - raise RuntimeError(f"TFTP download failed: {resp.strip()[-200:]}") - return resp + + def _crc_timeout_for_size(size: int) -> float: + # Large extracted UBIFS payloads can be tens of MiB. Give slow + # ARM9-class U-Boot software CRC loops a size-scaled budget. + return max(10.0, (size / (1024 * 1024)) * 2.0) + + crc32_available: bool | None = None async def _verify_tftp_ram( name: str, tftp_name: str, orig_data: bytes, - expected_crc: int, - ) -> int: - """Verify TFTP staging before any persistent write.""" - attempts = TFTP_RAM_VERIFY_RETRIES + 1 - for attempt in range(attempts): + ) -> int | None: + """Verify TFTP staging before any persistent write. + + NAND keeps compatibility with older U-Boot builds that do not + provide crc32: transfer completion/size checks remain mandatory, + but the missing optional checksum capability is warned once and + does not make an otherwise supported NAND install impossible. + Timeouts, malformed CRC output, and mismatches still fail closed. + """ + + nonlocal crc32_available + + expected_crc = zlib.crc32(orig_data) & 0xFFFFFFFF + attempt = 0 + fallback_active = False + + while True: failure: str try: tftp_resp = await _tftp_to_ram(tftp_name, timeout=120.0) @@ -987,29 +1052,59 @@ async def _verify_tftp_ram( f"reported {reported_size} bytes, " f"expected {len(orig_data)}" ) + elif nand and crc32_available is False: + if fallback_active and tftp_protocol is not None: + tftp_protocol.set_max_blocksize(MAX_BLOCKSIZE) + return None else: - crc_resp = await _cmd( + crc_ok, crc_resp = await _cmd_result( f"crc32 0x{ram_addr:x} 0x{len(orig_data):x}", - timeout=10.0, + timeout=_crc_timeout_for_size(len(orig_data)), + allow_failure=True, + ) + crc_text = crc_resp.lower() + crc_unsupported = ( + "unknown command" in crc_text + and "crc32" in crc_text ) - ram_crc = parse_uboot_crc32(crc_resp) - if ram_crc is None: - failure = ( - "CRC response did not contain a complete " - f"checksum: {crc_resp.strip()[-120:]}" - ) - elif ram_crc == expected_crc: - return ram_crc + if nand and crc_unsupported: + if crc32_available is not False: + warn( + "U-Boot crc32 is unavailable on this NAND target; " + "continuing with TFTP completion/size checks only." + ) + crc32_available = False + if fallback_active and tftp_protocol is not None: + tftp_protocol.set_max_blocksize(MAX_BLOCKSIZE) + return None + + if not crc_ok: + detail = crc_resp.strip()[-120:] or "" + failure = f"CRC command failed or timed out: {detail}" else: - failure = ( - f"CRC expected={expected_crc:08X} " - f"got={ram_crc:08X}" - ) + ram_crc = parse_uboot_crc32(crc_resp) + if ram_crc is None: + failure = ( + "CRC response did not contain a complete " + f"checksum: {crc_resp.strip()[-120:]}" + ) + elif ram_crc == expected_crc: + crc32_available = True + if fallback_active and tftp_protocol is not None: + tftp_protocol.set_max_blocksize(MAX_BLOCKSIZE) + return ram_crc + else: + failure = ( + f"CRC expected={expected_crc:08X} " + f"got={ram_crc:08X}" + ) if attempt < TFTP_RAM_VERIFY_RETRIES: - next_attempt = attempt + 2 + attempt += 1 + next_attempt = attempt + 1 if tftp_protocol is not None: tftp_protocol.set_max_blocksize(DEFAULT_BLOCKSIZE) + fallback_active = True warn( f"Attempt {next_attempt}: fetching TFTP file " f"{tftp_name!r} again for {name} after RAM " @@ -1026,17 +1121,15 @@ async def _verify_tftp_ram( console.print( f"[red]{name} TFTP RAM verification failed after " - f"{attempts} attempt(s):[/red] {failure}" + f"{attempt + 1} attempt(s):[/red] {failure}" ) raise typer.Exit(1) - raise AssertionError("unreachable TFTP verification state") - async def tftp_and_flash( name: str, tftp_name: str, orig_data: bytes, flash_off: int, erase_sz: int, ) -> None: - """TFTP download, full-partition erase/write, and CRC verify.""" + """TFTP download, RAM validation, flash write, and readback verify.""" if output == "human": console.print( f"\n [bold]Flashing {name}[/bold] → 0x{flash_off:X} " @@ -1050,10 +1143,14 @@ async def tftp_and_flash( name, tftp_name, orig_data, - expected_crc, ) if output == "human": - console.print(f" TFTP CRC verified: {ram_crc:08X}") + if ram_crc is None: + console.print( + " TFTP transfer accepted; U-Boot CRC32 unavailable" + ) + else: + console.print(f" TFTP CRC verified: {ram_crc:08X}") # OpenIPC NOR layouts define fixed kernel/rootfs partitions. # Erase the whole partition so no stock filesystem tail survives @@ -1099,9 +1196,10 @@ async def tftp_and_flash( ) raise typer.Exit(1) - # Verify flash write by reading back and checking CRC. - # Skip for NAND — ECC/OOB makes raw read-back differ from - # the original data; the TFTP-to-RAM CRC above is sufficient. + # Verify NOR flash writes by reading back and checking CRC. + # Skip raw NAND readback because ECC/OOB changes the byte stream. + # NAND validates staged RAM by CRC when the command is available; + # older U-Boot falls back to TFTP completion/size checks above. if not nand: read_resp = await _cmd( f"{flash_cmd} read 0x{ram_addr:x} 0x{flash_off:x} 0x{len(orig_data):x}", @@ -1116,7 +1214,7 @@ async def tftp_and_flash( raise typer.Exit(1) resp = await _cmd( f"crc32 0x{ram_addr:x} 0x{len(orig_data):x}", - timeout=10.0, + timeout=_crc_timeout_for_size(len(orig_data)), ) flash_crc = parse_uboot_crc32(resp) if flash_crc is None: @@ -1165,17 +1263,20 @@ async def tftp_and_flash( ) await replace_in_tftp(tftp_alias["rootfs"], ubifs_data) - ubifs_crc = zlib.crc32(ubifs_data) & 0xFFFFFFFF verified_ubifs_crc = await _verify_tftp_ram( "rootfs (UBI)", tftp_alias["rootfs"], ubifs_data, - ubifs_crc, ) if output == "human": - console.print( - f" TFTP CRC verified: {verified_ubifs_crc:08X}" - ) + if verified_ubifs_crc is None: + console.print( + " TFTP transfer accepted; U-Boot CRC32 unavailable" + ) + else: + console.print( + f" TFTP CRC verified: {verified_ubifs_crc:08X}" + ) await _cmd(f"nand erase 0x{r_off:x} 0x{r_sz:x}", timeout=120.0) nand_name = "hinand" @@ -1265,7 +1366,7 @@ async def tftp_and_flash( # After that Defib re-applies only install invariants plus instance # identity; device policy remains the firmware/profile's responsibility. if "env" in stage_set and has_stock_uboot and not nand: - pre_reset_eth_resp = await _cmd("printenv ethaddr", timeout=5.0) + pre_reset_eth_resp = await _optional_printenv("ethaddr", timeout=5.0) pre_reset_eth = parse_printenv_value(pre_reset_eth_resp, "ethaddr") preserved_eth = preserved_stock_env.get("ethaddr") reset_eth, _ = select_install_ethaddr( @@ -1323,6 +1424,18 @@ async def tftp_and_flash( await _reset_command(timeout=1.0) await _wait_for_openipc_shell_after_reset() + + # Reset starts a new U-Boot instance. Reinitialize SPI state and + # clear protection again before the later saveenv write. + reprobe_resp = await _cmd("sf probe 0", timeout=5.0) + reprobe_error = uboot_flash_command_error(reprobe_resp) + if reprobe_error: + raise RuntimeError( + f"sf probe after env reset failed ({reprobe_error}): " + f"{reprobe_resp.strip()}" + ) + await _unlock_nor_or_fail() + if output == "human": console.print(" [green]OpenIPC U-Boot defaults loaded[/green]") @@ -1361,7 +1474,7 @@ async def tftp_and_flash( if "env" in stage_set: # Preserve a factory MAC captured from stock U-Boot. For normal # boot-ROM installs keep the existing generic rescue-MAC behavior. - eth_resp = await _cmd("printenv ethaddr", timeout=5.0) + eth_resp = await _optional_printenv("ethaddr", timeout=5.0) current_eth = parse_printenv_value(eth_resp, "ethaddr") preserved_eth = preserved_stock_env.get("ethaddr") selected_eth, eth_source = select_install_ethaddr( @@ -1410,7 +1523,7 @@ async def tftp_and_flash( ram_addr=ram_addr, ) - verify_resp = await _cmd("printenv ethaddr", timeout=5.0) + verify_resp = await _optional_printenv("ethaddr", timeout=5.0) saved_eth = parse_printenv_value(verify_resp, "ethaddr") if saved_eth is None or saved_eth.lower() != selected_eth.lower(): raise RuntimeError( @@ -1418,7 +1531,7 @@ async def tftp_and_flash( f"expected={selected_eth!r} got={saved_eth!r}" ) - verify_mtd_resp = await _cmd("printenv mtdparts", timeout=5.0) + verify_mtd_resp = await _optional_printenv("mtdparts", timeout=5.0) saved_mtdparts = parse_printenv_value(verify_mtd_resp, "mtdparts") expected_mtdparts = nor_mtdparts(nor_size) if saved_mtdparts != expected_mtdparts: diff --git a/src/defib/uboot_tftp.py b/src/defib/uboot_tftp.py new file mode 100644 index 0000000..be39f74 --- /dev/null +++ b/src/defib/uboot_tftp.py @@ -0,0 +1,69 @@ +"""Shared U-Boot TFTP command sequencing. + +The transport policy belongs to the caller: dump/restore can provide a lenient +runner, while install provides a strict status-preserving runner. This module +only owns command spelling, tftpboot→tftp fallback, and transfer-result parsing. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable + +CommandResultRunner = Callable[[str, float], Awaitable[tuple[bool, str]]] + + +def _reports_unknown_command(response: str, command_name: str) -> bool: + """Match U-Boot's Unknown command result for the command just attempted.""" + + expected = command_name.lower() + for raw_line in response.splitlines(): + line = raw_line.strip().lower() + if not line.startswith("unknown command"): + continue + if expected in line or line == "unknown command": + return True + return False + + +def uboot_tftp_commands( + filename: str, + ram_addr: int, + *, + use_loadaddr: bool, +) -> tuple[str, str]: + """Return primary/fallback U-Boot TFTP commands for one staged file.""" + + if use_loadaddr: + return f"tftpboot {filename}", f"tftp {filename}" + address = f"0x{ram_addr:x}" + return f"tftpboot {address} {filename}", f"tftp {address} {filename}" + + +async def run_uboot_tftp( + run_command: CommandResultRunner, + filename: str, + ram_addr: int, + *, + use_loadaddr: bool, + timeout: float = 120.0, +) -> str: + """Fetch one file into RAM using the caller's command-completion policy.""" + + primary, fallback = uboot_tftp_commands( + filename, + ram_addr, + use_loadaddr=use_loadaddr, + ) + ok, response = await run_command(primary, timeout) + if _reports_unknown_command(response, primary.split()[0]): + ok, response = await run_command(fallback, timeout) + + if not ok: + detail = response.strip()[-200:] or "" + raise RuntimeError(f"TFTP command failed or timed out: {detail}") + + text = response.lower() + if "done" not in text and "bytes transferred" not in text: + raise RuntimeError(f"TFTP download failed: {response.strip()[-200:]}") + + return response diff --git a/tests/test_ds_i203_final_contract.py b/tests/test_ds_i203_final_contract.py index 38e06f1..ae51727 100644 --- a/tests/test_ds_i203_final_contract.py +++ b/tests/test_ds_i203_final_contract.py @@ -651,7 +651,15 @@ async def fail_env_verify(*args, **kwargs): @pytest.mark.asyncio @pytest.mark.parametrize( "crc_failure", - [None, "tftp", "tftp-once", "readback", "env"], + [ + None, + "tftp", + "tftp-once", + "tftp-timeout-once", + "tftp-unparseable-once", + "readback", + "env", + ], ) async def test_ds_i203_stock_install_persists_detected_layout_but_not_camera_policy( monkeypatch, tmp_path, crc_failure, capsys @@ -859,6 +867,10 @@ async def fake_send_command( return "==> 00000000\nOpenIPC # " if crc_failure == "tftp-once" and crc_calls == 1: return "==> 00000000\nOpenIPC # " + if crc_failure == "tftp-timeout-once" and crc_calls == 1: + raise TransportTimeout("synthetic CRC prompt timeout") + if crc_failure == "tftp-unparseable-once" and crc_calls == 1: + return "CRC32 output truncated\nOpenIPC # " if crc_failure == "readback" and crc_calls == 2: return "CRC32 output truncated\nOpenIPC # " parts = command.split() @@ -937,7 +949,13 @@ async def fake_send_command( output="json", ) - if crc_failure in (None, "tftp-once"): + retry_success_cases = { + None, + "tftp-once", + "tftp-timeout-once", + "tftp-unparseable-once", + } + if crc_failure in retry_success_cases: await orchestrator.run_install(request) else: with pytest.raises(typer.Exit) as exc_info: @@ -955,10 +973,12 @@ async def fake_send_command( assert any(cmd.startswith("cmp.l 0x82000000 ") for cmd in commands) return - if crc_failure == "tftp-once": + if crc_failure in {"tftp-once", "tftp-timeout-once", "tftp-unparseable-once"}: + from defib.network.tftp_server import MAX_BLOCKSIZE + assert commands.count("tftpboot u") == 2 assert tftp_protocol_obj is not None - assert tftp_protocol_obj.blocksize_caps == [512] + assert tftp_protocol_obj.blocksize_caps == [512, MAX_BLOCKSIZE] warning_lines = [ line @@ -970,7 +990,12 @@ async def fake_send_command( assert warning["message"].startswith( "Attempt 2: fetching TFTP file 'u' again for U-Boot " ) - assert "CRC expected=" in warning["message"] + if crc_failure == "tftp-once": + assert "CRC expected=" in warning["message"] + elif crc_failure == "tftp-timeout-once": + assert "CRC command failed or timed out:" in warning["message"] + else: + assert "CRC response did not contain a complete checksum:" in warning["message"] assert "using 512-byte blocks." in warning["message"] uboot_tftp = [ @@ -998,6 +1023,18 @@ async def fake_send_command( < first_erase ) + first_internal_reset = commands.index("reset") + probe_indices = [i for i, command in enumerate(commands) if command == "sf probe 0"] + unlock_indices = [i for i, command in enumerate(commands) if command == "sf lock 0"] + assert len(probe_indices) >= 2 + assert len(unlock_indices) >= 2 + assert ( + first_internal_reset + < probe_indices[1] + < unlock_indices[1] + < commands.index("saveenv") + ) + expected_mtdparts = nor_mtdparts(16) # Installer-only PHY override must be established before the first TFTP. diff --git a/tests/test_flashdump.py b/tests/test_flashdump.py index 348af4a..85bbfa5 100644 --- a/tests/test_flashdump.py +++ b/tests/test_flashdump.py @@ -199,6 +199,18 @@ def test_hi3516cv610_ram_at_0x40(self): class TestCrc32Detection: """Test CRC32 command detection and parsing.""" + @pytest.mark.asyncio + async def test_crc32_probe_silence_degrades_to_unverified_dump(self): + from defib.flashdump import _detect_crc32 + + class SilentTransport(MockTransport): + async def write(self, data: bytes) -> None: + self._tx_log.append(data) + + transport = SilentTransport() + assert await _detect_crc32(transport) is False + + def test_parse_crc32_response(self): """U-Boot crc32 output: '... ==> abcd1234'.""" import re diff --git a/tests/test_install_flash_helpers.py b/tests/test_install_flash_helpers.py index 14aecb2..117ee7a 100644 --- a/tests/test_install_flash_helpers.py +++ b/tests/test_install_flash_helpers.py @@ -4,6 +4,7 @@ import pytest from defib.install.firmware import uboot_tftp_commands +from defib.uboot_tftp import run_uboot_tftp from defib.install.layout import ( align_up, detect_nor_size_mb, @@ -14,6 +15,7 @@ select_nor_size_mb, set_uboot_env_verified, uboot_flash_command_error, + uboot_sf_lock_unsupported, verify_spi_environment_crc, ) @@ -32,6 +34,68 @@ def test_vendor_tftp_uses_verified_loadaddr_for_short_command(): ) +def test_shared_tftp_runner_falls_back_from_tftpboot(): + calls: list[tuple[str, float]] = [] + + async def run_command(command: str, timeout: float) -> tuple[bool, str]: + calls.append((command, timeout)) + if command.startswith("tftpboot "): + return False, "Unknown command 'tftpboot'\nOpenIPC # " + return True, "Bytes transferred = 1024\nOpenIPC # " + + response = asyncio.run( + run_uboot_tftp( + run_command, + "k", + 0x82000000, + use_loadaddr=False, + timeout=17.0, + ) + ) + + assert "Bytes transferred = 1024" in response + assert calls == [ + ("tftpboot 0x82000000 k", 17.0), + ("tftp 0x82000000 k", 17.0), + ] + + +def test_shared_tftp_runner_preserves_non_fallback_failure_status(): + async def run_command(command: str, timeout: float) -> tuple[bool, str]: + return False, "network timeout" + + with pytest.raises(RuntimeError, match="failed or timed out"): + asyncio.run( + run_uboot_tftp( + run_command, + "k", + 0x82000000, + use_loadaddr=False, + ) + ) + + + +def test_shared_tftp_runner_ignores_unrelated_unknown_command(): + calls: list[str] = [] + + async def run_command(command: str, timeout: float) -> tuple[bool, str]: + calls.append(command) + return False, "Unknown command 'crc32'\n" + + with pytest.raises(RuntimeError, match="failed or timed out"): + asyncio.run( + run_uboot_tftp( + run_command, + "k", + 0x82000000, + use_loadaddr=False, + ) + ) + + assert calls == ["tftpboot 0x82000000 k"] + + def test_align_up_for_nand_page_write(): assert align_up(0x12345, 2048) == 0x12800 @@ -102,6 +166,28 @@ def test_uboot_flash_error_detects_missing_spi_probe(): assert uboot_flash_command_error(response) is not None +def test_uboot_flash_error_detects_failed_to_initialize_probe(): + response = "Failed to initialize SPI flash at 0:0 (error -2)\nOpenIPC # " + assert uboot_flash_command_error(response) is not None + + +def test_sf_lock_usage_without_lock_entry_is_unsupported(): + response = ( + "Usage:\n" + "sf probe [[bus:]cs] [hz] [mode]\n" + "sf read addr offset len\n" + "sf write addr offset len\n" + "OpenIPC # " + ) + assert uboot_sf_lock_unsupported(response) is True + + +def test_sf_lock_usage_with_lock_entry_is_not_unsupported(): + response = "Usage:\nsf lock [offset] [len]\nOpenIPC # " + assert uboot_sf_lock_unsupported(response) is False + assert uboot_flash_command_error(response) is not None + + def test_uboot_flash_error_accepts_successful_progress(): response = "Erasing at 0x240000 -- 100% complete.\nOpenIPC # " assert uboot_flash_command_error(response) is None diff --git a/tests/test_install_nor_unlock.py b/tests/test_install_nor_unlock.py index f8ea177..daf1fa1 100644 --- a/tests/test_install_nor_unlock.py +++ b/tests/test_install_nor_unlock.py @@ -48,6 +48,9 @@ async def _run_env_only_install( probe_response: str = 'Spi(cs1): Block:64KB Chip:8MB Name:"XT25F64B"\nOpenIPC # ', download_mode: bool = False, download_unlock_ok: bool = True, + download_ethaddr_ok: bool = True, + download_ethaddr_response: str = "ethaddr=00:12:41:8e:c6:0e\n", + shell_unlock_timeout: bool = False, ): import defib.flashdump import defib.recovery.session @@ -95,6 +98,8 @@ async def fake_send_command( if command == "sf probe 0": return probe_response if command == "sf lock 0": + if shell_unlock_timeout: + raise TransportTimeout("synthetic sf lock prompt timeout") return unlock_response if command == "printenv ethaddr": return "ethaddr=00:12:41:8e:c6:0e\nOpenIPC # " @@ -119,7 +124,7 @@ async def send_command( if command == "sf lock 0": return download_unlock_ok, unlock_response if command == "printenv ethaddr": - return True, "ethaddr=00:12:41:8e:c6:0e\n" + return download_ethaddr_ok, download_ethaddr_response if command == "saveenv": return True, "Saving Environment to SPI Flash... done\n" return True, "" @@ -182,7 +187,7 @@ async def test_nor_unlock_unsupported_is_compatible(monkeypatch, tmp_path): await run_install(request) - assert "sf lock 0" in commands + assert commands.count("sf lock 0") == 1 assert "saveenv" in commands assert transport.closed is True @@ -285,3 +290,97 @@ async def prompt_sensitive_send_command( assert commands == ["sf probe 0", "reset"] assert transport.closed is True + + +@pytest.mark.asyncio +async def test_nor_probe_failed_to_initialize_stops_before_unlock(monkeypatch, tmp_path): + run_install, request, commands, transport = await _run_env_only_install( + monkeypatch, + tmp_path, + "OpenIPC # ", + probe_response="Failed to initialize SPI flash at 0:0 (error -2)\nOpenIPC # ", + ) + + with pytest.raises(typer.Exit) as exc_info: + await run_install(request) + + assert exc_info.value.exit_code == 1 + assert commands == ["sf probe 0"] + assert transport.closed is True + + +@pytest.mark.asyncio +async def test_shell_unlock_prompt_timeout_stops_before_persistent_write( + monkeypatch, tmp_path +): + run_install, request, commands, transport = await _run_env_only_install( + monkeypatch, + tmp_path, + "", + shell_unlock_timeout=True, + ) + + with pytest.raises(typer.Exit) as exc_info: + await run_install(request) + + assert exc_info.value.exit_code == 1 + assert "sf lock 0" in commands + assert "saveenv" not in commands + assert transport.closed is True + + +@pytest.mark.asyncio +async def test_sf_lock_usage_for_supported_subcommand_is_not_treated_as_unsupported( + monkeypatch, tmp_path +): + run_install, request, commands, transport = await _run_env_only_install( + monkeypatch, + tmp_path, + "Usage:\nsf lock [offset] [len]\nOpenIPC # ", + ) + + with pytest.raises(typer.Exit) as exc_info: + await run_install(request) + + assert exc_info.value.exit_code == 1 + assert "sf lock 0" in commands + assert "saveenv" not in commands + assert transport.closed is True + + +@pytest.mark.asyncio +async def test_download_missing_optional_ethaddr_still_generates_rescue_mac( + monkeypatch, tmp_path +): + run_install, request, commands, transport = await _run_env_only_install( + monkeypatch, + tmp_path, + "OpenIPC # ", + download_mode=True, + download_ethaddr_ok=False, + download_ethaddr_response='## Error: "ethaddr" not defined\n', + ) + + await run_install(request) + + assert any(command.startswith("setenv ethaddr ") for command in commands) + assert "saveenv" in commands + assert transport.closed is True + + + +@pytest.mark.asyncio +async def test_download_unsupported_sf_lock_remains_compatible(monkeypatch, tmp_path): + run_install, request, commands, transport = await _run_env_only_install( + monkeypatch, + tmp_path, + "Usage:\nsf probe [[bus:]cs] [hz] [mode]\nsf read addr offset len\n", + download_mode=True, + download_unlock_ok=False, + ) + + await run_install(request) + + assert "sf lock 0" in commands + assert "saveenv" in commands + assert transport.closed is True diff --git a/tests/test_install_rootfs_data.py b/tests/test_install_rootfs_data.py index 854ab42..866e21b 100644 --- a/tests/test_install_rootfs_data.py +++ b/tests/test_install_rootfs_data.py @@ -143,7 +143,7 @@ async def test_generic_rootfs_data_stage_keeps_existing_noop_behavior( @pytest.mark.asyncio -async def test_wipe_rootfs_data_flag_erases_and_verifies_without_stage_dependency( +async def test_wipe_rootfs_data_flag_erases_and_verifies_in_exact_stage_plan( monkeypatch, tmp_path ): run_install, request, commands, transport = await _prepare_generic_install( @@ -152,7 +152,7 @@ async def test_wipe_rootfs_data_flag_erases_and_verifies_without_stage_dependenc await run_install( request( - stages=("reset",), + stages=("rootfs-data", "reset"), wipe_rootfs_data=True, ) ) @@ -177,7 +177,7 @@ async def test_wipe_rootfs_data_rejects_nand_before_transport(monkeypatch, tmp_p with pytest.raises(typer.Exit) as exc_info: await run_install( request( - stages=("reset",), + stages=("rootfs-data", "reset"), wipe_rootfs_data=True, nand=True, ) @@ -203,3 +203,76 @@ async def test_wipe_rootfs_data_conflicts_with_skip_stage(monkeypatch, tmp_path) assert exc_info.value.exit_code == 2 assert commands == [] + + +@pytest.mark.asyncio +async def test_wipe_rootfs_data_conflicts_with_exact_stage_omitting_rootfs_data( + monkeypatch, tmp_path +): + run_install, request, commands, _transport = await _prepare_generic_install( + monkeypatch, tmp_path + ) + + with pytest.raises(typer.Exit) as exc_info: + await run_install( + request( + stages=("uboot",), + wipe_rootfs_data=True, + ) + ) + + assert exc_info.value.exit_code == 2 + assert commands == [] + + +@pytest.mark.asyncio +async def test_chainloaded_stock_uboot_wipe_requires_uboot_stage(monkeypatch, tmp_path): + import defib.vendors.registry + from defib.vendors.base import UBootBootstrapResult + from defib.vendors.registry import StockUBootTarget + + run_install, request, commands, transport = await _prepare_generic_install( + monkeypatch, tmp_path + ) + + target = StockUBootTarget( + selector="hi3516ev200:test-stock", + handler="synthetic", + load_address=0x82000000, + display_name="Synthetic stock board", + vendor="Test", + stock_uboot_name="Synthetic U-Boot", + ) + + class ChainloadedBootstrap: + requires_echo_verification = False + + async def bootstrap(self, transport_obj, firmware, *, filename): + assert transport_obj is transport + return UBootBootstrapResult( + recovery=RecoveryResult(success=True), + chainloaded=True, + ) + + monkeypatch.setattr( + defib.vendors.registry, + "get_stock_uboot_target", + lambda selector: target, + ) + monkeypatch.setattr( + defib.vendors.registry, + "create_uboot_bootstrap", + lambda *args, **kwargs: ChainloadedBootstrap(), + ) + + with pytest.raises(typer.Exit) as exc_info: + await run_install( + request( + stages=("rootfs-data",), + wipe_rootfs_data=True, + ) + ) + + assert exc_info.value.exit_code == 2 + assert not any(command.startswith("sf erase ") for command in commands) + assert transport.closed is True diff --git a/tests/test_nand_install.py b/tests/test_nand_install.py index 2063d2f..8d97a2c 100644 --- a/tests/test_nand_install.py +++ b/tests/test_nand_install.py @@ -1,5 +1,13 @@ """Tests for NAND flash install support and protocol robustness fixes.""" +from __future__ import annotations + +import io +import tarfile +from contextlib import asynccontextmanager + +import pytest + from defib.cli.app import ( _NAND_LAYOUT, _NOR8M_LAYOUT, @@ -110,3 +118,163 @@ def test_bootargs_is_single_line(self): assert "\n" not in args assert "\x00" not in args assert "\r" not in args + + +@pytest.mark.asyncio +async def test_nand_install_without_crc32_keeps_legacy_compatibility( + monkeypatch, tmp_path, capsys +): + import defib.flashdump + import defib.network.ip_manager + import defib.network.tftp_server + import defib.recovery.session + import defib.transport.serial_platform + from defib.install import InstallRequest + from defib.install.orchestrator import run_install + from defib.recovery.events import RecoveryResult + from defib.transport.base import Transport, TransportTimeout + + firmware_tar = tmp_path / "firmware.tgz" + kernel = b"K" * 1024 + rootfs = b"R" * 2048 + with tarfile.open(firmware_tar, "w:gz") as archive: + for name, payload in ( + ("uImage.hi3516cv300", kernel), + ("rootfs.squashfs.hi3516cv300", rootfs), + ): + info = tarfile.TarInfo(name) + info.size = len(payload) + archive.addfile(info, io.BytesIO(payload)) + + uboot = tmp_path / "u-boot.bin" + uboot.write_bytes(b"U" * 1024) + + class ShellTransport(Transport): + def __init__(self) -> None: + self.rx = bytearray() + self.closed = False + + async def read(self, size: int, timeout: float | None = None) -> bytes: + if not self.rx: + raise TransportTimeout("no data") + data = bytes(self.rx[:size]) + del self.rx[:size] + return data + + async def write(self, data: bytes) -> None: + if b"\x03" in data: + self.rx.extend(b"OpenIPC # ") + + async def flush_input(self) -> None: + self.rx.clear() + + async def flush_output(self) -> None: + return None + + async def bytes_waiting(self) -> int: + return len(self.rx) + + async def close(self) -> None: + self.closed = True + + transport = ShellTransport() + commands: list[str] = [] + tftp_files: dict[str, bytes] = {} + + class FakeRecoverySession: + def __init__(self, *args, **kwargs) -> None: + pass + + async def run(self, transport_obj, **kwargs): + assert transport_obj is transport + return RecoveryResult(success=True) + + async def fake_create_transport(port: str): + assert port == "COM15" + return transport + + @asynccontextmanager + async def fake_temporary_ip(interface: str, ip: str, netmask: str): + assert interface == "Ethernet" + yield + + class FakeTFTPTransport: + def close(self) -> None: + pass + + class FakeTFTPProtocol: + def __init__(self, files): + self._files = dict(files) + + def set_max_blocksize(self, blocksize: int) -> None: + raise AssertionError("no retry block-size change expected") + + async def fake_start_tftp_server(*, files, bind_addr, port, done_count): + assert done_count == 2 + tftp_files.clear() + tftp_files.update(files) + return FakeTFTPTransport(), FakeTFTPProtocol(files) + + async def fake_send_command( + transport_obj, + command: str, + timeout: float = 0.0, + **kwargs, + ) -> str: + assert transport_obj is transport + commands.append(command) + + if command == "nand info": + return "Device 0: nand0, sector size 128 KiB\nOpenIPC # " + if command.startswith(("tftpboot ", "tftp ")): + name = command.split()[-1] + return f"Bytes transferred = {len(tftp_files[name])}\nOpenIPC # " + if command.startswith("crc32 "): + return "Unknown command 'crc32'\nOpenIPC # " + if command.startswith(("nand erase ", "nand write ")): + return "OK\nOpenIPC # " + return "OpenIPC # " + + monkeypatch.setattr(defib.flashdump, "send_command", fake_send_command) + monkeypatch.setattr(defib.recovery.session, "RecoverySession", FakeRecoverySession) + monkeypatch.setattr( + defib.transport.serial_platform, + "create_transport", + fake_create_transport, + ) + monkeypatch.setattr( + defib.transport.serial_platform, + "normalize_port_name", + lambda port: port, + ) + monkeypatch.setattr( + defib.network.ip_manager, + "temporary_ip", + fake_temporary_ip, + ) + monkeypatch.setattr( + defib.network.tftp_server, + "start_tftp_server", + fake_start_tftp_server, + ) + + await run_install( + InstallRequest( + chip="hi3516cv300", + firmware_path=str(firmware_tar), + uboot_path=str(uboot), + port="COM15", + nic="Ethernet", + nand=True, + stages=("kernel", "rootfs"), + output="json", + ) + ) + + assert commands.count("crc32 0x82000000 0x400") == 1 + assert "tftpboot 0x82000000 k" in commands + assert "tftpboot 0x82000000 r" in commands + assert any(command.startswith("nand erase ") for command in commands) + assert any(command.startswith("nand write ") for command in commands) + assert capsys.readouterr().out.count("crc32 is unavailable") == 1 + assert transport.closed is True diff --git a/tests/test_uart_command_integrity.py b/tests/test_uart_command_integrity.py index d62f54d..c28df6f 100644 --- a/tests/test_uart_command_integrity.py +++ b/tests/test_uart_command_integrity.py @@ -107,4 +107,39 @@ async def test_wait_for_prompt_rejects_partial_command_response() -> None: "tftpboot 0x82000000 k", timeout=0.05, wait_for="# ", + require_prompt=True, ) + + +@pytest.mark.asyncio +async def test_wait_for_prompt_keeps_legacy_partial_response_when_not_required() -> None: + transport = PartialResponseNoPromptTransport(flush_clears_buffer=False) + + response = await send_command( + transport, + "tftpboot 0x82000000 k", + timeout=0.05, + wait_for="# ", + ) + + assert "Bytes transferred = 1558411" in response + + + +class SilentAfterCommandTransport(MockTransport): + async def write(self, data: bytes) -> None: + await super().write(data) + + +@pytest.mark.asyncio +async def test_legacy_wait_for_allows_reset_without_returned_prompt() -> None: + transport = SilentAfterCommandTransport(flush_clears_buffer=False) + + response = await send_command( + transport, + "reset", + timeout=0.05, + wait_for="# ", + ) + + assert response == ""