diff --git a/CLAUDE.md b/CLAUDE.md index 0bbb57a..3c6ebf9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -221,6 +221,12 @@ 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. 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 subtract stages. The stage names are `uboot`, `kernel`, `rootfs`, diff --git a/README.md b/README.md index 6a05e10..82f4ad2 100644 --- a/README.md +++ b/README.md @@ -96,6 +96,14 @@ 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. +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 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 +134,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..5fb6300 100644 --- a/src/defib/cli/app.py +++ b/src/defib/cli/app.py @@ -2179,6 +2179,16 @@ 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; with exact " + "--stage selection include --stage rootfs-data. Registered stock-U-Boot " + "migrations keep their existing migration behavior." + ), + ), final_reset: bool = typer.Option( True, "--final-reset/--no-final-reset", @@ -2234,6 +2244,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/flashdump.py b/src/defib/flashdump.py index 6908c60..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: @@ -316,7 +323,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 and require_prompt: + 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( @@ -325,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: @@ -421,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/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 a2439c9..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,14 @@ 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 async def run_install(request: InstallRequest) -> None: @@ -50,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 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 @@ -74,6 +83,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 @@ -109,6 +119,29 @@ 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", + 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"}) @@ -451,6 +484,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: @@ -555,22 +590,48 @@ 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]") + if allow_failure: + return False, str(exc) + 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( @@ -579,17 +640,104 @@ async def _cmd(cmd: str, timeout: float = 60.0, **kw: object) -> str: 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: - return out + 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) + 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 _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: + # 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 +777,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 @@ -658,6 +809,10 @@ 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 or wipe_rootfs_data: + await _unlock_nor_or_fail() + if output == "human": console.print( f" [green]SPI flash detected[/green]: {nor_size} MiB ({nor_source}), " @@ -825,57 +980,177 @@ 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, + timeout=timeout, ) - resp = await _cmd(tftpboot_cmd, timeout=timeout) - if "unknown command" in resp.lower(): - resp = await _cmd(tftp_cmd, timeout=timeout) - 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, + ) -> 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) + 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)}" + ) + 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_ok, crc_resp = await _cmd_result( + f"crc32 0x{ram_addr:x} 0x{len(orig_data):x}", + 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 + ) + 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: + 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: + 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 " + 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"{attempt + 1} attempt(s):[/red] {failure}" + ) + raise typer.Exit(1) 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} " 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, ) - 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}") + 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 @@ -921,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}", @@ -938,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: @@ -987,13 +1263,20 @@ 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) + verified_ubifs_crc = await _verify_tftp_ram( + "rootfs (UBI)", + tftp_alias["rootfs"], + ubifs_data, + ) if output == "human": - console.print(" TFTP OK") + 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" @@ -1020,7 +1303,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: @@ -1076,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( @@ -1132,8 +1422,20 @@ 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() + + # 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]") @@ -1172,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( @@ -1221,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( @@ -1229,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: @@ -1248,7 +1550,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/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_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_ds_i203_final_contract.py b/tests/test_ds_i203_final_contract.py index 7f09991..ae51727 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,20 @@ 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", + "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 + monkeypatch, tmp_path, crc_failure, capsys ): """Exercise the complete stock->OpenIPC NOR install contract. @@ -666,6 +678,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 +766,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 +863,14 @@ 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 == "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() @@ -922,7 +949,13 @@ async def fake_send_command( output="json", ) - if crc_failure is None: + 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: @@ -940,6 +973,68 @@ async def fake_send_command( assert any(cmd.startswith("cmp.l 0x82000000 ") for cmd in commands) return + 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, MAX_BLOCKSIZE] + + 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 " + ) + 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 = [ + 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 + ) + + 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 new file mode 100644 index 0000000..daf1fa1 --- /dev/null +++ b/tests/test_install_nor_unlock.py @@ -0,0 +1,386 @@ +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, + *, + 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 + 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) + + 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: + 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 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 # " + if command == "saveenv": + return "Saving Environment to SPI Flash... done\nOpenIPC # " + 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 download_ethaddr_ok, download_ethaddr_response + 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", + 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 commands.count("sf lock 0") == 1 + 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 + + +@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 + + +@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 new file mode 100644 index 0000000..866e21b --- /dev/null +++ b/tests/test_install_rootfs_data.py @@ -0,0 +1,278 @@ +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_in_exact_stage_plan( + monkeypatch, tmp_path +): + run_install, request, commands, transport = await _prepare_generic_install( + monkeypatch, tmp_path + ) + + await run_install( + request( + stages=("rootfs-data", "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=("rootfs-data", "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 == [] + + +@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_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..c28df6f 100644 --- a/tests/test_uart_command_integrity.py +++ b/tests/test_uart_command_integrity.py @@ -87,3 +87,59 @@ 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="# ", + 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 == ""