feat(flash): pure-Python firmware flasher over the OLD-frame protocol

Drives the full flash flow over the existing USB-serial link — enter
bootloader, arm window, stream the ELF block-by-block, reset — with no
external pm3-flash/DFU/JTAG.

- core/flash.py: Flasher (device_info, auto OS->bootloader handover,
  flash_image block loop with ACK/NACK+EFC status, reset, end-to-end flash);
  ELF32-ARM PT_LOAD parsing; 0xFF-padded block alignment; image resolver.
- core/transport.py: OLD 544-byte frame encode/decode, send_old /
  send_old_no_response, magic-branching frame reader, reopen() for USB
  re-enumeration.
- core/protocol.py: DEVICE_INFO flags, START_FLASH_MAGIC, flash geometry,
  BL version helpers.
- Wired as pm3.flasher; reachable through the sync proxy.

Safety: fullimage-only by default; bootrom region refused unless
allow_bootrom=True; every block bounds-checked against the flash window.
Verification is per-block ACK/NACK (as the C flasher does); read-back verify
and hardware validation are follow-ups. 22 tests; suite 1026 pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

pm3py/core/client.py

@@ -15,6 +15,7 @@ from .hf_iso14a import HF14ACommands
 from .hf_iso15 import HF15Commands
 from .hf_mfc import HFMFCommands
 from .hf_mfu import HFMFUCommands
+from .flash import Flasher
 
 log = logging.getLogger(__name__)
 
@@ -74,6 +75,7 @@ class Proxmark3:
         self.hf.iso15 = HF15Commands(self._transport)
         self.hf.mf = HFMFCommands(self._transport)
         self.hf.mfu = HFMFUCommands(self._transport)
+        self.flasher = Flasher(self._transport)
         self.firmware = FirmwareInfo()
 
     async def connect(self) -> None:

pm3py/core/flash.py

