diff --git a/CLAUDE.md b/CLAUDE.md index 7e08c90..a45ec70 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -126,7 +126,7 @@ the handshake, reverting to 115200 after ~30 s idle. 13 commands: `INFO 0x01`, optional features through a capability bitmask (`client.py`). Backends: `spi_flash.c` (fmc100), `spi_flash_hisfc350.c` (V1-era parts), -`emmc_himci.c`. Eleven SoCs are supported; each has its own `ifeq` stanza in +`emmc_himci.c`. Twelve SoCs are supported; each has its own `ifeq` stanza in `agent/Makefile` carrying `LOAD_ADDR` (and `SPI_DRIVER` where it differs). `link.ld` itself is generic — it just places `. = LOAD_ADDR`. diff --git a/agent/Makefile b/agent/Makefile index 0575507..b6f9132 100644 --- a/agent/Makefile +++ b/agent/Makefile @@ -69,6 +69,22 @@ else ifeq ($(SOC),gk7605v100) WDT_BASE = 0x12030000 CRG_BASE = 0x12010000 SYSCTRL_REBOOT = 0x12020004 +else ifeq ($(SOC),gk7205v500) + # V500 bootrom (gk7205v500/v510/v530): the agent rides in the boot-code + # area of a V500 boot image (see wrap_v500_payload). After a UART + # download the bootrom runs that code in place, at the 0x41000000 load + # address + 8 KiB key area + 20 KiB aux area; the image's own entry + # field (0x40707000) is not used on this path. Peripherals match + # gk7205v300. + UART_BASE = 0x12040000 + UART_CLOCK = 24000000 + LOAD_ADDR = 0x41007000 + FLASH_MEM = 0x14000000 + FMC_BASE = 0x10000000 + RAM_BASE = 0x40000000 + WDT_BASE = 0x12030000 + CRG_BASE = 0x12010000 + SYSCTRL_REBOOT = 0x12020004 else ifeq ($(SOC),hi3516cv300) # V3 generation: ARM926EJ-S (ARMv5TEJ), not Cortex-A7. # UART is PL011 (same as ev300 family) at a different base. @@ -188,7 +204,7 @@ else ifeq ($(SOC),hi3520dv200) SYSCTRL_REBOOT = 0x20050004 SPI_DRIVER = hisfc350 else - $(error Unknown SOC: $(SOC). Supported: hi3516ev300 hi3516ev200 gk7205v200 gk7205v300 gk7605v100 hi3516cv300 hi3516cv500 hi3518ev200 hi3516cv610 hi3519v101 hi3520dv200) + $(error Unknown SOC: $(SOC). Supported: hi3516ev300 hi3516ev200 gk7205v200 gk7205v300 gk7605v100 gk7205v500 hi3516cv300 hi3516cv500 hi3518ev200 hi3516cv610 hi3519v101 hi3520dv200) endif # Per-SoC CPU. V3 chips (cv300) are ARM926EJ-S (ARMv5TEJ); diff --git a/agent/README.md b/agent/README.md index bd4ad70..ab2453c 100644 --- a/agent/README.md +++ b/agent/README.md @@ -75,6 +75,7 @@ Requires `arm-none-eabi-gcc` (Arch: `pacman -S arm-none-eabi-gcc arm-none-eabi-n | gk7205v200 | 0x12040000 | 0x14000000 | 0x40000000 | 0x41000000 | | gk7205v300 | 0x12040000 | 0x14000000 | 0x40000000 | 0x41000000 | | gk7605v100 | 0x12040000 | 0x14000000 | 0x40000000 | 0x41000000 | +| gk7205v500 | 0x12040000 | 0x14000000 | 0x40000000 | 0x41007000 | | hi3516cv300 | 0x12100000 | 0x14000000 | 0x80000000 | 0x81000000 | | hi3516cv500 | 0x12100000 | 0x14000000 | 0x80000000 | 0x81000000 | | hi3518ev200 | 0x12100000 | 0x14000000 | 0x80000000 | 0x81000000 | @@ -82,6 +83,11 @@ Requires `arm-none-eabi-gcc` (Arch: `pacman -S arm-none-eabi-gcc arm-none-eabi-n | hi3519v101 | 0x12100000 | 0x14000000 | 0x80000000 | 0x81000000 | | hi3520dv200 | 0x20080000 | 0x58000000 | 0x80000000 | 0x81000000 | +gk7205v500 also serves gk7205v510/v530. The V500 bootrom has no SPL stage: +`defib agent upload` puts the agent in the boot-code slot of an OpenIPC +u-boot-xmedia image (downloaded, or `-f` for your own), whose DDR-init code +runs first; the bootrom then executes the slot in place at `0x41007000`. + Addresses from [qemu-hisilicon](https://github.com/OpenIPC/LoTool) hardware definitions. ### Where defib looks for a built binary diff --git a/agent/spi_flash.c b/agent/spi_flash.c index 29903b3..16d8b23 100644 --- a/agent/spi_flash.c +++ b/agent/spi_flash.c @@ -117,7 +117,7 @@ #define NAND_STATUS_E_FAIL (1 << 2) /* Erase Fail */ #define NAND_STATUS_P_FAIL (1 << 3) /* Program Fail */ -/* NAND geometry — currently only MX35LF1GE4AB (1Gbit) is recognized. +/* NAND geometry shared by every nand_ids[] part (1 Gbit). * On-chip ECC is enabled by default; reads return ECC-corrected data. */ #define NAND_PAGE_SIZE 2048 #define NAND_BLOCK_SIZE (64 * NAND_PAGE_SIZE) /* 128 KiB */ @@ -421,16 +421,30 @@ static void nand_write_enable(void) { fmc_wait_ready(); } -/* Identify SPI NAND chip from JEDEC ID. Returns 1 if recognized, 0 otherwise. - * Currently only MX35LF1GE4AB (Macronix, c2 12, 1Gbit / 128MB). The agent's - * flash_read_id reads bytes [0..2] of an 8-byte fetch; some SPI NAND chips - * return the manufacturer ID with a leading dummy byte, so we accept the ID - * shifted by one position too. */ +/* SPI NAND chips the agent drives, by manufacturer + first device ID byte. + * All are 1 Gbit, 2 KiB pages, 64 pages per 128 KiB block — the geometry + * flash_init hardcodes. Only two ID bytes are matched because a chip that + * answers 0x9F with a leading dummy byte pushes its third byte out of the + * three-byte window flash_read_id captures (W25N01GV reads back 00 EF AA). + * None of these pairs collides with a SPI NOR ID. + * + * An unrecognised NAND falls through to the NOR path, which is not just + * wrong but can hang: flash_global_unlock() polls a NOR status register a + * NAND does not implement (GD5F1GM7 never clears "WIP"). */ +static const uint8_t nand_ids[][2] = { + { 0xC2, 0x12 }, /* Macronix MX35LF1GE4AB */ + { 0xEF, 0xAA }, /* Winbond W25N01GV (EF AA 21) */ + { 0xC8, 0x91 }, /* GigaDevice GD5F1GM7UE, 3.3 V */ + { 0xC8, 0x81 }, /* GigaDevice GD5F1GM7RE, 1.8 V */ +}; + +/* Returns 1 if id[] is a known SPI NAND, read either directly or shifted + * by one dummy byte (id[0] = dummy). */ static int nand_identify(const uint8_t id[3]) { - /* Direct: id[0]=0xC2 id[1]=0x12 */ - if (id[0] == 0xC2 && id[1] == 0x12) return 1; - /* Shifted by 1 (dummy byte at id[0]): id[1]=0xC2 id[2]=0x12 */ - if (id[1] == 0xC2 && id[2] == 0x12) return 1; + for (unsigned i = 0; i < sizeof(nand_ids) / sizeof(nand_ids[0]); i++) { + if (id[0] == nand_ids[i][0] && id[1] == nand_ids[i][1]) return 1; + if (id[1] == nand_ids[i][0] && id[2] == nand_ids[i][1]) return 1; + } return 0; } @@ -459,7 +473,7 @@ int flash_init(flash_info_t *info) { * memory-mapped boot mode and uses different protection (BP bits * via SET_FEATURE 0xA0 instead of write-status-register). */ info->flash_type = FLASH_TYPE_NAND; - info->size = 128u * 1024u * 1024u; /* MX35LF1GE4AB = 128 MiB */ + info->size = 128u * 1024u * 1024u; /* every nand_ids[] part is 1 Gbit */ info->sector_size = NAND_BLOCK_SIZE; /* 128 KiB erase block */ info->page_size = NAND_PAGE_SIZE; /* 2 KiB read/program page */ current_flash_type = FLASH_TYPE_NAND; diff --git a/src/defib/agent/client.py b/src/defib/agent/client.py index 4c25242..6f9ba7f 100644 --- a/src/defib/agent/client.py +++ b/src/defib/agent/client.py @@ -163,6 +163,9 @@ def bad_block(self) -> list[SectorResult]: "hi3516cv608": "hi3516cv610", # cv6xx-family, same memory map "hi3518ev200": "hi3518ev200", "hi3520dv200": "hi3520dv200", # V1-era, HISFC350 SPI controller + "gk7205v500": "gk7205v500", + "gk7205v510": "gk7205v500", # V500 family, same memory map + "gk7205v530": "gk7205v500", } diff --git a/src/defib/cli/app.py b/src/defib/cli/app.py index 837779d..2d3ca2d 100644 --- a/src/defib/cli/app.py +++ b/src/defib/cli/app.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections.abc import Callable, Coroutine -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, NoReturn import typer @@ -1080,7 +1080,7 @@ def agent_upload( chip: str = typer.Option(..., "-c", "--chip", help="Chip model name"), port: str = typer.Option("/dev/ttyUSB0", "-p", "--port", help="Serial device (/dev/ttyUSB0), tcp://host:port, rfc2217://host:port, or socket:///path"), output: str = typer.Option("human", "--output", help="Output mode: human, json"), - file: str | None = typer.Option(None, "-f", "--file", help="CV6xx composite boot file (GSL+DDR+U-Boot); required for CV6xx, ignored for other protocols"), + file: str | None = typer.Option(None, "-f", "--file", help="CV6xx: composite boot file (GSL+DDR+U-Boot), required. V500: U-Boot image whose header and DDR-init code carry the agent (default: OpenIPC u-boot-xmedia download). Ignored for other protocols"), power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via configured controller (DEFIB_POWER_TYPE)"), poe_port_override: str = typer.Option("", "--poe-port", help="Explicit MikroTik ether port (e.g. ether3) — overrides comment-based auto-discovery. Requires --power-cycle."), ) -> None: @@ -1107,6 +1107,7 @@ async def _agent_upload_async( from defib.profiles.loader import load_profile from defib.protocol.hisilicon_cv6xx import HiSiliconCV6xx from defib.protocol.hisilicon_standard import HiSiliconStandard + from defib.protocol.hisilicon_v500 import HiSiliconV500 from defib.protocol.registry import find_protocol from defib.recovery.events import ProgressEvent, Stage from defib.transport.serial_platform import ( @@ -1149,7 +1150,24 @@ async def _agent_upload_async( ) return - # HiSiliconStandard / V500 path — needs SoC profile for the SPL+agent + # V500 has no SPL stage and no profile: the agent replaces the boot code + # of a V500 boot image, and the bootrom runs it after the image's own + # DDR-init (aux) code. + if protocol_cls is HiSiliconV500: + await _agent_upload_v500( + chip=chip, + port=port, + output=output, + console=console, + agent_path=agent_path, + agent_data=agent_data, + donor_path=composite_path, + power_cycle=power_cycle, + poe_port_override=poe_port_override, + ) + return + + # HiSiliconStandard path — needs SoC profile for the SPL+agent # two-stage upload. profile = load_profile(chip) cached_fw = get_cached_path(chip) @@ -1394,6 +1412,154 @@ async def _power_cycle_into_handshake( return hs +# How long agent upload waits for a human to power-cycle a V500 board. +MANUAL_HANDSHAKE_TIMEOUT = 60.0 + + +async def _agent_upload_v500( + *, + chip: str, + port: str, + output: str, + console: Any, + agent_path: Any, + agent_data: bytes, + donor_path: str | None, + power_cycle: bool, + poe_port_override: str = "", +) -> None: + """Upload the agent to a V500-family SoC inside a donor boot image.""" + import asyncio + import json as json_mod + from pathlib import Path + + from defib.agent.client import FlashAgentClient + from defib.firmware import download_v500_donor + from defib.power.base import PowerControllerError + from defib.protocol.hisilicon_v500 import ( + V500_AGENT_LOAD_ADDR, V500_CHIP_IDS, HiSiliconV500, v500_member, + wrap_v500_payload, + ) + from defib.recovery.events import ProgressEvent, Stage + from defib.transport.serial_platform import ( + create_transport, normalize_port_name, + ) + + def fail(message: str) -> NoReturn: + if output == "json": + print(json_mod.dumps({"event": "error", "message": message})) + else: + console.print(f"[red]{message}[/red]") + raise typer.Exit(1) + + try: + donor_file = Path(donor_path) if donor_path else download_v500_donor(chip) + wrapped = wrap_v500_payload( + donor_file.read_bytes(), agent_data, V500_AGENT_LOAD_ADDR, + ) + except (OSError, ValueError, ConnectionError) as e: + fail(f"Cannot build the V500 boot image: {e}") + + power = None + power_port = "" + if power_cycle: + from defib.power.factory import power_controller_from_env + try: + power = power_controller_from_env() + power_port = await _resolve_power_port(power, port, poe_port_override) + except Exception as e: + if power is not None: + await power.close() + fail(f"Power controller error: {e}") + + if output == "human": + console.print(f"Agent: [cyan]{agent_path.name}[/cyan] ({len(agent_data)} bytes)") + console.print(f"Boot image: [cyan]{donor_file.name}[/cyan] header + DDR init, " + f"{len(wrapped)} bytes with the agent") + if power is None: + console.print("\n[yellow]Power-cycle the camera now![/yellow]\n") + + def on_progress(e: ProgressEvent) -> None: + if e.message: + if output == "human": + console.print(f" {e.message}") + elif output == "json": + print(json_mod.dumps({"event": "progress", "message": e.message}), flush=True) + + try: + transport = await create_transport(normalize_port_name(port)) + except Exception as e: + if power is not None: + await power.close() + fail(f"Cannot open {port}: {e}") + if power is not None: + _attach_power_transport(power, transport) + try: + protocol = HiSiliconV500() + if power is not None: + def on_power_log(message: str) -> None: + on_progress(ProgressEvent( + stage=Stage.POWER_CYCLE, bytes_sent=0, bytes_total=1, message=message, + )) + + try: + hs = await _power_cycle_into_handshake( + power, power_port, transport, + lambda: protocol.handshake(transport, on_progress), + on_power_log, proactive=True, + ) + except PowerControllerError as e: + fail(f"Power cycle failed: {e}") + finally: + await power.close() + else: + # Without a power controller nobody retries for us; give the + # human a generous window, then report instead of hanging. + try: + hs = await asyncio.wait_for( + protocol.handshake(transport, on_progress), + timeout=MANUAL_HANDSHAKE_TIMEOUT, + ) + except asyncio.TimeoutError: + fail(f"No bootrom response within {MANUAL_HANDSHAKE_TIMEOUT:.0f}s") + if not hs.success: + fail("Handshake failed") + + # The donor's DDR init is per family member: a gk7205v500 image does + # not bring up a V510's DDR, and then the agent just never answers. + detected = V500_CHIP_IDS.get(hs.chip_id or 0) + if detected and not donor_path and detected != v500_member(chip): + fail( + f"The board answered as a {detected.upper()} (chip ID " + f"0x{hs.chip_id:08x}), not {chip}: its boot image carries the " + f"wrong DDR init. Re-run with -c gk7205{detected}." + ) + + result = await protocol.send_firmware(transport, wrapped, on_progress) + if not result.success: + fail(f"Upload failed: {result.error}") + + if output == "human": + console.print("[green]Agent uploaded![/green] Waiting for READY...") + + client = FlashAgentClient(transport, chip) + if not await client.connect(timeout=10.0): + fail( + "Agent not responding. The bootrom accepted the image, so the " + "donor's DDR init or the boot entry is the suspect: try " + "-f with the board's own U-Boot as the donor." + ) + info = await client.get_info() + if output == "human": + console.print("[green bold]Agent ready![/green bold]") + console.print(f" RAM: 0x{info.get('ram_base', 0):08x}") + console.print(f" Flash: {int(info.get('flash_size', 0)) // 1024}KB") + elif output == "json": + print(json_mod.dumps({"event": "ready", **info})) + finally: + await transport.close() + + async def _agent_upload_cv6xx( *, chip: str, diff --git a/src/defib/firmware.py b/src/defib/firmware.py index aaa5347..6e06bab 100644 --- a/src/defib/firmware.py +++ b/src/defib/firmware.py @@ -257,7 +257,15 @@ def download_firmware( # Download name = asset_name(chip) assert name is not None # firmware_url() is None otherwise - dest = get_cache_dir() / name + return _download(url, get_cache_dir() / name, on_progress) + + +def _download( + url: str, + dest: Path, + on_progress: Callable[[int, int], None] | None = None, +) -> Path: + """Fetch ``url`` into ``dest``, removing any partial file on failure.""" logger.info("Downloading firmware from %s", url) try: @@ -287,3 +295,26 @@ def download_firmware( if isinstance(e, (ValueError, ConnectionError)): raise raise ConnectionError(f"Failed to download firmware: {e}") from e + + +# V500-family (gk7205v500/v510/v530) U-Boot is built in OpenIPC/u-boot-xmedia, +# not OpenIPC/firmware. defib only needs its boot-image header and DDR-init +# (aux) code to carry the flash agent, and those do not depend on the flash +# type, so the smaller NOR build is the donor. +XMEDIA_UBOOT_BASE_URL = ( + "https://github.com/OpenIPC/u-boot-xmedia/releases/download/latest" +) + + +def download_v500_donor(chip: str) -> Path: + """Fetch (or reuse the cached) V500 U-Boot used as the agent's boot image. + + Raises: + ConnectionError: If the download fails. + """ + name = f"u-boot-{_strip_variant(chip).lower()}-nor.bin" + dest = get_cache_dir() / name + if dest.exists(): + logger.info("Using cached V500 donor: %s", dest) + return dest + return _download(f"{XMEDIA_UBOOT_BASE_URL}/{name}", dest) diff --git a/src/defib/protocol/hisilicon_v500.py b/src/defib/protocol/hisilicon_v500.py index 7e6c1b4..936cdab 100644 --- a/src/defib/protocol/hisilicon_v500.py +++ b/src/defib/protocol/hisilicon_v500.py @@ -35,6 +35,21 @@ ]) HANDSHAKE_TIMEOUT = 20.0 # seconds + +# Chip ID in the handshake reply -> V500 family member. 0x72050510 was read +# from a GK7205V510; the other two follow the same pattern and are unverified. +# The xm7205v5x0 parts are the same silicon under another name. +V500_CHIP_IDS = { + 0x72050500: "v500", + 0x72050510: "v510", + 0x72050530: "v530", +} + + +def v500_member(chip: str) -> str | None: + """``"v510"`` for ``gk7205v510`` / ``xm7205v510:anything``, else None.""" + base = chip.lower().split(":", 1)[0] + return base[-4:] if base in V500_SOCS else None HANDSHAKE_BURST_FRAMES = 8 # 112 B, ~10 ms at 115200 baud HANDSHAKE_REPLY_LEN = 14 CHUNK_ACK_TIMEOUT = 4.0 # seconds @@ -42,6 +57,57 @@ BOOT_LOAD_ADDR = 0x41000000 +# V500 boot image layout (u-boot-xmedia include/configs/xm72050500.h): +# 0x0000 key area (8 KiB; RSA key fields, zero when unsigned) +# 0x0400 params: aux-area length, boot-code length, total boot length +# (= code + 0x200 tail), aux/boot encryption flags, boot entry +# 0x2000 aux area (DDR init/training code), aux-area-length bytes +# then boot code +# After a UART download the bootrom runs the boot code in place, at +# BOOT_LOAD_ADDR + its file offset; the entry field in the params is not +# used on this path (a probe placed there reported pc=0x41007000, not the +# 0x40707000 the field holds). +V500_KEY_AREA_LEN = 0x2000 +V500_AUX_AREA_LEN_OFF = 0x400 +V500_BOOT_CODE_LEN_OFF = 0x404 +V500_TOTAL_BOOT_LEN_OFF = 0x408 +V500_BOOT_TAIL_LEN = 0x200 +V500_BOOT_CODE_ALIGN = 0x400 +# Where agent/Makefile links the gk7205v500 agent: BOOT_LOAD_ADDR + key area +# + the 20 KiB aux area every published V500 U-Boot carries. +V500_AGENT_LOAD_ADDR = 0x41007000 + + +def wrap_v500_payload(donor: bytes, payload: bytes, load_addr: int) -> bytes: + """Put ``payload`` in the boot-code area of a V500 boot image. + + ``donor`` is a complete V500 U-Boot image (e.g. OpenIPC u-boot-xmedia's + ``u-boot-gk7205v500-nor.bin``). Its key area, params and aux (DDR init) + area are kept verbatim; its boot code is replaced by ``payload`` and the + two length fields are patched to match. The bootrom runs the boot code + in place, so ``payload`` must be linked at BOOT_LOAD_ADDR plus the + donor's boot-code offset; ``load_addr`` is checked against that to catch + a mismatch before it turns into a silent hang on the board. + """ + if len(donor) < V500_TOTAL_BOOT_LEN_OFF + 4: + raise ValueError("donor is too short to be a V500 boot image") + aux_len = struct.unpack_from(" len(donor): + raise ValueError(f"donor aux-area length 0x{aux_len:x} is not plausible") + if BOOT_LOAD_ADDR + code_pos != load_addr: + raise ValueError( + f"donor boot code runs at 0x{BOOT_LOAD_ADDR + code_pos:08x}, but the " + f"payload is linked at 0x{load_addr:08x}" + ) + code = bytearray(payload) + code += b"\x00" * (-len(code) % V500_BOOT_CODE_ALIGN) + image = bytearray(donor[:code_pos]) + code + b"\x00" * V500_BOOT_TAIL_LEN + struct.pack_into(" None: if callback is not None: callback(event) diff --git a/tests/test_agent_upload_v500.py b/tests/test_agent_upload_v500.py new file mode 100644 index 0000000..5a8998c --- /dev/null +++ b/tests/test_agent_upload_v500.py @@ -0,0 +1,151 @@ +"""Tests for the V500 agent-upload path's failure handling.""" + +from __future__ import annotations + +import asyncio +import io +import struct +from pathlib import Path +from typing import Any + +import pytest +import typer +from rich.console import Console + +from defib.cli import app as cli_app +from defib.power.base import PowerController +from defib.protocol.hisilicon_v500 import HiSiliconV500, v500_member +from defib.recovery.events import HandshakeResult +from defib.transport.mock import MockTransport + + +def _write_donor(tmp_path: Path) -> Path: + image = bytearray(b"\xa5" * (0x7000 + 0x800 + 0x200)) + struct.pack_into("<3I", image, 0x400, 0x5000, 0x800, 0xa00) + path = tmp_path / "donor.bin" + path.write_bytes(bytes(image)) + return path + + +class ClosingPower(PowerController): + def __init__(self) -> None: + self.closed = False + + @classmethod + def name(cls) -> str: + return "test" + + async def power_off(self, port: str) -> None: + return None + + async def power_on(self, port: str) -> None: + return None + + async def close(self) -> None: + self.closed = True + + +async def _run( + tmp_path: Path, chip: str, *, donor: bool = False, power_cycle: bool = False, +) -> str: + out = io.StringIO() + with pytest.raises(typer.Exit): + await cli_app._agent_upload_v500( + chip=chip, port="/dev/null", output="human", + console=Console(file=out, width=200), + agent_path=tmp_path / "agent.bin", agent_data=b"\x00" * 64, + donor_path=str(_write_donor(tmp_path)) if donor else None, + power_cycle=power_cycle, + ) + return out.getvalue() + + +@pytest.fixture +def fake_link(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> MockTransport: + transport = MockTransport() + + async def fake_create(_url: str) -> MockTransport: + return transport + + monkeypatch.setattr("defib.transport.serial_platform.create_transport", fake_create) + monkeypatch.setattr("defib.firmware.download_v500_donor", + lambda chip: _write_donor(tmp_path)) + return transport + + +def test_v500_member() -> None: + assert v500_member("gk7205v510") == "v510" + assert v500_member("XM7205V530:board") == "v530" + assert v500_member("hi3516ev300") is None + + +async def test_wrong_chip_is_named_before_upload( + fake_link: MockTransport, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +) -> None: + async def handshake(self: HiSiliconV500, transport: Any, cb: Any = None) -> HandshakeResult: + return HandshakeResult(success=True, chip_id=0x72050510) + + monkeypatch.setattr(HiSiliconV500, "handshake", handshake) + text = await _run(tmp_path, "gk7205v500") + assert "V510" in text and "-c gk7205v510" in text + # Nothing beyond the handshake went out: no HEAD frame. + assert b"\xfe\x00\xff\x01" not in fake_link.all_tx_data + + +async def test_explicit_donor_skips_the_chip_check( + fake_link: MockTransport, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +) -> None: + async def handshake(self: HiSiliconV500, transport: Any, cb: Any = None) -> HandshakeResult: + return HandshakeResult(success=True, chip_id=0x72050510) + + async def send(self: HiSiliconV500, *a: Any, **k: Any) -> Any: + from defib.recovery.events import RecoveryResult + return RecoveryResult(success=False, error="stop here") + + monkeypatch.setattr(HiSiliconV500, "handshake", handshake) + monkeypatch.setattr(HiSiliconV500, "send_firmware", send) + text = await _run(tmp_path, "gk7205v500", donor=True) + assert "stop here" in text + + +async def test_manual_handshake_times_out( + fake_link: MockTransport, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +) -> None: + async def handshake(self: HiSiliconV500, transport: Any, cb: Any = None) -> HandshakeResult: + await asyncio.sleep(10) + raise AssertionError("unreachable") + + monkeypatch.setattr(HiSiliconV500, "handshake", handshake) + monkeypatch.setattr(cli_app, "MANUAL_HANDSHAKE_TIMEOUT", 0.05) + text = await _run(tmp_path, "gk7205v510") + assert "No bootrom response" in text + + +async def test_power_closed_when_port_cannot_open( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, +) -> None: + power = ClosingPower() + + async def broken_create(_url: str) -> MockTransport: + raise OSError("no such device") + + monkeypatch.setattr("defib.transport.serial_platform.create_transport", broken_create) + monkeypatch.setattr("defib.power.factory.power_controller_from_env", lambda: power) + text = await _run(tmp_path, "gk7205v510", donor=True, power_cycle=True) + assert "no such device" in text + assert power.closed + + +def test_donor_name_is_lowercase(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + from defib import firmware + + monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path)) + urls: list[str] = [] + + def fake_download(url: str, dest: Path, on_progress: Any = None) -> Path: + urls.append(url) + return dest + + monkeypatch.setattr(firmware, "_download", fake_download) + firmware.download_v500_donor("GK7205V510") + assert urls[0].endswith("/u-boot-gk7205v510-nor.bin") diff --git a/tests/test_firmware.py b/tests/test_firmware.py index 43bbdd7..a3971c6 100644 --- a/tests/test_firmware.py +++ b/tests/test_firmware.py @@ -223,3 +223,35 @@ def test_board_variant_never_falls_back_to_universal_cache( dedicated = cache / "u-boot-hi3518ev100-ddr3-256m-universal.bin" dedicated.write_bytes(b"D" * 182580) assert get_cached_path("hi3518ev100:hiwatch-ds-i203") == dedicated + + +class TestV500Donor: + def test_downloads_the_nor_build_once(self, monkeypatch, tmp_path): + import io + + from defib import firmware + + monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path)) + urls: list[str] = [] + + class Resp(io.BytesIO): + headers: dict[str, str] = {} + + def __enter__(self): + return self + + def __exit__(self, *a): + self.close() + + def fake_urlopen(req, timeout=None): + urls.append(req.full_url) + return Resp(b"\x00" * 4096) + + monkeypatch.setattr(firmware.urllib.request, "urlopen", fake_urlopen) + first = firmware.download_v500_donor("gk7205v510") + again = firmware.download_v500_donor("gk7205v510") + assert first == again == get_cache_dir() / "u-boot-gk7205v510-nor.bin" + assert urls == [ + "https://github.com/OpenIPC/u-boot-xmedia/releases/download/latest/" + "u-boot-gk7205v510-nor.bin" + ] diff --git a/tests/test_protocol_v500.py b/tests/test_protocol_v500.py index 6930e43..c193058 100644 --- a/tests/test_protocol_v500.py +++ b/tests/test_protocol_v500.py @@ -152,3 +152,71 @@ async def test_send_firmware_with_acks(self): assert Stage.HEAD_AREA in result.stages_completed assert Stage.AUX_AREA in result.stages_completed assert Stage.BOOT_IMAGE in result.stages_completed + + +def _donor(aux_len: int = 0x5000, code_len: int = 0x800) -> bytes: + from defib.protocol.hisilicon_v500 import V500_BOOT_TAIL_LEN, V500_KEY_AREA_LEN + + total = V500_KEY_AREA_LEN + aux_len + code_len + V500_BOOT_TAIL_LEN + image = bytearray(b"\xa5" * total) + struct.pack_into("<6I", image, 0x400, aux_len, code_len, + code_len + V500_BOOT_TAIL_LEN, 0x12345678, 0x12345678, 0x40707000) + return bytes(image) + + +class TestWrapV500Payload: + def test_layout(self): + from defib.protocol.hisilicon_v500 import V500_AGENT_LOAD_ADDR, wrap_v500_payload + + donor = _donor() + agent = b"\x00\x00\x00\xea" + b"\x11" * 1000 + image = wrap_v500_payload(donor, agent, V500_AGENT_LOAD_ADDR) + + # Header, params and aux (DDR init) area come from the donor. + assert image[:0x400] == donor[:0x400] + assert image[0x40c:0x7000] == donor[0x40c:0x7000] + # Boot code is the agent, padded to 1 KiB, followed by the zero tail. + assert image[0x7000:0x7000 + len(agent)] == agent + assert len(image) == 0x7000 + 0x400 + 0x200 + assert image[0x7000 + len(agent):] == b"\x00" * (len(image) - 0x7000 - len(agent)) + aux, code, total = struct.unpack_from("<3I", image, 0x400) + assert (aux, code, total) == (0x5000, 0x400, 0x600) + + def test_load_address_must_match_code_offset(self): + from defib.protocol.hisilicon_v500 import wrap_v500_payload + + with pytest.raises(ValueError, match="linked at 0x40707000"): + wrap_v500_payload(_donor(), b"\x00" * 16, 0x40707000) + # A donor with a different aux area moves the code, so the same + # agent no longer fits. + with pytest.raises(ValueError, match="runs at 0x41006000"): + wrap_v500_payload(_donor(aux_len=0x4000), b"\x00" * 16, 0x41007000) + + @pytest.mark.parametrize("aux_len", [0, 0x123, 0x100000]) + def test_rejects_implausible_donor(self, aux_len): + from defib.protocol.hisilicon_v500 import wrap_v500_payload + + donor = bytearray(_donor()) + struct.pack_into("II", len(image), 0x41000000) in sent