dump-flash, restore: stage in the board's own DRAM

src/defib/cli/app.py

@@ -731,7 +731,7 @@ def dump_flash_cmd(
     output_file: str = typer.Option("flash_dump.bin", "-o", "--output", help="Output binary file"),
     size: str = typer.Option("", "--size", help="Flash size (e.g., 8MB, 16MB) — auto-detect if empty"),
     output: str = typer.Option("human", "--output-mode", help="Output mode: human, json"),
-    chip: str = typer.Option("", "-c", "--chip", help="Chip model. Only needed for chips that dump over USB rather than a U-Boot console."),
+    chip: str = typer.Option("", "-c", "--chip", help="Chip model. Picks the RAM staging address (otherwise asked from U-Boot); required for chips that dump over USB."),
     partition: str = typer.Option("", "--partition", help="USB-recovery chips: dump one named partition instead of the whole flash"),
     ddr: str = typer.Option("", "--ddr", help="USB-recovery chips: DDR-init blob (rkbin rv1106_ddr_*.bin)"),
     usbplug: str = typer.Option("", "--usbplug", help="USB-recovery chips: usbplug blob (rkbin rv1106_usbplug_*.bin)"),
@@ -830,7 +830,7 @@ def on_log(msg: str) -> None:
     try:
         bytes_dumped = await dump_flash(
             transport, output_file,
-            flash_size=flash_size,
+            flash_size=flash_size, chip=chip,
             on_progress=on_progress, on_log=on_log,
         )
         if output == "human":
@@ -2650,7 +2650,7 @@ async def _restore_async(
 
     from rich.console import Console
 
-    from defib.flashdump import send_command
+    from defib.flashdump import get_ram_staging_addr, send_command
     from defib.recovery.events import LogEvent, ProgressEvent
     from defib.recovery.session import RecoverySession
     from defib.transport.serial_platform import create_transport, normalize_port_name
@@ -3117,7 +3117,9 @@ async def _replace_in_tftp(name: str, data: bytes) -> None:
             tftp_proto._files[name] = data
     try:
 
-        ram_addr = 0x82000000
+        # DRAM starts at 0x40000000 on ev200-era parts; a fixed 0x82000000
+        # there points the TFTP load and every flash write at no memory.
+        ram_addr = get_ram_staging_addr(chip)
         offset = 0
         ubi_partmap: dict[int, tuple[str, int, int]] | None = None
 

src/defib/flashdump.py

@@ -84,15 +84,11 @@
 )
 
 
-def get_ram_staging_addr(chip: str) -> int:
-    """Get safe RAM address for flash staging based on SoC type.
-
-    Looks up the chip's RAM base from the known table, falling back to
-    the profile's U-Boot load address if not found.
-    """
+def _exact_ram_staging_addr(chip: str) -> int | None:
+    """Staging address for a chip known to the RAM table or a profile, else None."""
     # Board/vendor selectors use ``soc:variant`` but RAM geometry belongs to
     # the SoC.  Strip the suffix before the exact table lookup so staging does
-    # not depend on the order of the broad legacy prefix fallbacks below.
+    # not depend on the order of the broad legacy prefix fallbacks.
     chip_lower = chip.lower().split(":", 1)[0]
     # Direct match
     if chip_lower in RAM_BASE:
@@ -102,14 +98,26 @@ def get_ram_staging_addr(chip: str) -> int:
         from defib.profiles.loader import load_profile
         profile = load_profile(chip_lower)
         # The property raises for USB-recovery chips, which have no load
-        # addresses at all — the except below then falls through to the
-        # prefix lookup, which is the right answer for them.
+        # addresses at all — the caller then falls back to the prefix
+        # lookup, which is the right answer for them.
         uboot_addr = profile.uboot_address
         # Derive RAM base from U-Boot address (aligned to 0x40000000 boundary)
         ram_base = uboot_addr & 0xF0000000
         return ram_base + RAM_STAGING_OFFSET
     except Exception:
-        pass
+        return None
+
+
+def get_ram_staging_addr(chip: str) -> int:
+    """Get safe RAM address for flash staging based on SoC type.
+
+    Looks up the chip's RAM base from the known table, falling back to
+    the profile's U-Boot load address if not found.
+    """
+    exact = _exact_ram_staging_addr(chip)
+    if exact is not None:
+        return exact
+    chip_lower = chip.lower().split(":", 1)[0]
     # Last resort: use profile name for lookup
     for prefix, base in RAM_BASE.items():
         if chip_lower.startswith(prefix[:8]):
@@ -467,6 +475,41 @@ async def _get_device_crc32(
     return None
 
 
+async def detect_ram_staging_addr(
+    transport: Transport,
+    on_log: Callable[[str], None] | None = None,
+) -> int:
+    """Ask the running U-Boot where DRAM is, for callers that know no chip.
+
+    There is no safe guess: DRAM starts at 0x40000000 on ev200-era parts and
+    at 0x80000000 on older ones, and an ``sf read`` aimed at the wrong half
+    wedges U-Boot.  Mainline ``bdinfo`` reports the bank start; vendor builds
+    without it (XM's xm720x0200 U-Boot) usually still carry ``soc=`` in the
+    env.  When neither answers, refuse rather than guess.
+    """
+    resp = await send_command(transport, "bdinfo", timeout=3.0, wait_for="# ")
+    m = re.search(r"(?:->\s*start|memstart)\s*=\s*(0x[0-9a-fA-F]+)", resp)
+    if m:
+        ram_base = int(m.group(1), 16)
+        if on_log:
+            on_log(f"DRAM start from bdinfo: 0x{ram_base:08x}")
+        return ram_base + RAM_STAGING_OFFSET
+
+    resp = await send_command(transport, "printenv soc", timeout=3.0, wait_for="# ")
+    m = re.search(r"(?m)^soc=([\w:.-]+)\s*$", resp)
+    if m:
+        addr = _exact_ram_staging_addr(m.group(1))
+        if addr is not None:
+            if on_log:
+                on_log(f"SoC from U-Boot env: {m.group(1)}")
+            return addr
+
+    raise RuntimeError(
+        "Could not tell where this board's RAM is (no bdinfo, no known soc= "
+        "in the U-Boot env). Pass the chip with -c, e.g. -c gk7205v200."
+    )
+
+
 async def dump_flash(
     transport: Transport,
     output_path: str,
@@ -486,6 +529,8 @@ async def dump_flash(
         output_path: Path to save the binary dump.
         flash_size: Flash size in bytes (auto-detected if None).
         ram_addr: RAM address to load flash contents into.
+        chip: SoC name that picks ``ram_addr`` when it is not given; with
+            neither, the address is asked from U-Boot itself.
         on_progress: Callback(bytes_done, bytes_total).
         on_log: Callback for log messages.
 
@@ -498,10 +543,14 @@ async def dump_flash(
 
     # Resolve RAM staging address
     if ram_addr is None:
+        # Only an exact table/profile hit is trusted: a mistyped -c must not
+        # reach get_ram_staging_addr()'s prefix or 0x82000000 fallbacks.
         if chip:
-            ram_addr = get_ram_staging_addr(chip)
-        else:
-            ram_addr = 0x82000000  # Legacy default, may crash on 0x40000000 boards
+            ram_addr = _exact_ram_staging_addr(chip)
+            if ram_addr is None and on_log:
+                on_log(f"Unknown chip {chip!r}; asking U-Boot where RAM is")
+        if ram_addr is None:
+            ram_addr = await detect_ram_staging_addr(transport, on_log)
     if on_log:
         on_log(f"RAM staging address: 0x{ram_addr:08x}")
 

tests/test_flashdump.py

@@ -425,3 +425,138 @@ async def test_resume_skips_completed_blocks(self):
         finally:
             if os.path.exists(outpath):
                 os.unlink(outpath)
+
+
+class TestDumpFlashRamDetection:
+    """Regression: with no chip, dump-flash staged at 0x82000000 regardless of
+    the board. On a gk7201v200 (DRAM at 0x40000000) that sf read wedged U-Boot.
+    """
+
+    VALID_HEX = "15 05 00 ea fe ff ff ea fe ff ff ea fe ff ff ea"
+
+    def _responses(self, extra: dict[str, str]) -> dict[str, str]:
+        responses = {
+            "sf probe": "# \n",
+            "sf read": "SF: 16 bytes @ 0x0 Read: OK\n# \n",
+            "md.b 0x42000000 10": f"42000000: {self.VALID_HEX}    ................\n# \n",
+            "crc32": "Unknown command\n# \n",
+            "md.b 0x42000000 0x10": f"42000000: {self.VALID_HEX}    ................\n# \n",
+        }
+        responses.update(extra)
+        return responses
+
+    async def _dump(self, transport: "MockTransport", **kwargs: object) -> None:
+        from defib.flashdump import dump_flash
+
+        outpath = os.path.join(tempfile.gettempdir(), "test_ram_detect.bin")
+        try:
+            await dump_flash(transport, outpath, flash_size=16, **kwargs)  # type: ignore[arg-type]
+        finally:
+            if os.path.exists(outpath):
+                os.unlink(outpath)
+
+    @staticmethod
+    def _sent(transport: "MockTransport") -> list[str]:
+        return [d.decode("ascii", errors="replace").strip() for d in transport._tx_log]
+
+    @pytest.mark.asyncio
+    async def test_bdinfo_dram_start_picks_staging(self):
+        transport = MockTransport(self._responses({
+            "bdinfo": (
+                "arch_number = 0x00000000\n"
+                "DRAM bank   = 0x00000000\n"
+                "-> start    = 0x40000000\n"
+                "-> size     = 0x04000000\n"
+                "# \n"
+            ),
+        }))
+        await self._dump(transport)
+        assert "sf read 0x42000000 0x0 0x10" in self._sent(transport)
+
+    @pytest.mark.asyncio
+    async def test_xm_uboot_without_bdinfo_uses_env_soc(self):
+        # XM U-Boot 2020.01 (xm72010200) on gk7201v200: no bdinfo, soc= in env.
+        transport = MockTransport(self._responses({
+            "bdinfo": "Unknown command 'bdinfo' - try 'help'\n# \n",
+            "printenv soc": "soc=gk7201v200\n# \n",
+        }))
+        await self._dump(transport)
+        sent = self._sent(transport)
+        assert "sf read 0x42000000 0x0 0x10" in sent
+        assert not any(cmd.startswith("sf read 0x82") for cmd in sent)
+
+    @pytest.mark.asyncio
+    async def test_unknown_board_refuses_instead_of_guessing(self):
+        transport = MockTransport(self._responses({
+            "bdinfo": "Unknown command 'bdinfo' - try 'help'\n# \n",
+            "printenv soc": "## Error: \"soc\" not defined\n# \n",
+        }))
+        with pytest.raises(RuntimeError, match="-c"):
+            await self._dump(transport)
+        assert not any(cmd.startswith("sf read") for cmd in self._sent(transport))
+
+    @pytest.mark.asyncio
+    async def test_unrecognised_env_soc_refuses(self):
+        # A soc= the profiles do not know must not fall into the prefix guess.
+        transport = MockTransport(self._responses({
+            "bdinfo": "Unknown command 'bdinfo' - try 'help'\n# \n",
+            "printenv soc": "soc=hi3516zz900\n# \n",
+        }))
+        with pytest.raises(RuntimeError, match="-c"):
+            await self._dump(transport)
+
+    @pytest.mark.asyncio
+    async def test_mistyped_chip_falls_back_to_uboot(self):
+        transport = MockTransport(self._responses({
+            "bdinfo": "Unknown command 'bdinfo' - try 'help'\n# \n",
+            "printenv soc": "soc=gk7201v200\n# \n",
+        }))
+        await self._dump(transport, chip="gk7201v201")
+        sent = self._sent(transport)
+        assert "sf read 0x42000000 0x0 0x10" in sent
+        assert not any(cmd.startswith("sf read 0x82") for cmd in sent)
+
+    @pytest.mark.asyncio
+    async def test_mistyped_chip_on_unknown_board_refuses(self):
+        transport = MockTransport(self._responses({
+            "bdinfo": "Unknown command 'bdinfo' - try 'help'\n# \n",
+            "printenv soc": "## Error: \"soc\" not defined\n# \n",
+        }))
+        with pytest.raises(RuntimeError, match="-c"):
+            await self._dump(transport, chip="gk7201v201")
+        assert not any(cmd.startswith("sf read") for cmd in self._sent(transport))
+
+    @pytest.mark.asyncio
+    async def test_chip_skips_detection(self):
+        transport = MockTransport(self._responses({}))
+        await self._dump(transport, chip="gk7201v200")
+        sent = self._sent(transport)
+        assert "sf read 0x42000000 0x0 0x10" in sent
+        assert "bdinfo" not in sent
+
+
+def test_cli_dump_flash_passes_chip_to_uart_dump(monkeypatch):
+    from typer.testing import CliRunner
+
+    import defib.flashdump
+    import defib.transport.serial_platform
+    from defib.cli.app import app
+
+    seen: dict[str, object] = {}
+
+    async def fake_create_transport(port: str) -> MockTransport:
+        return MockTransport()
+
+    async def fake_dump_flash(transport: object, output_path: str, **kwargs: object) -> int:
+        seen.update(kwargs)
+        return 0
+
+    monkeypatch.setattr(defib.transport.serial_platform, "create_transport", fake_create_transport)
+    monkeypatch.setattr(defib.flashdump, "dump_flash", fake_dump_flash)
+
+    result = CliRunner().invoke(app, [
+        "dump-flash", "-p", "/dev/null", "-c", "gk7201v200", "--size", "8MB",
+        "-o", os.path.join(tempfile.gettempdir(), "test_cli_chip.bin"),
+    ])
+    assert result.exit_code == 0, result.output
+    assert seen["chip"] == "gk7201v200"