@@ -0,0 +1,274 @@
+"""Pure-Python Proxmark3 firmware flasher.
+
+Drives the whole flash flow over the existing USB-serial link — enter bootloader,
+arm the write window, stream the image block by block, reset — with no external
+`pm3-flash`/DFU/JTAG tooling.
+
+Flashing uses the legacy fixed-size OLD frame (not the NG frames the rest of the
+client speaks); the OLD-frame plumbing lives in transport.py. See
+docs/plans for the protocol map.
+
+Safety: fullimage-only by default. The bootrom region ([0x100000, 0x102000)) is
+never written unless ``allow_bootrom=True`` is passed explicitly — a botched
+bootrom write bricks the device to JTAG/SAM-BA recovery, and fork updates only
+ever change the OS image anyway.
+"""
+import os
+import struct
+from pathlib import Path
+
+from .protocol import (
+    Cmd,
+    OLD_FRAME_SIZE,
+    DEVICE_INFO_FLAG_BOOTROM_PRESENT,
+    DEVICE_INFO_FLAG_OSIMAGE_PRESENT,
+    DEVICE_INFO_FLAG_CURRENT_MODE_BOOTROM,
+    DEVICE_INFO_FLAG_CURRENT_MODE_OS,
+    START_FLASH_MAGIC,
+    FLASH_START, BOOTLOADER_END, FLASH_END_256K, FLASH_END_512K,
+    FLASH_BLOCK_SIZE, BL_VERSION_1_0_0,
+)
+from .transport import PM3Transport, PM3Response, PM3Error
+
+# ELF constants
+_ELF_MAGIC = b"\x7fELF"
+_ELFCLASS32 = 1
+_ELFDATA2LSB = 1
+_ET_EXEC = 2
+_EM_ARM = 40
+_PT_LOAD = 1
+
+# Where to look for a locally-built image when none is given (dev checkout).
+_DEFAULT_IMAGE_CANDIDATES = (
+    "firmware/armsrc/obj/fullimage.elf",
+)
+
+
+class FlashError(PM3Error):
+    """Firmware flashing failed."""
+
+
+def parse_elf_segments(image: bytes) -> list[tuple[int, bytes]]:
+    """Extract loadable segments from a PM3 firmware ELF.
+
+    Returns a list of (physical_address, data) for every PT_LOAD program header
+    with a non-zero file size, sorted by address. Validates that the file is a
+    little-endian 32-bit ARM executable, matching the C flasher's gate.
+    """
+    if image[:4] != _ELF_MAGIC:
+        raise FlashError("Not an ELF file (bad magic)")
+    if image[4] != _ELFCLASS32:
+        raise FlashError("Not a 32-bit ELF (ELFCLASS32 expected)")
+    if image[5] != _ELFDATA2LSB:
+        raise FlashError("Not a little-endian ELF (ELFDATA2LSB expected)")
+
+    e_type, e_machine = struct.unpack_from("<HH", image, 16)
+    if e_type != _ET_EXEC:
+        raise FlashError("ELF is not ET_EXEC")
+    if e_machine != _EM_ARM:
+        raise FlashError("ELF is not EM_ARM")
+
+    e_phoff = struct.unpack_from("<I", image, 28)[0]
+    e_phentsize, e_phnum = struct.unpack_from("<HH", image, 42)
+
+    segments: list[tuple[int, bytes]] = []
+    for i in range(e_phnum):
+        off = e_phoff + i * e_phentsize
+        (p_type, p_offset, p_vaddr, p_paddr,
+         p_filesz, p_memsz, p_flags, p_align) = struct.unpack_from("<IIIIIIII", image, off)
+        if p_type != _PT_LOAD or p_filesz == 0:
+            continue
+        segments.append((p_paddr, image[p_offset:p_offset + p_filesz]))
+
+    if not segments:
+        raise FlashError("No loadable (PT_LOAD) segments in ELF")
+    segments.sort()
+    return segments
+
+
+def _align_down(x: int, a: int) -> int:
+    return x & ~(a - 1)
+
+
+def _align_up(x: int, a: int) -> int:
+    return (x + a - 1) & ~(a - 1)
+
+
+def build_blocks(segments: list[tuple[int, bytes]],
+                 block_size: int = FLASH_BLOCK_SIZE) -> list[tuple[int, bytes]]:
+    """Turn segments into a list of (address, block) writes.
+
+    Each segment is padded out to ``block_size`` boundaries with 0xFF (erased
+    flash), then split into block-sized chunks. Addresses are block-aligned so
+    each chunk maps onto whole flash pages.
+    """
+    blocks: list[tuple[int, bytes]] = []
+    for paddr, data in segments:
+        start = _align_down(paddr, block_size)
+        pre = paddr - start
+        end = _align_up(paddr + len(data), block_size)
+        post = end - (paddr + len(data))
+        blob = b"\xFF" * pre + data + b"\xFF" * post
+        for i in range(0, len(blob), block_size):
+            blocks.append((start + i, blob[i:i + block_size]))
+    return blocks
+
+
+def resolve_image(image=None) -> bytes:
+    """Resolve a firmware image to raw ELF bytes.
+
+    Accepts raw bytes, a path, or ``None``. With ``None`` it searches for a
+    locally-built image: the ``PM3PY_FIRMWARE`` env var, then conventional build
+    output paths relative to the working directory. (Auto-download of the pinned
+    release is a separate, later layer that also feeds this function.)
+    """
+    if isinstance(image, (bytes, bytearray)):
+        return bytes(image)
+    if image is not None:
+        p = Path(image)
+        if not p.exists():
+            raise FlashError(f"Firmware image not found: {p}")
+        return p.read_bytes()
+
+    candidates = []
+    env = os.environ.get("PM3PY_FIRMWARE")
+    if env:
+        candidates.append(Path(env))
+    candidates.extend(Path(c) for c in _DEFAULT_IMAGE_CANDIDATES)
+    for p in candidates:
+        if p.exists():
+            return p.read_bytes()
+    raise FlashError(
+        "No firmware image given and no local build found. Build the firmware "
+        "(make) and/or set PM3PY_FIRMWARE, or pass image=<path to fullimage.elf>."
+    )
+
+
+class Flasher:
+    """Flash Proxmark3 firmware over the wire (OLD-frame bootloader protocol)."""
+
+    def __init__(self, transport: PM3Transport):
+        self._t = transport
+
+    @staticmethod
+    def _is(resp: PM3Response, cmd: int) -> bool:
+        """True if an OLD reply carries the given command (low 16 bits)."""
+        return (resp.cmd & 0xFFFF) == cmd
+
+    async def device_info(self) -> dict:
+        """Probe device mode/capabilities via CMD_DEVICE_INFO."""
+        resp = await self._t.send_old(Cmd.DEVICE_INFO)
+        flags = resp.oldarg[0] if resp.oldarg else 0
+        return {
+            "flags": flags,
+            "in_bootrom": bool(flags & DEVICE_INFO_FLAG_CURRENT_MODE_BOOTROM),
+            "in_os": bool(flags & DEVICE_INFO_FLAG_CURRENT_MODE_OS),
+            "bootrom_present": bool(flags & DEVICE_INFO_FLAG_BOOTROM_PRESENT),
+            "osimage_present": bool(flags & DEVICE_INFO_FLAG_OSIMAGE_PRESENT),
+        }
+
+    async def bl_version(self) -> int:
+        """Read the bootloader version code (packed BL_MAKE_VERSION)."""
+        resp = await self._t.send_old(Cmd.BL_VERSION)
+        return resp.oldarg[0] if resp.oldarg else 0
+
+    async def enter_bootloader(self, reopen: bool = True) -> dict:
+        """Put the device into the bootloader, ready to accept flash writes.
+
+        If already in the bootrom, returns immediately. From the OS this issues
+        the automatic CMD_START_FLASH handover (no button press) and re-opens the
+        port after the device re-enumerates.
+        """
+        info = await self.device_info()
+        if info["in_bootrom"]:
+            return info
+        if info["in_os"] and not info["bootrom_present"]:
+            raise FlashError(
+                "Device is in the OS but reports no bootrom — cannot auto-enter "
+                "the bootloader; hold the button while replugging to flash."
+            )
+        # Modern handover: CMD_START_FLASH with zero args -> reset into bootrom.
+        await self._t.send_old_no_response(Cmd.START_FLASH)
+        if reopen:
+            await self._t.reopen()
+        info = await self.device_info()
+        if not info["in_bootrom"]:
+            raise FlashError("Device did not enter the bootloader after handover")
+        return info
+
+    async def flash_image(self, image, *, allow_bootrom: bool = False,
+                          on_progress=None) -> dict:
+        """Write an image to flash. Device must already be in the bootloader.
+
+        Returns {'success', 'blocks', 'bytes', 'flash_end'}.
+        """
+        raw = resolve_image(image)
+        segments = parse_elf_segments(raw)
+
+        # Safety gate: never touch the bootrom region unless explicitly allowed.
+        for paddr, data in segments:
+            if paddr < BOOTLOADER_END and not allow_bootrom:
+                raise FlashError(
+                    f"Image writes the bootrom region (0x{paddr:08X}); refusing. "
+                    f"Pass allow_bootrom=True only if you really mean to reflash "
+                    f"the bootloader (brick risk)."
+                )
+
+        blocks = build_blocks(segments)
+
+        # Choose the write window. Only reach into the second bank (>256 KB) on a
+        # bootloader new enough to drive it.
+        blv = await self.bl_version()
+        flash_end = FLASH_END_512K if blv >= BL_VERSION_1_0_0 else FLASH_END_256K
+
+        low = FLASH_START if allow_bootrom else BOOTLOADER_END
+        magic = START_FLASH_MAGIC if allow_bootrom else 0
+        for addr, block in blocks:
+            if addr < low or addr + len(block) > flash_end:
+                raise FlashError(
+                    f"Block at 0x{addr:08X} is outside the flash window "
+                    f"[0x{low:08X}, 0x{flash_end:08X}); wrong image for this device?"
+                )
+
+        # Arm the write window.
+        resp = await self._t.send_old(Cmd.START_FLASH, low, flash_end, magic)
+        if not self._is(resp, Cmd.ACK):
+            raise FlashError(f"START_FLASH not acknowledged (reply cmd=0x{resp.cmd:X})")
+
+        written = 0
+        total = len(blocks)
+        for i, (addr, block) in enumerate(blocks):
+            resp = await self._t.send_old(Cmd.FINISH_WRITE, addr, 0, 0, block)
+            if self._is(resp, Cmd.NACK):
+                sr = resp.oldarg[0] if resp.oldarg else 0
+                raise FlashError(
+                    f"Write failed at 0x{addr:08X} (EFC status 0x{sr:X})")
+            if not self._is(resp, Cmd.ACK):
+                raise FlashError(
+                    f"Unexpected reply writing 0x{addr:08X} (cmd=0x{resp.cmd:X})")
+            written += len(block)
+            if on_progress:
+                on_progress({"block": i + 1, "total": total,
+                             "addr": addr, "bytes": written})
+
+        return {"success": True, "blocks": total, "bytes": written,
+                "flash_end": flash_end}
+
+    async def reset(self) -> None:
+        """Reboot the device (bootrom then jumps into the OS image)."""
+        await self._t.send_old_no_response(Cmd.HARDWARE_RESET)
+
+    async def flash(self, image=None, *, allow_bootrom: bool = False,
+                    on_progress=None, reset: bool = True) -> dict:
+        """End-to-end: enter bootloader, write the image, reboot.
+
+        ``image`` may be raw ELF bytes, a path to a ``fullimage.elf``, or ``None``
+        to use a locally-built image (see resolve_image).
+        """
+        raw = resolve_image(image)
+        await self.enter_bootloader()
+        result = await self.flash_image(raw, allow_bootrom=allow_bootrom,
+                                        on_progress=on_progress)
+        if reset:
+            await self.reset()
+        return result

