diff --git a/CMakeLists.txt b/CMakeLists.txt index 64938c2..9066098 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -25,16 +25,28 @@ if(PROJECT_IS_TOP_LEVEL) # Python are missing, only the size tests run. find_program(_host_cc NAMES cc gcc) find_package(Python3 COMPONENTS Interpreter) - if(_host_cc AND Python3_FOUND AND LIBAVR_MCU STREQUAL "atmega328p") - set(TSB_DEVICE ${CMAKE_BINARY_DIR}/tsb_device) + if(_host_cc AND Python3_FOUND) + set(PB_DEVICE ${CMAKE_BINARY_DIR}/pureboot_device) execute_process( COMMAND ${_host_cc} -O2 -I/usr/include/simavr -I/usr/include/simavr/parts - -o ${TSB_DEVICE} ${CMAKE_CURRENT_SOURCE_DIR}/test/device.c - -lsimavr -lsimavrparts -lelf - RESULT_VARIABLE _dev_res ERROR_VARIABLE _dev_err) - if(NOT _dev_res EQUAL 0) - message(STATUS "tsb_device not built (${_dev_err}) — protocol tests skipped") - unset(TSB_DEVICE) + -o ${PB_DEVICE} ${CMAKE_CURRENT_SOURCE_DIR}/test/pureboot_device.c + -lsimavr -lsimavrparts -lelf -lutil + RESULT_VARIABLE _pbdev_res ERROR_VARIABLE _pbdev_err) + if(NOT _pbdev_res EQUAL 0) + message(STATUS "pureboot_device not built (${_pbdev_err}) — protocol tests skipped") + unset(PB_DEVICE) + endif() + if(LIBAVR_MCU STREQUAL "atmega328p") + set(TSB_DEVICE ${CMAKE_BINARY_DIR}/tsb_device) + execute_process( + COMMAND ${_host_cc} -O2 -I/usr/include/simavr -I/usr/include/simavr/parts + -o ${TSB_DEVICE} ${CMAKE_CURRENT_SOURCE_DIR}/test/device.c + -lsimavr -lsimavrparts -lelf + RESULT_VARIABLE _dev_res ERROR_VARIABLE _dev_err) + if(NOT _dev_res EQUAL 0) + message(STATUS "tsb_device not built (${_dev_err}) — protocol tests skipped") + unset(TSB_DEVICE) + endif() endif() endif() endif() @@ -109,12 +121,24 @@ endif() if(LIBAVR_MCU STREQUAL "attiny13a") set(_pb_flash 1024) set(_pb_wrap "") + set(_pb_page 32) + set(_pb_hz 9600000) + set(_pb_baud 57600) + set(_pb_eeprom 64) elseif(LIBAVR_MCU STREQUAL "attiny85") set(_pb_flash 8192) set(_pb_wrap -Wl,--pmem-wrap-around=8k) + set(_pb_page 64) + set(_pb_hz 8000000) + set(_pb_baud 57600) + set(_pb_eeprom 512) else() set(_pb_flash 32768) set(_pb_wrap -Wl,--pmem-wrap-around=32k) + set(_pb_page 128) + set(_pb_hz 16000000) + set(_pb_baud 115200) + set(_pb_eeprom 1024) endif() math(EXPR _pb_base "${_pb_flash} - 512") math(EXPR _pb_base_hex "${_pb_base}" OUTPUT_FORMAT HEXADECIMAL) @@ -133,4 +157,21 @@ if(PROJECT_IS_TOP_LEVEL) add_test(NAME pureboot.size COMMAND ${CMAKE_COMMAND} -DSIZE_TOOL=${CMAKE_SIZE} -DELF=$ -DLIMIT=512 -P ${CMAKE_CURRENT_SOURCE_DIR}/test/check_size.cmake) + + # The protocol test flashes this fixture through the loader with the real + # host tool and expects its banner after the hand-over; a normally linked + # application whose reset vector is what the tinies' surgery re-homes. + if(DEFINED PB_DEVICE) + add_executable(pbapp test/pbapp.cpp) + target_link_libraries(pbapp PRIVATE libavr) + add_custom_command(TARGET pbapp POST_BUILD + COMMAND ${CMAKE_OBJCOPY} -O binary $ $.bin) + add_test(NAME pureboot.protocol + COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test/pbtest.py + ${PB_DEVICE} $ ${LIBAVR_MCU} ${_pb_hz} ${_pb_base_hex} + ${_pb_page} ${_pb_baud} ${_pb_eeprom} $.bin + ${CMAKE_CURRENT_SOURCE_DIR}/pureboot/pureboot.py + ${CMAKE_BINARY_DIR}/pbtest-work) + set_tests_properties(pureboot.protocol PROPERTIES TIMEOUT 180) + endif() endif() diff --git a/pureboot/README.md b/pureboot/README.md index 4a74cc5..b88fdea 100644 --- a/pureboot/README.md +++ b/pureboot/README.md @@ -94,3 +94,30 @@ word just below the loader (`base - 2`, where `G` jumps), and word 0 is rewritten to `rjmp` to the loader base. Every other vector stays the application's. Page 0 is written last, so an interrupted flash leaves word 0 erased and the chip still falls through to the loader on the next reset. + +## Host tool + +`pureboot.py` — Python 3, standard library only (termios drives any tty, +a USB adapter as well as a simavr pty): + + pureboot.py --port /dev/ttyUSB0 --baud 57600 \ + --info --fuses --flash app.hex --timeout 10 + +Operations run in a fixed order within one session: info, fuses, flash +(erase / program / read / verify), EEPROM (erase / program / read / verify), +timeout — then the loader hands over to the application; `--stay` keeps the +session alive instead, and a later invocation reconnects into it (the knock +converges there too). `--flash` and `--eeprom` verify by read-back unless +`--no-verify`; images are raw binary, or Intel HEX by extension. + +## Tests + +Per chip preset, `ctest` runs the 512-byte size gate and the end-to-end +protocol test: a simavr device (`test/pureboot_device.c` — the mega's USART +as a pty; on the tinies a cycle-timed GPIO⇄pty bridge for the software UART, +plus the SPM/NVM module simavr's tiny cores lack) driven by the real host +tool through knock-from-reset, program + verify of both memories, timeout +configuration, session reconnect, an external reset through the patched +vector, and the hand-over to a fixture application whose banner proves the +launch — cross-checked against the simulator's ground-truth memory dumps and +an independent decode of the surgery's rjmp words. diff --git a/pureboot/__pycache__/pureboot.cpython-313.pyc b/pureboot/__pycache__/pureboot.cpython-313.pyc new file mode 100644 index 0000000..3c80528 Binary files /dev/null and b/pureboot/__pycache__/pureboot.cpython-313.pyc differ diff --git a/pureboot/pureboot.py b/pureboot/pureboot.py new file mode 100644 index 0000000..98befab --- /dev/null +++ b/pureboot/pureboot.py @@ -0,0 +1,447 @@ +#!/usr/bin/env python3 +"""pureboot host tool — the smart half of the pureboot protocol (README.md). + +The device exposes primitives; this tool composes them: image loading (raw +binary or Intel HEX), flash programming with read-back verification, erase as +writing 0xff, EEPROM programming, fuse and info readout, activation-timeout +configuration, and — on chips without a hardware boot section — the +reset-vector surgery that re-homes the application's entry through the +trampoline word below the loader, writing page 0 last so an interrupted +flash still falls through to the loader. + +Python standard library only; the serial port is driven with termios, so any +tty works — a USB adapter as well as a simavr pty. +""" + +import argparse +import os +import select +import sys +import termios +import time + +PROMPT = b"+" +PROTOCOL_VERSION = 1 + + +class Error(Exception): + pass + + +# ---------------------------------------------------------------- serial --- + + +class Port: + """A raw serial port with deadline-based reads.""" + + def __init__(self, path, baud): + self.fd = os.open(path, os.O_RDWR | os.O_NOCTTY) + attrs = termios.tcgetattr(self.fd) + attrs[0] = 0 # iflag + attrs[1] = 0 # oflag + attrs[2] = termios.CREAD | termios.CLOCAL | termios.CS8 # cflag + attrs[3] = 0 # lflag + try: + speed = getattr(termios, f"B{baud}") + except AttributeError: + raise Error(f"unsupported baud rate {baud}") from None + attrs[4] = attrs[5] = speed + attrs[6][termios.VMIN] = 0 + attrs[6][termios.VTIME] = 0 + termios.tcsetattr(self.fd, termios.TCSANOW, attrs) + + def close(self): + os.close(self.fd) + + def write(self, data): + os.write(self.fd, data) + + def flush_input(self): + termios.tcflush(self.fd, termios.TCIFLUSH) + + def read_available(self, wait): + """Everything that arrives within `wait` seconds of quiet start.""" + ready, _, _ = select.select([self.fd], [], [], wait) + return os.read(self.fd, 4096) if ready else b"" + + def read_exact(self, count, timeout): + data = b"" + deadline = time.monotonic() + timeout + while len(data) < count: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise Error(f"timeout: got {len(data)} of {count} bytes") + ready, _, _ = select.select([self.fd], [], [], remaining) + if ready: + data += os.read(self.fd, count - len(data)) + return data + + +# -------------------------------------------------------------- protocol --- + + +class Info: + """The 12-byte info block.""" + + def __init__(self, raw): + if len(raw) != 12 or raw[0:2] != b"PB": + raise Error(f"bad info block: {raw.hex()}") + if raw[2] != PROTOCOL_VERSION: + raise Error(f"protocol version {raw[2]}, tool speaks {PROTOCOL_VERSION}") + self.signature = raw[3:6] + self.page = raw[6] + self.base = raw[7] | (raw[8] << 8) + self.eeprom_size = raw[9] | (raw[10] << 8) + self.patch_vector = bool(raw[11] & 1) + self.flash_size = self.base + 512 + + def describe(self): + sig = " ".join(f"{b:02x}" for b in self.signature) + vector = "host-patched reset vector" if self.patch_vector else "hardware boot section" + return ( + f"signature {sig}, page {self.page} B, " + f"app flash {self.base} B (loader at {self.base:#06x}), " + f"EEPROM {self.eeprom_size} B, {vector}" + ) + + +class Loader: + """A pureboot session. Between commands the loader has prompted `+` and + awaits a command byte; every method restores that invariant.""" + + def __init__(self, port): + self.port = port + self.info = None + + def connect(self, wait): + """Knock until the activation window answers, then read the info + block. Also converges when the loader already sits in its command + loop: the knock bytes are ignored-or-executed there, and the drain + absorbs whatever they produced.""" + self.port.flush_input() + deadline = time.monotonic() + wait + while True: + self.port.write(b"pb") + if PROMPT in self.port.read_available(0.4): + break + if time.monotonic() > deadline: + raise Error("no answer — reset the device within its activation window") + while self.port.read_available(0.3): + pass + self.port.write(b"b") + self.info = Info(self.port.read_exact(12, 2.0)) + self._expect_prompt() + return self.info + + def _expect_prompt(self, timeout=2.0): + byte = self.port.read_exact(1, timeout) + if byte != PROMPT: + raise Error(f"expected prompt, got {byte.hex()}") + + def _command(self, tx, reply_len=0, timeout=2.0): + self.port.write(tx) + reply = self.port.read_exact(reply_len, timeout) if reply_len else b"" + self._expect_prompt(timeout) + return reply + + def _stream_read(self, command, address, count): + data = b"" + while count: + chunk = min(count, 256) + head = bytes((ord(command), address & 0xFF, address >> 8, chunk & 0xFF)) + data += self._command(head, chunk, 5.0) + address += chunk + count -= chunk + return data + + def read_flash(self, address, count): + return self._stream_read("R", address, count) + + def read_eeprom(self, address, count): + return self._stream_read("r", address, count) + + def write_page(self, address, data): + assert len(data) == self.info.page and address % self.info.page == 0 + head = bytes((ord("W"), address & 0xFF, address >> 8)) + self._command(head + data, 0, 2.0) + + def write_eeprom(self, address, data): + offset = 0 + while offset < len(data): + chunk = data[offset : offset + 256] + head = bytes((ord("w"), address & 0xFF, address >> 8, len(chunk) & 0xFF)) + self.port.write(head) + for byte in chunk: + self.port.write(bytes((byte,))) + self._expect_prompt() # per-byte ack: the write has begun + self._expect_prompt() # the next command prompt + address += len(chunk) + offset += len(chunk) + + def read_fuses(self): + return self._command(b"F", 4, 2.0) + + def run_application(self): + self.port.write(b"G") + self._expect_prompt() + + +# ---------------------------------------------------------------- images --- + + +def load_image(path): + """Raw binary, or Intel HEX by extension (.hex/.ihx/.ihex).""" + data = open(path, "rb").read() + if not path.lower().endswith((".hex", ".ihx", ".ihex")): + return data + memory = {} + for number, line in enumerate(data.decode("ascii", "replace").splitlines(), 1): + line = line.strip() + if not line: + continue + if not line.startswith(":"): + raise Error(f"{path}:{number}: not an Intel HEX record") + record = bytes.fromhex(line[1:]) + if sum(record) & 0xFF: + raise Error(f"{path}:{number}: checksum mismatch") + count, address, kind = record[0], (record[1] << 8) | record[2], record[3] + payload = record[4 : 4 + count] + if kind == 0: + for i, byte in enumerate(payload): + memory[address + i] = byte + elif kind == 1: + break + elif kind in (2, 4) and not any(payload): + continue # a zero base extends nothing + elif kind in (3, 5): + continue # start address: irrelevant, reset is the entry + else: + raise Error(f"{path}:{number}: record type {kind} reaches beyond the 16-bit space") + if not memory: + raise Error(f"{path}: empty image") + return bytes(memory.get(i, 0xFF) for i in range(max(memory) + 1)) + + +# --------------------------------------------------------------- surgery --- + + +def rjmp_target(word_address, opcode, flash_words): + return (word_address + 1 + (opcode & 0x0FFF)) % flash_words + + +def rjmp_to(word_address, destination, flash_words): + return 0xC000 | ((destination - word_address - 1) % flash_words % 0x1000) + + +def plan_flash(image, info): + """The pages to program, as {page_address: bytes}, already carrying the + reset-vector surgery where the chip needs it. Page 0 must go last — + callers get it separated.""" + page = info.page + limit = info.base - (2 if info.patch_vector else 0) + if len(image) > limit: + raise Error(f"image is {len(image)} B, application flash ends at {limit}") + final = bytearray(image) + bytearray([0xFF] * (-len(image) % page)) + + if info.patch_vector: + flash_words = info.flash_size // 2 + word0 = final[0] | (final[1] << 8) + if word0 & 0xF000 != 0xC000: + raise Error( + "the image's reset vector is not an rjmp — pureboot's vector " + "surgery cannot re-home it (crt-less entry at address 0?)" + ) + entry = rjmp_target(0, word0, flash_words) + if entry >= info.base // 2: + raise Error( + "the image's reset vector already targets the loader — this " + "is a read-back of a patched image; flash the original" + ) + trampoline_word = (info.base - 2) // 2 + patch = rjmp_to(0, info.base // 2, flash_words) + final[0], final[1] = patch & 0xFF, patch >> 8 + trampoline_page = info.base - page + if len(final) < trampoline_page + page: + final += bytearray([0xFF] * (trampoline_page + page - len(final))) + jump = rjmp_to(trampoline_word, entry, flash_words) + final[info.base - 2], final[info.base - 1] = jump & 0xFF, jump >> 8 + + pages = {a: bytes(final[a : a + page]) for a in range(0, len(final), page)} + return pages + + +def covered(pages, skip_blank): + """Pages in programming order: ascending, page 0 last; optionally + dropping all-0xff pages (sound only over erased flash) — never the + load-bearing page 0.""" + rest = [a for a in sorted(pages) if a != 0] + if skip_blank: + rest = [a for a in rest if pages[a].count(0xFF) != len(pages[a])] + return rest + [0] + + +# ------------------------------------------------------------ operations --- + + +def op_erase_flash(loader): + """0xff over the whole application area, page 0 first — an interrupted + erase leaves the entry word blank and the chip still boots the loader.""" + blank = bytes([0xFF] * loader.info.page) + for address in range(0, loader.info.base, loader.info.page): + loader.write_page(address, blank) + print(f"erase: {loader.info.base // loader.info.page} pages") + + +def op_erase_eeprom(loader): + loader.write_eeprom(0, bytes([0xFF] * loader.info.eeprom_size)) + print(f"erase: {loader.info.eeprom_size} B of EEPROM") + + +def op_flash(loader, path, erase, verify): + image = load_image(path) + pages = plan_flash(image, loader.info) + if erase: + op_erase_flash(loader) + order = covered(pages, skip_blank=erase) + for address in order: + loader.write_page(address, pages[address]) + print(f"flash: {path}: {len(order)} pages") + if verify: + verify_pages(loader, pages) + + +def verify_pages(loader, pages): + for address in sorted(pages): + got = loader.read_flash(address, loader.info.page) + if got != pages[address]: + first = next(i for i in range(len(got)) if got[i] != pages[address][i]) + raise Error( + f"verify failed at {address + first:#06x}: " + f"wrote {pages[address][first]:02x}, read {got[first]:02x}" + ) + print(f"verify: {len(pages)} pages ok") + + +def op_verify_flash(loader, path): + verify_pages(loader, plan_flash(load_image(path), loader.info)) + + +def op_read_flash(loader, path): + data = loader.read_flash(0, loader.info.base) + open(path, "wb").write(data) + print(f"read flash: {len(data)} B -> {path}") + + +def op_eeprom(loader, path, erase, verify): + image = load_image(path) + if len(image) > loader.info.eeprom_size: + raise Error(f"EEPROM image is {len(image)} B, device has {loader.info.eeprom_size}") + if erase: + op_erase_eeprom(loader) + loader.write_eeprom(0, image) + print(f"eeprom: {path}: {len(image)} B") + if verify: + got = loader.read_eeprom(0, len(image)) + if got != image: + first = next(i for i in range(len(got)) if got[i] != image[i]) + raise Error(f"verify failed at EEPROM {first:#06x}: wrote {image[first]:02x}, read {got[first]:02x}") + print(f"verify: {len(image)} B ok") + + +def op_verify_eeprom(loader, path): + image = load_image(path) + got = loader.read_eeprom(0, len(image)) + if got != image: + first = next(i for i in range(len(got)) if got[i] != image[i]) + raise Error(f"verify failed at EEPROM {first:#06x}: expected {image[first]:02x}, read {got[first]:02x}") + print(f"verify: {len(image)} B of EEPROM ok") + + +def op_read_eeprom(loader, path): + data = loader.read_eeprom(0, loader.info.eeprom_size) + open(path, "wb").write(data) + print(f"read EEPROM: {len(data)} B -> {path}") + + +def op_timeout(loader, seconds): + loader.write_eeprom(loader.info.eeprom_size - 1, bytes((seconds,))) + label = f"{seconds} s" if seconds else "the device default" + print(f"activation timeout: {label}") + + +def op_fuses(loader): + low, lock, extended, high = loader.read_fuses() + print(f"fuses: low {low:02x} high {high:02x} extended {extended:02x} lock {lock:02x}") + + +# -------------------------------------------------------------------- cli --- + + +def main(): + parser = argparse.ArgumentParser( + description="pureboot host tool", epilog="operations run in the order listed above" + ) + parser.add_argument("--port", required=True, help="serial device (or simavr pty)") + parser.add_argument("--baud", type=int, default=115200, help="115200 mega, 57600 tinies") + parser.add_argument("--wait", type=float, default=30.0, help="seconds to keep knocking") + parser.add_argument("--info", action="store_true", help="print the device info block") + parser.add_argument("--fuses", action="store_true", help="read the fuse and lock bytes") + parser.add_argument("--erase-flash", action="store_true", help="0xff over the application flash") + parser.add_argument("--flash", metavar="FILE", help="program an application (bin or ihex)") + parser.add_argument("--no-verify", action="store_true", help="skip read-back after writes") + parser.add_argument("--read-flash", metavar="FILE", help="dump the application flash") + parser.add_argument("--verify-flash", metavar="FILE", help="compare flash against an image") + parser.add_argument("--erase-eeprom", action="store_true", help="0xff over the EEPROM") + parser.add_argument("--eeprom", metavar="FILE", help="program the EEPROM (bin or ihex)") + parser.add_argument("--read-eeprom", metavar="FILE", help="dump the EEPROM") + parser.add_argument("--verify-eeprom", metavar="FILE", help="compare EEPROM against an image") + parser.add_argument("--timeout", type=int, metavar="S", help="activation window, 1-254 s (0: default)") + parser.add_argument("--stay", action="store_true", help="leave the loader in its session") + args = parser.parse_args() + + if args.timeout is not None and not 0 <= args.timeout <= 254: + parser.error("--timeout must be 0..254 (255 is the erased cell)") + + port = Port(args.port, args.baud) + try: + loader = Loader(port) + info = loader.connect(args.wait) + if args.info: + print(f"device: {info.describe()}") + if args.fuses: + op_fuses(loader) + if args.flash: + op_flash(loader, args.flash, args.erase_flash, not args.no_verify) + elif args.erase_flash: + op_erase_flash(loader) + if args.read_flash: + op_read_flash(loader, args.read_flash) + if args.verify_flash: + op_verify_flash(loader, args.verify_flash) + if args.eeprom: + op_eeprom(loader, args.eeprom, args.erase_eeprom, not args.no_verify) + elif args.erase_eeprom: + op_erase_eeprom(loader) + if args.read_eeprom: + op_read_eeprom(loader, args.read_eeprom) + if args.verify_eeprom: + op_verify_eeprom(loader, args.verify_eeprom) + if args.timeout is not None: + op_timeout(loader, args.timeout) + if args.stay: + print("loader stays in its session (reset to leave)") + else: + loader.run_application() + print("application running") + finally: + port.close() + + +if __name__ == "__main__": + try: + main() + except Error as error: + print(f"error: {error}", file=sys.stderr) + sys.exit(1) + except KeyboardInterrupt: + sys.exit(130) diff --git a/test/pbapp.cpp b/test/pbapp.cpp new file mode 100644 index 0000000..fbe3155 --- /dev/null +++ b/test/pbapp.cpp @@ -0,0 +1,51 @@ +// Test-fixture application for the pureboot protocol test: prints "APP" on +// the chip's serial link (the same link the loader uses) and idles — the +// proof that the loader's hand-over, and on the tinies the host's +// reset-vector surgery, actually launched it. Linked normally (crt, vectors +// at 0); on the tinies its reset vector is the rjmp the host re-homes. +#include + +using namespace avr::literals; + +namespace { + +consteval avr::hertz_t clock() +{ + if (avr::hw::db.name == "ATtiny13A") + return 9.6_MHz; + if (avr::hw::db.name == "ATtiny85") + return 8_MHz; + return 16_MHz; +} + +using dev = avr::device<{.clock = clock()}>; + +template +struct link { + using tx_t = avr::uart::usart0; + static void tx(char c) + { + tx_t::write(static_cast(c)); + } +}; + +template +struct link { + using tx_t = avr::uart::software_tx; + static void tx(char c) + { + tx_t::write(static_cast(c)); + } +}; + +} // namespace + +int main() +{ + avr::init::tx_t>(); + link::tx('A'); + link::tx('P'); + link::tx('P'); + while (true) { + } +} diff --git a/test/pbtest.py b/test/pbtest.py new file mode 100644 index 0000000..f923026 --- /dev/null +++ b/test/pbtest.py @@ -0,0 +1,178 @@ +#!/usr/bin/env python3 +"""End-to-end pureboot protocol test: spawn the simavr device, then drive it +with the real host tool (pureboot.py, as a subprocess over the device's pty) +through flash + EEPROM + timeout + fuse + hand-over scenarios, and cross-check +the tool's view against the simulator's ground-truth memory dumps. + +Usage: pbtest.py + +Exits 0 if every scenario passes. +""" + +import os +import signal +import subprocess +import sys +import time + + +def fail(message): + print(f"FAIL: {message}") + sys.exit(1) + + +def rjmp_decode(word, at, flash_words): + """Where an rjmp word at word-address `at` lands — deliberately written + against the instruction-set definition (12-bit signed offset), not with + the host tool's encoder, so an encoding bug cannot verify itself.""" + if word & 0xF000 != 0xC000: + fail(f"word at {at * 2:#06x} is {word:#06x}, not an rjmp") + offset = word & 0x0FFF + if offset >= 0x800: + offset -= 0x1000 + return (at + 1 + offset) % flash_words + + +class Device: + def __init__(self, binary, elf, mcu, hz, base, page, baud, dump): + self.proc = subprocess.Popen( + [binary, elf, mcu, hz, base, str(page), str(baud), dump], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + self.dump = dump + self.pty = None + deadline = time.time() + 5 + while time.time() < deadline: + line = self.proc.stdout.readline() + if not line: + break + if line.startswith("PB_PTY"): + self.pty = line.split()[1] + break + if not self.pty: + self.stop() + raise RuntimeError("device did not report a pty") + + def stop(self): + self.proc.terminate() + try: + self.proc.wait(timeout=3) + except subprocess.TimeoutExpired: + self.proc.kill() + + +def run_tool(tool, pty, baud, *args): + result = subprocess.run( + [sys.executable, tool, "--port", pty, "--baud", str(baud), "--wait", "20", *args], + capture_output=True, + text=True, + timeout=120, + ) + print(result.stdout, end="") + if result.returncode != 0: + fail(f"tool exited {result.returncode}: {result.stderr.strip()}") + return result.stdout + + +def main(): + (device_bin, elf, mcu, hz, base_hex, page, baud, eeprom_size, app_bin, tool, workdir) = sys.argv[1:] + base, page, baud, eeprom_size = int(base_hex, 0), int(page), int(baud), int(eeprom_size) + sys.path.insert(0, os.path.dirname(os.path.abspath(tool))) + import pureboot as pb + + os.makedirs(workdir, exist_ok=True) + ee_image = bytes(range(0xA0, 0xB0)) + ee_path = os.path.join(workdir, "ee.bin") + open(ee_path, "wb").write(ee_image) + dump = os.path.join(workdir, "flash_dump.bin") + read_flash = os.path.join(workdir, "readback_flash.bin") + read_eeprom = os.path.join(workdir, "readback_eeprom.bin") + + # The geometry the host will discover, for computing the expected image. + info = pb.Info( + bytes([ord("P"), ord("B"), 1, 0, 0, 0, page]) + + bytes([base & 0xFF, base >> 8, eeprom_size & 0xFF, eeprom_size >> 8]) + + bytes([0 if mcu == "atmega328p" else 1]) + ) + + device = Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump) + try: + # Session 1: knock from reset, identify, program everything, stay. + out = run_tool(tool, device.pty, baud, "--info", "--fuses", "--flash", app_bin, + "--eeprom", ee_path, "--timeout", "8", "--stay") + for needed in ("device: signature", "fuses:", "verify:", "activation timeout: 8 s", "stays"): + if needed not in out: + fail(f"session 1 output lacks {needed!r}") + + # Session 2: reconnect into the live session, verify, dump, hand over + # is deferred — the pty must be reopened for the APP banner first. + out = run_tool(tool, device.pty, baud, "--verify-flash", app_bin, "--verify-eeprom", ee_path, + "--read-flash", read_flash, "--read-eeprom", read_eeprom, "--stay") + if out.count("verify:") != 2: + fail("session 2 did not verify both memories") + + eeprom_back = open(read_eeprom, "rb").read() + if eeprom_back[: len(ee_image)] != ee_image: + fail("EEPROM read-back mismatch") + if eeprom_back[-1] != 8: + fail(f"timeout cell reads {eeprom_back[-1]}, expected 8") + + # The expected post-surgery flash, straight from the tool's planner. + pages = pb.plan_flash(open(app_bin, "rb").read(), info) + flash_back = open(read_flash, "rb").read() + for address, data in pages.items(): + if flash_back[address : address + page] != data: + fail(f"flash read-back mismatch in page {address:#06x}") + + # An external reset re-enters through the patched word 0 (tinies; the + # runner resets them to address 0 like silicon) or BOOTRST (mega). + # The loader must answer a fresh knock, and 'G' must land in the + # application, which banners on the same link. + device.proc.send_signal(signal.SIGUSR1) + port = pb.Port(device.pty, baud) + try: + loader = pb.Loader(port) + loader.connect(15) + port.write(b"G") + if port.read_exact(1, 5.0) != pb.PROMPT: + fail("no ack for G") + banner = port.read_exact(3, 5.0) + if banner != b"APP": + fail(f"application banner was {banner!r}") + finally: + port.close() + finally: + device.stop() + + # Ground truth: the simulator's own memories, against the host's view. + flash_true = open(dump, "rb").read() + if flash_true[:base] != flash_back: + fail("host flash read-back differs from the simulator's flash") + if flash_true[base] == 0xFF and flash_true[base + 1] == 0xFF: + fail("loader region looks erased in the ground-truth dump") + + # The surgery, decoded independently: the patched vector must land on the + # loader, the trampoline on the application's own entry. + if mcu != "atmega328p": + flash_words = (base + 512) // 2 + app = open(app_bin, "rb").read() + word0 = flash_true[0] | (flash_true[1] << 8) + if rjmp_decode(word0, 0, flash_words) != base // 2: + fail("patched reset vector does not land on the loader base") + trampoline = flash_true[base - 2] | (flash_true[base - 1] << 8) + original = app[0] | (app[1] << 8) + if rjmp_decode(trampoline, (base - 2) // 2, flash_words) != rjmp_decode(original, 0, flash_words): + fail("trampoline does not land on the application's own entry") + ee_true_path = dump + ".eeprom" + if os.path.exists(ee_true_path): + ee_true = open(ee_true_path, "rb").read() + if ee_true[: len(ee_image)] != ee_image or ee_true[-1] != 8: + fail("ground-truth EEPROM does not match what was programmed") + + print("pbtest: all scenarios pass") + + +if __name__ == "__main__": + main() diff --git a/test/pureboot_device.c b/test/pureboot_device.c new file mode 100644 index 0000000..d14352e --- /dev/null +++ b/test/pureboot_device.c @@ -0,0 +1,309 @@ +// simavr "device" for the pureboot protocol tests, all three chips. Loads +// the boot-linked ELF at the loader base, starts execution there (BOOTRST / +// the patched vector are not what is under test), and exposes the loader's +// serial link as a pty for the real host tool: +// +// - ATmega328P: the hardware USART0 through simavr's uart_pty. +// - Tinies: an 8N1 bridge between a pty and the GPIO software UART +// (drives PB0, the loader's RX; decodes PB1, its TX), timed against the +// simulated cycle counter. +// +// simavr's tiny cores decode the SPM opcode but attach no NVM module — SPM +// is a silent no-op (the mega's boot section has one, avr_flash). The +// missing module is supplied here: the SPM ioctl reads SPMCSR/Z/r1:r0 and +// implements buffer fill, page erase, page write, and CTPB, completing +// instantly. RFLB's LPM diversion (fuse readout) stays unmodeled, so the +// 'F' command answers with flash bytes — the tests assert transport only. +// +// On exit (or SIGTERM) the flash and EEPROM are dumped to files for a +// ground-truth cross-check against what the host read back. +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "avr_eeprom.h" +#include "avr_flash.h" +#include "avr_ioport.h" +#include "avr_uart.h" +#include "sim_avr.h" +#include "sim_elf.h" +#include "sim_io.h" +#include "uart_pty.h" + +static avr_t *avr; +static uart_pty_t uart_pty; +static int use_uart_pty; +static const char *dump_path; +static uint32_t reset_pc; +static volatile sig_atomic_t reset_requested; + +static void request_reset(int sig) +{ + (void)sig; + reset_requested = 1; +} + +// ------------------------------------------------------------- tiny NVM --- + +typedef struct { + avr_io_t io; + uint8_t buffer[128]; + unsigned page; +} tiny_nvm_t; + +static tiny_nvm_t nvm; + +static int nvm_ioctl(avr_io_t *io, uint32_t ctl, void *param) +{ + (void)param; + if (ctl != AVR_IOCTL_FLASH_SPM) + return -1; + tiny_nvm_t *n = (tiny_nvm_t *)io; + avr_t *mcu = io->avr; + uint8_t command = mcu->data[0x57] & 0x1f; // SPMCSR, both tinies + uint16_t z = (uint16_t)(mcu->data[30] | (mcu->data[31] << 8)); + uint32_t page_base = (uint32_t)(z & ~(n->page - 1)) % (mcu->flashend + 1); + if (command == 0x01) { // SPMEN alone: buffer fill from r1:r0 + unsigned offset = z & (n->page - 1) & ~1u; + n->buffer[offset] = mcu->data[0]; + n->buffer[offset + 1] = mcu->data[1]; + } else if (command == 0x03) { // PGERS + memset(mcu->flash + page_base, 0xff, n->page); + } else if (command == 0x05) { // PGWRT: programming only clears bits + for (unsigned i = 0; i < n->page; i++) + mcu->flash[page_base + i] &= n->buffer[i]; + memset(n->buffer, 0xff, n->page); + } else if (command == 0x11) { // CTPB + memset(n->buffer, 0xff, n->page); + } + mcu->data[0x57] &= (uint8_t)~0x1f; // the operation completes instantly + return 0; +} + +// ----------------------------------------------------------- GPIO bridge --- + +static int pty_master = -1; +static avr_irq_t *rx_pin; // the loader's RX (PB0), driven from the pty +static avr_cycle_count_t bit_cycles; + +static int tx_level = 1, tx_active, tx_bit; +static uint8_t tx_shift; + +static avr_cycle_count_t tx_sample(avr_t *mcu, avr_cycle_count_t when, void *param) +{ + (void)mcu; + (void)param; + tx_shift = (uint8_t)((tx_shift >> 1) | (tx_level ? 0x80 : 0)); + if (++tx_bit < 8) + return when + bit_cycles; + if (write(pty_master, &tx_shift, 1) != 1) + fprintf(stderr, "device: pty write lost a byte\n"); + tx_active = 0; + return 0; +} + +static void tx_hook(avr_irq_t *irq, uint32_t value, void *param) +{ + (void)irq; + (void)param; + int level = value & 1; + if (!tx_active && tx_level == 1 && level == 0) { // start edge + tx_active = 1; + tx_bit = 0; + avr_cycle_timer_register(avr, bit_cycles + bit_cycles / 2, tx_sample, NULL); + } + tx_level = level; +} + +static uint8_t rx_queue[8192]; +static unsigned rx_head, rx_tail; // ring: head = next to send +static int rx_active, rx_bit; +static uint8_t rx_byte; + +static void rx_start_next(void); + +static avr_cycle_count_t rx_step(avr_t *mcu, avr_cycle_count_t when, void *param) +{ + (void)mcu; + (void)param; + if (rx_bit < 8) { + avr_raise_irq(rx_pin, (rx_byte >> rx_bit) & 1); + rx_bit++; + return when + bit_cycles; + } + if (rx_bit == 8) { // stop bit, plus one idle bit of margin + avr_raise_irq(rx_pin, 1); + rx_bit++; + return when + 2 * bit_cycles; + } + rx_active = 0; + rx_start_next(); + return 0; +} + +static void rx_start_next(void) +{ + if (rx_active || rx_head == rx_tail) + return; + rx_byte = rx_queue[rx_head]; + rx_head = (rx_head + 1) % sizeof(rx_queue); + rx_active = 1; + rx_bit = 0; + avr_raise_irq(rx_pin, 0); // start bit + avr_cycle_timer_register(avr, bit_cycles, rx_step, NULL); +} + +static void poll_pty(void) +{ + uint8_t chunk[256]; + ssize_t got = read(pty_master, chunk, sizeof(chunk)); + for (ssize_t i = 0; i < got; i++) { + unsigned next = (rx_tail + 1) % sizeof(rx_queue); + if (next == rx_head) + break; // full: the host will retry on timeout + rx_queue[rx_tail] = chunk[i]; + rx_tail = next; + } + if (got > 0) + rx_start_next(); +} + +// ------------------------------------------------------------------ main --- + +static void finish(int sig) +{ + (void)sig; + if (dump_path) { + FILE *f = fopen(dump_path, "wb"); + if (f) { + fwrite(avr->flash, 1, avr->flashend + 1, f); + fclose(f); + } + avr_eeprom_desc_t ee = {.ee = NULL, .offset = 0, .size = 0}; + if (avr_ioctl(avr, AVR_IOCTL_EEPROM_GET, &ee) == 0 && ee.ee && ee.size) { + char path[512]; + snprintf(path, sizeof(path), "%s.eeprom", dump_path); + f = fopen(path, "wb"); + if (f) { + fwrite(ee.ee, 1, ee.size, f); + fclose(f); + } + } + } + if (use_uart_pty) + uart_pty_stop(&uart_pty); + _exit(0); +} + +int main(int argc, char *argv[]) +{ + if (argc != 8) { + fprintf(stderr, "usage: %s \n", argv[0]); + return 2; + } + const char *mcu_name = argv[2]; + uint32_t base = (uint32_t)strtoul(argv[4], NULL, 0); + unsigned page = (unsigned)atoi(argv[5]); + unsigned baud = (unsigned)atoi(argv[6]); + dump_path = argv[7]; + use_uart_pty = strcmp(mcu_name, "atmega328p") == 0; + + avr = avr_make_mcu_by_name(mcu_name); + if (!avr) { + fprintf(stderr, "device: no %s core\n", mcu_name); + return 1; + } + avr_init(avr); + avr->frequency = (uint32_t)strtoul(argv[3], NULL, 0); + memset(avr->flash, 0xff, avr->flashend + 1); // real flash powers up erased + + elf_firmware_t fw = {0}; + if (elf_read_firmware(argv[1], &fw) != 0) { + fprintf(stderr, "device: cannot read %s\n", argv[1]); + return 1; + } + memcpy(avr->flash + base, fw.flash, fw.flashsize); + // The mega enters the loader in hardware (BOOTRST, not modeled); the + // tinies reset to word 0 like silicon — erased flash walks up into the + // loader, and after the host's surgery the patched vector routes there. + reset_pc = use_uart_pty ? base : 0; + avr->pc = reset_pc; + avr->codeend = avr->flashend; + + // Erased EEPROM, as hardware powers up (simavr zeroes it). + uint8_t blank[1024]; + memset(blank, 0xff, sizeof(blank)); + avr_eeprom_desc_t seed = {.ee = blank, .offset = 0, .size = 0}; + if (avr_ioctl(avr, AVR_IOCTL_EEPROM_GET, &seed) == 0 && seed.size <= sizeof(blank)) { + seed.ee = blank; + avr_ioctl(avr, AVR_IOCTL_EEPROM_SET, &seed); + } + + if (use_uart_pty) { + // POLL_SLEEP paces an idle-polling loader in host real time (a + // no-hardware CPU-saving hack); clear it so cycles run free. + uint32_t flags = 0; + avr_ioctl(avr, AVR_IOCTL_UART_GET_FLAGS('0'), &flags); + flags &= ~AVR_UART_FLAG_POLL_SLEEP; + avr_ioctl(avr, AVR_IOCTL_UART_SET_FLAGS('0'), &flags); + uart_pty_init(avr, &uart_pty); + uart_pty_connect(&uart_pty, '0'); + printf("PB_PTY %s\n", uart_pty.pty.slavename); + } else { + nvm.page = page; + memset(nvm.buffer, 0xff, sizeof(nvm.buffer)); + nvm.io.kind = "tiny_nvm"; + nvm.io.ioctl = nvm_ioctl; + avr_register_io(avr, &nvm.io); + + bit_cycles = avr->frequency / baud; + rx_pin = avr_io_getirq(avr, AVR_IOCTL_IOPORT_GETIRQ('B'), 0); + avr_irq_register_notify(avr_io_getirq(avr, AVR_IOCTL_IOPORT_GETIRQ('B'), 1), tx_hook, NULL); + avr_raise_irq(rx_pin, 1); // idle line + + int slave; + struct termios raw; + cfmakeraw(&raw); + if (openpty(&pty_master, &slave, NULL, &raw, NULL) != 0) { + fprintf(stderr, "device: openpty failed\n"); + return 1; + } + fcntl(pty_master, F_SETFL, O_NONBLOCK); + printf("PB_PTY %s\n", ttyname(slave)); + } + fflush(stdout); + + signal(SIGTERM, finish); + signal(SIGINT, finish); + signal(SIGUSR1, request_reset); // an external reset line, for the tests + + long since_poll = 0; + for (;;) { + int state = avr_run(avr); + if (state == cpu_Done || state == cpu_Crashed) + break; + if (reset_requested) { + reset_requested = 0; + avr_reset(avr); + avr->pc = reset_pc; + if (use_uart_pty) { // reset restores the pacing hack; re-clear it + uint32_t flags = 0; + avr_ioctl(avr, AVR_IOCTL_UART_GET_FLAGS('0'), &flags); + flags &= ~AVR_UART_FLAG_POLL_SLEEP; + avr_ioctl(avr, AVR_IOCTL_UART_SET_FLAGS('0'), &flags); + } + } + if (!use_uart_pty && ++since_poll >= 2000) { + since_poll = 0; + poll_pty(); + } + } + finish(0); + return 0; +}