avr/bootloader
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

pureboot: host tool and end-to-end protocol tests, all three chips

Browse Source 
pureboot.py (Python stdlib only): images as raw binary or Intel HEX,
flash and EEPROM programming with read-back verify, erase composites,
fuse and info readout, activation-timeout configuration, and the
tinies' reset-vector surgery — the trampoline word below the loader,
page 0 written last.

The test spawns a simavr device (pureboot_device.c) — the mega's USART
as a pty; on the tinies a cycle-timed GPIO<->pty bridge for the polled
software UART plus the NVM module simavr's tiny cores lack (their SPM
opcode ioctls into a void and silently does nothing) — and drives it
with the real tool: knock from reset (erased-flash walk on the tinies),
program and verify both memories, timeout write, session reconnect, an
external reset through the patched vector, hand-over, and the fixture
application's banner. Results are cross-checked against ground-truth
memory dumps and an independent decode of the surgery's rjmp words,
red-verified against a sabotaged encoder.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
BlackMark
2026-07-20 05:54:15 +02:00
parent 47990349a4
commit 2133627057
7 changed files with 1061 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
CMakeLists.txt
View File

@@ -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=$<TARGET_FILE:pureboot>
-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 $<TARGET_FILE:pbapp> $<TARGET_FILE:pbapp>.bin)
add_test(NAME pureboot.protocol
COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test/pbtest.py
${PB_DEVICE} $<TARGET_FILE:pureboot> ${LIBAVR_MCU} ${_pb_hz} ${_pb_base_hex}
${_pb_page} ${_pb_baud} ${_pb_eeprom} $<TARGET_FILE:pbapp>.bin
${CMAKE_CURRENT_SOURCE_DIR}/pureboot/pureboot.py
${CMAKE_BINARY_DIR}/pbtest-work)
set_tests_properties(pureboot.protocol PROPERTIES TIMEOUT 180)
endif()
endif()

27
pureboot/README.md
View File

@@ -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.

BIN
pureboot/__pycache__/pureboot.cpython-313.pyc Normal file
View File

Binary file not shown.

447
pureboot/pureboot.py Normal file
View File

@@ -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)

51
test/pbapp.cpp Normal file
View File

@@ -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 <libavr/libavr.hpp>
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 <avr::hertz_t C, bool Hardware = avr::hw::db.has_reg("UDR0")>
struct link {
using tx_t = avr::uart::usart0<C, {.baud = 115200_Bd, .max_baud_error = 2.5_pct}>;
static void tx(char c)
{
tx_t::write(static_cast<std::uint8_t>(c));
}
};
template <avr::hertz_t C>
struct link<C, false> {
using tx_t = avr::uart::software_tx<C, avr::pb1, 57600_Bd>;
static void tx(char c)
{
tx_t::write(static_cast<std::uint8_t>(c));
}
};
} // namespace
int main()
{
avr::init<typename link<dev::clock>::tx_t>();
link<dev::clock>::tx('A');
link<dev::clock>::tx('P');
link<dev::clock>::tx('P');
while (true) {
}
}

178
test/pbtest.py Normal file
View File

@@ -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 <device_bin> <pureboot_elf> <mcu> <hz> <base_hex> <page>
<baud> <eeprom_size> <app_bin> <tool_py> <workdir>
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()

309
test/pureboot_device.c Normal file
View File

@@ -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 <fcntl.h>
#include <pty.h>
#include <signal.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <termios.h>
#include <unistd.h>
#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 <pureboot.elf> <mcu> <hz> <base_hex> <page> <baud> <flash_dump>\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;
}