pm3py/core/protocol.py

@@ -25,6 +25,42 @@ RESP_POSTAMBLE_SIZE = 2
 MIX_ARGS_FMT = "<QQQ"
 MIX_ARGS_SIZE = 24
 
+# --- Bootloader / flashing: legacy OLD frame ---
+# PacketCommandOLD / PacketResponseOLD: uint64 cmd + uint64 arg[3] + uint8 d[512].
+# No preamble magic, no CRC, no length field — always this fixed size on the wire.
+OLD_FRAME_FMT = "<QQQQ"          # cmd + arg0 + arg1 + arg2 (32 bytes), then 512 data bytes
+OLD_FRAME_HDR_SIZE = 32
+OLD_FRAME_SIZE = OLD_FRAME_HDR_SIZE + PM3_CMD_DATA_SIZE  # 544
+
+# CMD_DEVICE_INFO reply flags (carried in arg[0]), from pm3_cmd.h
+DEVICE_INFO_FLAG_BOOTROM_PRESENT = 1 << 0
+DEVICE_INFO_FLAG_OSIMAGE_PRESENT = 1 << 1
+DEVICE_INFO_FLAG_CURRENT_MODE_BOOTROM = 1 << 2
+DEVICE_INFO_FLAG_CURRENT_MODE_OS = 1 << 3
+DEVICE_INFO_FLAG_UNDERSTANDS_START_FLASH = 1 << 4
+DEVICE_INFO_FLAG_UNDERSTANDS_CHIP_INFO = 1 << 5
+DEVICE_INFO_FLAG_UNDERSTANDS_VERSION = 1 << 6
+DEVICE_INFO_FLAG_UNDERSTANDS_READ_MEM = 1 << 7
+
+# Required in CMD_START_FLASH arg[2] to unlock writes to the bootrom region ("DOIT")
+START_FLASH_MAGIC = 0x54494F44
+
+# AT91SAM7 flash geometry / memory map (bytes)
+FLASH_START = 0x00100000      # start of internal flash
+BOOTLOADER_END = 0x00102000   # end of bootrom / start of OS image (osimage origin)
+FLASH_END_256K = 0x00140000   # window end on a 256 KB part
+FLASH_END_512K = 0x00180000   # window end on a 512 KB part
+FLASH_PAGE_SIZE = 256         # SAM7 flash page
+FLASH_BLOCK_SIZE = 512        # one CMD_FINISH_WRITE programs 2 pages = one 512-byte block
+
+
+def bl_version_code(major: int, minor: int, patch: int) -> int:
+    """Pack a bootloader version the way the firmware does (BL_MAKE_VERSION)."""
+    return (major << 22) | (minor << 12) | patch
+
+
+BL_VERSION_1_0_0 = bl_version_code(1, 0, 0)
+
 # --- CRC-16/A (ISO 14443-3) ---
 _CRC_TABLE: list[int] = []
 

pm3py/core/transport.py

@@ -11,6 +11,7 @@ from .protocol import (
     CMD_PREAMBLE_SIZE, CMD_POSTAMBLE_SIZE,
     RESP_PREAMBLE_SIZE, RESP_POSTAMBLE_SIZE,
     PM3_CMD_DATA_SIZE, MIX_ARGS_SIZE,
+    OLD_FRAME_FMT, OLD_FRAME_HDR_SIZE, OLD_FRAME_SIZE,
 )
 
 
@@ -93,6 +94,30 @@ def decode_response_frame(data: bytes) -> dict:
     return result
 
 
+def encode_old_frame(cmd: int, arg0: int = 0, arg1: int = 0, arg2: int = 0,
+                     data: bytes = b"") -> bytes:
+    """Encode a legacy OLD command frame (used by the bootloader / flasher).
+
+    Fixed 544 bytes: uint64 cmd + uint64 arg[3] + uint8 d[512]. No magic, no CRC.
+    """
+    if len(data) > PM3_CMD_DATA_SIZE:
+        raise ValueError(f"OLD frame data too large: {len(data)} > {PM3_CMD_DATA_SIZE}")
+    header = struct.pack(OLD_FRAME_FMT, cmd, arg0, arg1, arg2)
+    return header + data + bytes(PM3_CMD_DATA_SIZE - len(data))
+
+
+def decode_old_frame(raw: bytes) -> dict:
+    """Decode a 544-byte OLD response frame into a dict (cmd, oldarg, data)."""
+    if len(raw) < OLD_FRAME_SIZE:
+        raise ValueError(f"OLD frame too short: {len(raw)} < {OLD_FRAME_SIZE}")
+    cmd, arg0, arg1, arg2 = struct.unpack_from(OLD_FRAME_FMT, raw, 0)
+    return {
+        "cmd": cmd,
+        "oldarg": [arg0, arg1, arg2],
+        "data": raw[OLD_FRAME_HDR_SIZE:OLD_FRAME_SIZE],
+    }
+
+
 class PM3Error(Exception):
     """Proxmark3 communication error."""
     def __init__(self, message: str, status: int = 0):
@@ -234,6 +259,84 @@ class PM3Transport:
             frame = encode_ng_frame(cmd, payload)
             await self.send_frame(frame)
 
+    async def send_old(self, cmd: int, arg0: int = 0, arg1: int = 0, arg2: int = 0,
+                       data: bytes = b"", timeout: float = 5.0) -> PM3Response:
+        """Send a legacy OLD frame (bootloader/flasher) and read one reply.
+
+        The reply may be an OLD frame (from the bootrom) or an NG frame (from the
+        OS) — the reader branches on the leading magic.
+        """
+        async with self._lock:
+            frame = encode_old_frame(cmd, arg0, arg1, arg2, data)
+            await self.send_frame(frame)
+            return await self._read_any_response(timeout=timeout)
+
+    async def send_old_no_response(self, cmd: int, arg0: int = 0, arg1: int = 0,
+                                   arg2: int = 0, data: bytes = b"") -> None:
+        """Send an OLD frame without waiting (e.g. START_FLASH handover, RESET)."""
+        async with self._lock:
+            frame = encode_old_frame(cmd, arg0, arg1, arg2, data)
+            await self.send_frame(frame)
+
+    async def _read_any_response(self, timeout: float = 5.0) -> PM3Response:
+        """Read one frame that may be NG or OLD and return a PM3Response."""
+        if not self._reader:
+            raise PM3Error("Not connected")
+        try:
+            raw = await asyncio.wait_for(self._read_any_frame(), timeout=timeout)
+        except asyncio.TimeoutError:
+            raise PM3Error("Response timeout", status=-4)
+        magic = struct.unpack_from("<I", raw, 0)[0]
+        if magic == RESP_PREAMBLE_MAGIC:
+            return PM3Response.from_dict(decode_response_frame(raw))
+        d = decode_old_frame(raw)
+        return PM3Response(cmd=d["cmd"], status=0, reason=0, ng=False,
+                           data=d["data"], oldarg=d["oldarg"])
+
+    async def _read_any_frame(self) -> bytes:
+        """Read a complete frame, branching NG vs OLD on the leading 4-byte magic.
+
+        Unlike _read_response_frame this does NOT scan/discard — the bootloader
+        is queried on a freshly (re)opened port with no in-flight junk.
+        """
+        assert self._reader is not None
+        head = await self._reader.readexactly(4)
+        magic = struct.unpack_from("<I", head, 0)[0]
+        if magic == RESP_PREAMBLE_MAGIC:
+            buf = bytearray(head)
+            buf.extend(await self._reader.readexactly(RESP_PREAMBLE_SIZE - 4))
+            payload_len = struct.unpack_from("<H", buf, 4)[0] & 0x7FFF
+            buf.extend(await self._reader.readexactly(payload_len + RESP_POSTAMBLE_SIZE))
+            return bytes(buf)
+        # OLD frame — fixed 544 bytes total, no magic/CRC/length
+        rest = await self._reader.readexactly(OLD_FRAME_SIZE - 4)
+        return head + rest
+
+    async def reopen(self, wait: float = 1.0, retries: int = 60,
+                     retry_delay: float = 1.0, port: str | None = None) -> None:
+        """Close and reopen the serial port, tolerating USB re-enumeration.
+
+        The device drops off the bus during the OS->bootloader handover and after
+        HARDWARE_RESET, then reappears (VID/PID or path may shift). Mirrors the C
+        client's close / wait / poll-reopen sequence.
+        """
+        try:
+            await self.disconnect()
+        except Exception:
+            pass
+        if port is not None:
+            self.port = port
+        await asyncio.sleep(wait)
+        last_err: Exception | None = None
+        for _ in range(retries):
+            try:
+                await self.connect()
+                return
+            except Exception as e:  # port not back yet
+                last_err = e
+                await asyncio.sleep(retry_delay)
+        raise PM3Error(f"Device did not reappear after reset: {last_err}")
+
     async def download_bigbuf(self, offset: int = 0, length: int = PM3_CMD_DATA_SIZE,
                               timeout: float = 4.0) -> tuple[bytes, int]:
         """Download data from device BigBuf.

tests/test_flash.py

@@ -0,0 +1,272 @@
+"""Tests for the pure-Python firmware flasher (no hardware)."""
+import asyncio
+import struct
+import pytest
+from unittest.mock import AsyncMock
+
+from pm3py.core.protocol import (
+    Cmd,
+    OLD_FRAME_SIZE, START_FLASH_MAGIC,
+    FLASH_START, BOOTLOADER_END, FLASH_END_512K, BL_VERSION_1_0_0,
+    DEVICE_INFO_FLAG_BOOTROM_PRESENT, DEVICE_INFO_FLAG_OSIMAGE_PRESENT,
+    DEVICE_INFO_FLAG_CURRENT_MODE_BOOTROM, DEVICE_INFO_FLAG_CURRENT_MODE_OS,
+)
+from pm3py.core.transport import PM3Response, encode_old_frame, decode_old_frame
+from pm3py.core.flash import (
+    Flasher, FlashError, parse_elf_segments, build_blocks, resolve_image,
+)
+
+OS_FLAGS = (DEVICE_INFO_FLAG_BOOTROM_PRESENT | DEVICE_INFO_FLAG_OSIMAGE_PRESENT
+            | DEVICE_INFO_FLAG_CURRENT_MODE_OS)
+BOOTROM_FLAGS = (DEVICE_INFO_FLAG_BOOTROM_PRESENT | DEVICE_INFO_FLAG_OSIMAGE_PRESENT
+                 | DEVICE_INFO_FLAG_CURRENT_MODE_BOOTROM)
+
+
+def run(coro):
+    # Match the rest of the suite's idiom — asyncio.run() would close and null the
+    # shared global loop, breaking later tests that use get_event_loop().
+    return asyncio.get_event_loop().run_until_complete(coro)
+
+
+def old_resp(cmd, arg0=0, arg1=0, arg2=0):
+    return PM3Response(cmd=int(cmd), status=0, reason=0, ng=False,
+                       data=b"\x00" * 512, oldarg=[arg0, arg1, arg2])
+
+
+def make_elf(paddr, data, *, machine=40, etype=2, cls=1, endian=1):
+    """Build a minimal single-PT_LOAD ELF32 image for tests."""
+    ident = b"\x7fELF" + bytes([cls, endian, 1, 0]) + bytes(8)
+    e_phoff = 52
+    e_phentsize = 32
+    header = ident + struct.pack(
+        "<HHIIIIIHHHHHH",
+        etype, machine, 1, paddr, e_phoff, 0, 0,
+        52, e_phentsize, 1, 0, 0, 0,
+    )
+    data_off = e_phoff + e_phentsize
+    phdr = struct.pack("<IIIIIIII",
+                       1,           # PT_LOAD
+                       data_off,    # p_offset
+                       paddr, paddr,
+                       len(data), len(data),
+                       5, 0x10000)
+    return header + phdr + data
+
+
+# --- OLD frame encoding ---
+
+def test_old_frame_size_and_roundtrip():
+    frame = encode_old_frame(Cmd.FINISH_WRITE, 0x102000, 1, 2, b"\xAB\xCD")
+    assert len(frame) == OLD_FRAME_SIZE == 544
+    d = decode_old_frame(frame)
+    assert d["cmd"] == Cmd.FINISH_WRITE
+    assert d["oldarg"] == [0x102000, 1, 2]
+    assert d["data"][:2] == b"\xAB\xCD"
+    assert d["data"][2:] == b"\x00" * 510  # zero-padded to 512
+
+
+def test_old_frame_rejects_oversized_data():
+    with pytest.raises(ValueError):
+        encode_old_frame(Cmd.FINISH_WRITE, data=b"\x00" * 513)
+
+
+# --- ELF parsing ---
+
+def test_parse_elf_single_segment():
+    segs = parse_elf_segments(make_elf(0x102000, b"hello world"))
+    assert segs == [(0x102000, b"hello world")]
+
+
+def test_parse_elf_rejects_bad_magic():
+    with pytest.raises(FlashError):
+        parse_elf_segments(b"NOTELF" + bytes(100))
+
+
+def test_parse_elf_rejects_non_arm():
+    with pytest.raises(FlashError):
+        parse_elf_segments(make_elf(0x102000, b"x", machine=3))  # EM_386
+
+
+def test_parse_elf_rejects_non_exec():
+    with pytest.raises(FlashError):
+        parse_elf_segments(make_elf(0x102000, b"x", etype=1))  # ET_REL
+
+
+# --- block building ---
+
+def test_build_blocks_aligns_and_pads():
+    # 700 bytes at an aligned address -> 2 blocks, second 0xFF-padded
+    blocks = build_blocks([(0x102000, b"\x11" * 700)])
+    assert len(blocks) == 2
+    assert blocks[0][0] == 0x102000
+    assert blocks[1][0] == 0x102200
+    assert all(len(b) == 512 for _, b in blocks)
+    # tail padding is erased-flash 0xFF
+    assert blocks[1][1][-100:] == b"\xFF" * 100
+
+
+def test_build_blocks_pads_low_side():
+    # unaligned start -> leading 0xFF padding down to the block boundary
+    blocks = build_blocks([(0x102010, b"\x22" * 16)])
+    assert len(blocks) == 1
+    addr, block = blocks[0]
+    assert addr == 0x102000
+    assert block[:0x10] == b"\xFF" * 0x10
+    assert block[0x10:0x20] == b"\x22" * 16
+
+
+# --- resolve_image ---
+
+def test_resolve_image_bytes_passthrough():
+    assert resolve_image(b"abc") == b"abc"
+
+
+def test_resolve_image_path(tmp_path):
+    p = tmp_path / "fw.elf"
+    p.write_bytes(b"\x7fELFdata")
+    assert resolve_image(str(p)) == b"\x7fELFdata"
+
+
+def test_resolve_image_missing_path_errors(tmp_path):
+    with pytest.raises(FlashError):
+        resolve_image(str(tmp_path / "nope.elf"))
+
+
+def test_resolve_image_none_no_build(tmp_path, monkeypatch):
+    monkeypatch.delenv("PM3PY_FIRMWARE", raising=False)
+    monkeypatch.chdir(tmp_path)  # no firmware/ build output here
+    with pytest.raises(FlashError):
+        resolve_image(None)
+
+
+# --- Flasher: probing ---
+
+def test_device_info_parses_flags():
+    t = AsyncMock()
+    t.send_old.return_value = old_resp(Cmd.DEVICE_INFO, arg0=BOOTROM_FLAGS)
+    info = run(Flasher(t).device_info())
+    assert info["in_bootrom"] is True
+    assert info["in_os"] is False
+    assert info["bootrom_present"] is True
+
+
+# --- Flasher: enter_bootloader ---
+
+def test_enter_bootloader_already_in_bootrom():
+    t = AsyncMock()
+    t.send_old.return_value = old_resp(Cmd.DEVICE_INFO, arg0=BOOTROM_FLAGS)
+    info = run(Flasher(t).enter_bootloader())
+    assert info["in_bootrom"] is True
+    t.send_old_no_response.assert_not_called()   # no handover needed
+    t.reopen.assert_not_called()
+
+
+def test_enter_bootloader_from_os_handover():
+    t = AsyncMock()
+    t.send_old.side_effect = [
+        old_resp(Cmd.DEVICE_INFO, arg0=OS_FLAGS),       # first probe: in OS
+        old_resp(Cmd.DEVICE_INFO, arg0=BOOTROM_FLAGS),  # after handover: in bootrom
+    ]
+    run(Flasher(t).enter_bootloader())
+    # handover = START_FLASH with no args, then reopen the re-enumerated port
+    t.send_old_no_response.assert_awaited_once_with(Cmd.START_FLASH)
+    t.reopen.assert_awaited_once()
+
+
+def test_enter_bootloader_fails_if_not_entered():
+    t = AsyncMock()
+    t.send_old.side_effect = [
+        old_resp(Cmd.DEVICE_INFO, arg0=OS_FLAGS),
+        old_resp(Cmd.DEVICE_INFO, arg0=OS_FLAGS),   # still in OS after handover
+    ]
+    with pytest.raises(FlashError):
+        run(Flasher(t).enter_bootloader())
+
+
+# --- Flasher: flash_image ---
+
+def test_flash_image_full_sequence():
+    t = AsyncMock()
+    t.send_old.side_effect = [
+        old_resp(Cmd.BL_VERSION, arg0=BL_VERSION_1_0_0),  # bl_version
+        old_resp(Cmd.ACK),                                # START_FLASH ack
+        old_resp(Cmd.ACK),                                # block 0
+        old_resp(Cmd.ACK),                                # block 1
+    ]
+    image = make_elf(0x102000, b"\x11" * 700)  # 2 blocks
+    result = run(Flasher(t).flash_image(image))
+
+    assert result == {"success": True, "blocks": 2, "bytes": 1024,
+                      "flash_end": FLASH_END_512K}
+
+    calls = t.send_old.call_args_list
+    # START_FLASH: fullimage-only window (low=BOOTLOADER_END, no magic)
+    assert calls[1].args == (Cmd.START_FLASH, BOOTLOADER_END, FLASH_END_512K, 0)
+    # FINISH_WRITE per block at the right addresses, 512-byte payloads
+    assert calls[2].args[0] == Cmd.FINISH_WRITE
+    assert calls[2].args[1] == 0x102000
+    assert len(calls[2].args[4]) == 512
+    assert calls[3].args[1] == 0x102200
+
+
+def test_flash_image_refuses_bootrom_region():
+    t = AsyncMock()
+    image = make_elf(FLASH_START, b"\x00" * 16)  # 0x100000 = bootrom region
+    with pytest.raises(FlashError, match="bootrom"):
+        run(Flasher(t).flash_image(image))
+    t.send_old.assert_not_called()
+
+
+def test_flash_image_allow_bootrom_uses_magic():
+    t = AsyncMock()
+    t.send_old.side_effect = [
+        old_resp(Cmd.BL_VERSION, arg0=BL_VERSION_1_0_0),
+        old_resp(Cmd.ACK),   # START_FLASH
+        old_resp(Cmd.ACK),   # 1 block
+    ]
+    image = make_elf(FLASH_START, b"\x00" * 16)
+    run(Flasher(t).flash_image(image, allow_bootrom=True))
+    # window opens at FLASH_START and carries the unlock magic
+    assert t.send_old.call_args_list[1].args == (
+        Cmd.START_FLASH, FLASH_START, FLASH_END_512K, START_FLASH_MAGIC)
+
+
+def test_flash_image_nack_raises_with_status():
+    t = AsyncMock()
+    t.send_old.side_effect = [
+        old_resp(Cmd.BL_VERSION, arg0=BL_VERSION_1_0_0),
+        old_resp(Cmd.ACK),                 # START_FLASH
+        old_resp(Cmd.NACK, arg0=0x8),      # write NACK, EFC PROGE
+    ]
+    image = make_elf(0x102000, b"\x11" * 16)
+    with pytest.raises(FlashError, match="0x8"):
+        run(Flasher(t).flash_image(image))
+
+
+def test_flash_image_start_flash_not_acked():
+    t = AsyncMock()
+    t.send_old.side_effect = [
+        old_resp(Cmd.BL_VERSION, arg0=BL_VERSION_1_0_0),
+        old_resp(Cmd.NACK),   # window arm rejected
+    ]
+    image = make_elf(0x102000, b"\x11" * 16)
+    with pytest.raises(FlashError, match="START_FLASH"):
+        run(Flasher(t).flash_image(image))
+
+
+# --- Flasher: end-to-end flash() ---
+
+def test_flash_end_to_end_resets():
+    t = AsyncMock()
+    t.send_old.side_effect = [
+        old_resp(Cmd.DEVICE_INFO, arg0=BOOTROM_FLAGS),   # enter_bootloader probe
+        old_resp(Cmd.BL_VERSION, arg0=BL_VERSION_1_0_0),
+        old_resp(Cmd.ACK),   # START_FLASH
+        old_resp(Cmd.ACK),   # block 0
+        old_resp(Cmd.ACK),   # block 1
+    ]
+    image = make_elf(0x102000, b"\x11" * 700)
+    result = run(Flasher(t).flash(image))
+    assert result["success"] is True
+    # reboot into the freshly written OS
+    t.send_old_no_response.assert_awaited_with(Cmd.HARDWARE_RESET)