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

pureboot moves to its own repo

Browse Source 
pureboot is at git.blackmark.me/avr/pureboot now, with its own history: the
files it kept in pureboot/ sit at the top level there, its CMakeLists is the
merged whole of that unit and the build around it, and the v1..v8 tags moved
with it - each one checks out and compiles to exactly the .text it compiled to
here. The loader that repo builds is byte-identical to the one this commit
removes, verified before the removal rather than after.

Nothing is rewritten on this side. The history and the tags are untouched, so
every commit before this one still has pureboot in it and still builds it;
this is one commit that stops carrying it forward.

What goes with it: the four pureboot files, the thirteen pb*.py protocol
drivers and their four host-side unit tests, pbapp and the pureboot simavr
runner, check_pi.py, pbhw.py, pbrig.py, sizes.py, and the Studio project.
check_unit.cmake goes too - it had no caller left once the unit tests moved.

What is left is the four TinySafeBoot tiers, and the build shrinks to fit
them: 757 lines of CMakeLists to 137, and the preset matrix from 37 chips to
one, because the tiers reimplement an ATmega328P-only protocol and every other
chip in that list was there for pureboot. check.sh loses its size-table pass -
the table it checked was pureboot's README - and the Studio solution loses the
project that is now in the other repo.

Gate green: nine tests, four tiers at their section sizes and each one's
protocol suite against the simulator.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
BlackMark
2026-08-23 06:05:45 +02:00
parent bf9b86d2fc
commit 7c1a6a140b
35 changed files with 44 additions and 10815 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
test/check_pi.py
View File

@@ -1,79 +0,0 @@
#!/usr/bin/env python3
"""Position-independence lint: the property that lets the identical image run
from any slot, asserted from the built ELF and its object.
1. No absolute jmp/call - -mrelax normally guarantees it, but a branch that
grows out of relaxation range would break it silently.
2. Nothing flash-resident to address: the image is .text alone, so there is
no table whose runtime address has to be reconstructed.
3. The image is byte-identical when linked at a different base. This is
position independence itself rather than a proxy for it - an absolute
address anywhere in the image would move with the link and show up as a
differing byte.
Usage: check_pi.py <objdump> <objcopy> <cxx> <mcu> <elf> <object> <text_start_hex>
"""
import os
import re
import subprocess
import sys
import tempfile
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def main():
objdump, objcopy, cxx, mcu, elf, obj, text_start = sys.argv[1:]
text_start = int(text_start, 0)
listing = subprocess.run([objdump, "-d", elf], capture_output=True, text=True, check=True).stdout
absolute = [line for line in listing.splitlines() if re.search(r"\t(jmp|call)\t", line)]
if absolute:
fail("absolute control flow in the image:\n" + "\n".join(absolute))
# Allocated flash beyond .text would be data the running copy has to find.
# Only ALLOC sections reach the device at all; .comment and the debug
# sections ride along in the ELF container and are never flashed. objdump
# prints each section's flags on the line following its header.
headers = subprocess.run([objdump, "-h", elf], capture_output=True, text=True, check=True).stdout.splitlines()
for index, line in enumerate(headers):
fields = line.split()
if len(fields) < 6 or not fields[0].isdigit():
continue
name, size = fields[1], int(fields[2], 16)
flags = headers[index + 1] if index + 1 < len(headers) else ""
if "ALLOC" not in flags or not size:
continue
if name not in (".text", ".noinit", ".bss"):
fail(f"flash-resident section {name} ({size} bytes): the image must be .text alone")
# Relink at a different base and compare the bytes.
with tempfile.TemporaryDirectory() as work:
elsewhere = text_start - 0x200 if text_start >= 0x200 else text_start + 0x200
images = []
for base, tag in ((text_start, "here"), (elsewhere, "there")):
relinked = os.path.join(work, f"{tag}.elf")
binary = os.path.join(work, f"{tag}.bin")
subprocess.run(
[cxx, f"-mmcu={mcu}", "-nostartfiles", f"-Wl,--section-start=.text={base:#x}",
"-Wl,--defsym=pureboot_app=0", "-mrelax", obj, "-o", relinked],
check=True, capture_output=True)
subprocess.run([objcopy, "-O", "binary", relinked, binary], check=True)
images.append(open(binary, "rb").read())
if len(images[0]) != len(images[1]):
fail(f"the image is {len(images[0])} B linked at {text_start:#x} and "
f"{len(images[1])} B at {elsewhere:#x} - relaxation followed the address")
if images[0] != images[1]:
differing = [i for i, (a, b) in enumerate(zip(*images)) if a != b]
fail(f"the image changes when linked at {elsewhere:#x} instead of {text_start:#x}: "
f"{len(differing)} byte(s) differ, first at offset {differing[0]:#x}")
print(f"PI lint: control flow PC-relative, .text only, identical linked at {text_start:#x} and {elsewhere:#x}")
if __name__ == "__main__":
main()

36
test/check_unit.cmake
View File

@@ -1,36 +0,0 @@
# Asserts the autobaud loader's measured unit sits where the host will read
# it (--info's measured clock - the address is wire contract). Two homes: on
# a chip with the GPIOR pair the unit lives there and the image must carry no
# RAM word for it at all; elsewhere it is the first RAM object at SRAM start.
# Run as
# cmake -DOBJDUMP=... -DELF=... -DRAM_START=<data address> [-DGPIOR=<data address>]
# -P check_unit.cmake
execute_process(COMMAND ${OBJDUMP} -t ${ELF} OUTPUT_VARIABLE _syms RESULT_VARIABLE _res)
if(NOT _res EQUAL 0)
message(FATAL_ERROR "${OBJDUMP} -t ${ELF} failed")
endif()
# The symbol line: "00800100 l O .noinit 00000002 <mangled>6m_unitE".
string(REGEX MATCH "\n0*([0-9a-f]+)[^\n]+[ \t][^ \t\n]*6m_unitE\n" _line "${_syms}")
if(GPIOR)
if(_line)
message(FATAL_ERROR "m_unit RAM symbol present although the unit's home is GPIOR ${GPIOR} - "
"the host peeks the pair, and a RAM copy would be dead weight")
endif()
message(STATUS "no m_unit RAM object - the unit lives in the GPIOR pair at ${GPIOR}")
return()
endif()
if(NOT _line)
message(FATAL_ERROR "no m_unit symbol in ${ELF} - is this the autobaud loader?")
endif()
# AVR data-space symbols carry the 0x800000 VMA offset.
math(EXPR _want "0x800000 + ${RAM_START}" OUTPUT_FORMAT HEXADECIMAL)
math(EXPR _have "0x${CMAKE_MATCH_1}" OUTPUT_FORMAT HEXADECIMAL)
if(NOT _have STREQUAL _want)
message(FATAL_ERROR "m_unit sits at ${_have}, ram_start is ${_want} - the host peeks ram_start")
endif()
message(STATUS "m_unit at ${_have} == ram_start")

192
test/pbapp.cpp
View File

@@ -1,192 +0,0 @@
// Test-fixture application for the pureboot protocol tests: prints "APP" on
// the chip's serial link (the same link the loader uses) - 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.
//
// On the hardware-USART link it then listens, and an 'L' makes it jump into
// the resident loader - the application-owned loader entry a
// BOOTRST-unprogrammed mega relies on (reset always boots the application
// there), exercised by the self-update tests. The software link idles:
// reset reaches those loaders through the patched vector (or the runner
// models BOOTRST), so the application owes them nothing.
//
// PUREBOOT_HANDOVER drops the listening and jumps straight in, leaving the
// USART enabled behind it - the hand-over state a loader bit-banging on that
// USART's own pins has to survive.
//
// The fixture speaks the deployment its loader was built for: the same
// PUREBOOT_* defines configure it, and without them it assumes the stock
// deployment (the crystal/RC clock table below, the chip's natural link).
#include <libavr/libavr.hpp>
using namespace avr::literals;
namespace {
consteval avr::hertz_t clock()
{
#if defined(PUREBOOT_CLOCK_HZ)
return avr::hertz_t{PUREBOOT_CLOCK_HZ};
#else
auto name = std::string_view{avr::hw::db.name};
if (name.starts_with("ATtiny13")) {
return 9.6_MHz;
}
if (name.starts_with("ATtiny")) {
return 8_MHz;
}
return 16_MHz;
#endif
}
#if !defined(PUREBOOT_TX)
#define PUREBOOT_TX pb1
#endif
#if !defined(PUREBOOT_RX)
#define PUREBOOT_RX pb0
#endif
#if !defined(PUREBOOT_USART)
#define PUREBOOT_USART 0
#endif
consteval bool use_hardware()
{
#if defined(PUREBOOT_SOFT_SERIAL)
return false;
#else
return avr::uart::has_usart<0>();
#endif
}
using dev = avr::device<{.clock = clock()}>;
template <avr::hertz_t C, bool Hardware = use_hardware()>
struct link {
#if defined(PUREBOOT_BAUD)
static constexpr avr::baud_t baud{PUREBOOT_BAUD};
#else
static constexpr avr::baud_t baud{115200};
#endif
// The fixture speaks whatever rate the loader was built for, stock
// 115200 at 16 MHz included, which sits past the receiver-tolerance
// table's bound - the same deployment envelope the loader itself states.
using tx_t = avr::uart::usart<PUREBOOT_USART, C, {.baud = baud, .allow_baud_error = true}>;
static void init()
{
avr::init<tx_t>();
}
static void tx(char c)
{
tx_t::write(static_cast<std::uint8_t>(c));
}
// The loader sits in the top slot - 512 bytes on every chip. The jump
// takes a word address, which is what makes the >64 KiB chips' entry
// reachable through a 16-bit pointer at all.
static void enter_loader()
{
constexpr std::uint32_t slot = 512;
reinterpret_cast<void (*)()>(static_cast<std::uint16_t>((avr::hw::db.mem.flash_size - slot) / 2))();
}
[[noreturn]] static void idle()
{
#if defined(PUREBOOT_HANDOVER)
// Hand back at once, with this USART still enabled - the state that
// leaves a bit-banged loader on its pins mute unless the loader
// releases it. Unconditional because there is no command wire to
// wait on: that loader's link is the pins, not this peripheral.
enter_loader();
__builtin_unreachable();
#else
for (;;) {
auto command = tx_t::read_blocking();
if (command == 'L') {
enter_loader();
}
// 'D' leaves every word of the SPM page buffer dirty, so that a
// following 'L' enters the loader with the buffer it never clears.
// Hardware refuses application-section SPM on a boot-sectioned
// part; simavr dispatches it anyway, which is the whole reason the
// state is constructible - the stated section is the compilable
// fiction that matches what the simulator runs.
if (command == 'D') {
const auto open = avr::spm::page::begin<avr::spm::from::boot_section>(0);
for (std::uint16_t at = 0; at < avr::spm::page_bytes; at += 2) {
avr::spm::fill(open, at, 0xdead);
}
tx('D');
}
}
#endif
}
};
template <avr::hertz_t C>
struct link<C, false> {
#if defined(PUREBOOT_BAUD)
static constexpr avr::baud_t baud{PUREBOOT_BAUD};
#else
static constexpr avr::baud_t baud{57600};
#endif
// A shared-pin deployment (RX == TX) banners as a guest on its own line:
// the pull-up input is the released line, the transmitter takes the pin
// for exactly one frame per byte - the shape a real one-wire application
// beside this loader uses.
static constexpr bool one_wire = avr::PUREBOOT_RX == avr::PUREBOOT_TX;
using tx_t = avr::uart::software_tx<C, avr::PUREBOOT_TX, baud, one_wire>;
static void init()
{
// The guest transmitter configures no pin; the released line - the
// pull-up input a receiver would own - is established here.
if constexpr (one_wire) {
avr::init<avr::io::input<avr::PUREBOOT_TX, avr::io::pull::up>, tx_t>();
} else {
avr::init<tx_t>();
}
}
static void tx(char c)
{
tx_t::write(static_cast<std::uint8_t>(c));
}
[[noreturn]] static void idle()
{
#if defined(PUREBOOT_HEARTBEAT)
// Repeat the banner forever, which turns the fixture into a fixed
// cycles-per-bit transmitter: `tools/pbrig.py rate` sweeps the host rate
// against it to find the part's true bit rate, and from that the clock
// its RC oscillator is really running at. Only the *bit* timing carries
// the measurement - the delay merely spaces the lines out, so its own
// error does not matter. Software link only: the hardware-link idle owes
// the self-update tests a command loop, and a crystal deployment has
// nothing to measure.
while (true) {
tx('A');
tx('P');
tx('P');
tx('\r');
tx('\n');
dev::delay<50_ms>();
}
#else
while (true) {
}
#endif
}
};
} // namespace
int main()
{
link<dev::clock>::init();
#if !defined(PUREBOOT_HANDOVER)
link<dev::clock>::tx('A');
link<dev::clock>::tx('P');
link<dev::clock>::tx('P');
#endif
// The hand-over fixture stays silent: nothing is listening on the USART it
// brings up - the loader it hands to speaks those pins directly - so its
// banner would be a write into a peer that does not exist.
link<dev::clock>::idle();
}

210
test/pbautobaud.py
View File

@@ -1,210 +0,0 @@
#!/usr/bin/env python3
"""End-to-end autobaud test: drive an autobaud loader in simavr through the
calibration handshake and a flash + EEPROM + fuse round-trip, cross-checked
against the simulator's ground-truth memory - then repeat at a second F_CPU with
the *same* loader binary, which is the property autobaud exists for: one
clock-agnostic image that locks onto whatever rate the host sends.
Usage: pbautobaud.py <device_bin> <loader_elf> <mcu> <base_hex> <page>
<app_bin> <app_hz> <app_baud> <tool_py> <workdir> [link]
The loader is a software-serial build, driven over the GPIO<->pty bridge; the
optional link overrides the default -l sw:B0,B1 - RX == TX in it is the
one-wire deployment, and every session then runs with the host's echo
discard on. The app fixture is built for (app_hz, app_baud); the hand-over
is checked at that point, and a second point at half the clock proves the
lock is measured, not baked in.
"""
import os
import re
import sys
import time
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def main():
args = sys.argv[1:]
link = args.pop() if len(args) == 11 else "sw:B0,B1"
(device_bin, elf, mcu, base_hex, page, app_bin, app_hz, app_baud, tool, workdir) = args
base, page, app_hz, app_baud = int(base_hex, 0), int(page), int(app_hz), int(app_baud)
one_wire = re.fullmatch(r"sw:([A-H][0-7]),\1(@[01])?", link) is not None
extra = ("--one-wire",) if one_wire else ()
sys.path.insert(0, os.path.dirname(os.path.abspath(tool)))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pbsim
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)
# The geometry the surgery planner needs, from the chip class the runner is
# told - the same derivation pbtest.py makes: the boot-sectioned megas need
# no vector surgery, the tinies and the boot-section-less m48s do, and the
# large chips speak word addresses.
mega = mcu.startswith("atmega")
patch = not mega or mcu.startswith("atmega48")
word_flash = base + pb.SLOT > 0x10000
wire_base = base // 2 if word_flash else base
flags = (1 if patch else 0) | (2 if word_flash else 0)
ground_truth = pb.Info(bytes([ord("P"), ord("B"), pb.NEWEST_LOADER, 0, 0, 0, page & 0xFF,
wire_base & 0xFF, wire_base >> 8, 0, 0, flags]))
def round_trip(hz, baud, label, hand_over):
"""One clock point: reset, calibrate + knock, program, verify against the
simulator's own flash, and (at the app's point) hand over to the fixture."""
dump = os.path.join(workdir, f"flash_{label}.bin")
device = pbsim.Device(device_bin, elf, mcu, str(hz), base_hex, page, baud, dump, link=link)
try:
# The host tool, in autobaud mode, sends the 0xC0 calibration pulse
# and a single knock at `baud`; the loader locks to it.
out = pbsim.run_tool(tool, device.pty, baud, *extra, "--autobaud", "--info", "--clock", str(hz),
"--fuses", "--flash", app_bin, "--eeprom", ee_path, "--stay")
for needed in ("version", "signature", "fuses", "verify:", "stays"):
if needed not in out:
fail(f"{label}: session output lacks {needed!r}\n{out}")
# The measured clock, decoded from the unit at whichever home this
# version keeps it in. The runner's clock is exact, so the figure
# must land inside the
# encoding's own envelope: the loader floors the bit period to
# 4-cycle spin granules after an 8-cycle discount, and the edge
# poll can shave a few cycles more - one granule of slack below
# the true clock, none above (in cycles per bit, times the rate).
measured = re.search(r"measured\s+(\d+) Hz", out)
if not measured:
fail(f"{label}: --info lacks the measured clock\n{out}")
measured = int(measured.group(1))
if not hz - 19 * baud <= measured <= hz + 4 * baud:
fail(f"{label}: measured clock {measured} Hz is {measured - hz:+d} off the true {hz}")
# Read both memories back over the locked link and check them.
read_flash = os.path.join(workdir, f"rf_{label}.bin")
read_eeprom = os.path.join(workdir, f"re_{label}.bin")
out = pbsim.run_tool(tool, device.pty, baud, *extra, "--autobaud", "--verify-flash", app_bin,
"--verify-eeprom", ee_path, "--read-flash", read_flash,
"--read-eeprom", read_eeprom, "--stay")
if out.count("verify:") != 2:
fail(f"{label}: did not verify both memories\n{out}")
if open(read_eeprom, "rb").read()[: len(ee_image)] != ee_image:
fail(f"{label}: EEPROM read-back mismatch")
if hand_over:
# A calibration pulse with no knock behind it must not wedge
# the loader: the whole activation is bounded, including the
# knock's edge wait, so one stray low pulse - EMI, or a host
# that opens the port and never knocks - closes the window and
# boots the application. The banner is the proof.
# (The pause lets the loader reach its measurement loop, so the
# pulse is genuinely seen and the test cannot pass vacuously.)
device.reset()
port = pb.Port(device.pty, baud)
if one_wire:
port = pb.OneWirePort(port)
try:
time.sleep(0.2)
port.write(bytes((pb.CALIBRATE,)))
# Accumulate rather than match exactly: the reset leaves the
# idle line a framing artefact ahead of the banner, which is
# noise here - the question is only whether the app ran.
seen = b""
deadline = time.monotonic() + 180.0
while b"APP" not in seen and time.monotonic() < deadline:
seen += port.read_available(1.0)
if b"APP" not in seen:
fail(f"{label}: lone calibration pulse wedged the loader - app never bannered, saw {seen!r}")
print(f" {label}: lone calibration pulse does not wedge the loader")
finally:
port.close()
device.reset()
port = pb.Port(device.pty, baud)
if one_wire:
port = pb.OneWirePort(port)
try:
loader = pb.Loader(port)
live = loader.connect_autobaud(15)
if not pb.OLDEST_LOADER <= live.version <= pb.NEWEST_LOADER:
fail(f"{label}: loader reports pureboot {live.version}")
if loader.unified:
# pureboot 5's data space. 0x0200 is clear of the
# loader's own .noinit unit at the bottom of SRAM and of
# the stack at the top. Reading it back over the same
# locked link proves both directions of the new space.
probe = bytes(range(0x30, 0x40))
loader.write_data(0x0200, probe)
if loader.read_data(0x0200, len(probe)) != probe:
fail(f"{label}: RAM round-trip mismatch")
# The register file and the I/O space share the data
# address space on AVR, so the same command reaches a
# peripheral register. SPMCSR reads back as idle here.
verbose_ram = loader.read_data(0x0200, 4)
print(f" {label}: RAM read/write ok ({verbose_ram.hex()})")
loader.run_application()
banner = port.read_exact(3, 5.0)
if banner != b"APP":
fail(f"{label}: application banner was {banner!r}")
finally:
port.close()
finally:
device.stop()
# Ground truth (read after the runner exits and writes its dump): what
# the tool programmed must be what the simulator actually holds.
pages = pb.plan_flash(open(app_bin, "rb").read(), ground_truth)
flash_true = open(dump, "rb").read()
for address, data in pages.items():
if flash_true[address : address + page] != data:
fail(f"{label}: simulator flash differs from the programmed image at {address:#06x}")
print(f" {label}: locked at {hz} Hz / {baud} Bd, flash+EEPROM verified"
+ (", hand-over ok" if hand_over else ""))
def must_lock(hz, baud, label):
"""The calibration alone, at a tight bit period. Nothing is programmed -
the question is only whether the loader can still measure the pulse."""
dump = os.path.join(workdir, f"flash_{label}.bin")
device = pbsim.Device(device_bin, elf, mcu, str(hz), base_hex, page, baud, dump,
link=link)
try:
port = pb.Port(device.pty, baud)
if one_wire:
port = pb.OneWirePort(port)
try:
live = pb.Loader(port).connect_autobaud(15)
if live.version != pb.NEWEST_LOADER:
fail(f"{label}: loader reports pureboot {live.version}")
finally:
port.close()
finally:
device.stop()
print(f" {label}: locked at {hz} Hz / {baud} Bd ({hz / baud:.0f} cycles a bit)")
# The app fixture is built for one clock; the hand-over banners there. A
# second point at double that clock, same loader binary, proves the lock is
# measured, not baked in - the whole point of autobaud. (Doubling keeps the
# bit period healthy; halving would drop it below the software UART's floor.)
round_trip(app_hz, app_baud, "clock-a", hand_over=True)
round_trip(app_hz * 2, app_baud, "clock-b", hand_over=False)
# Both points above sit near 100 cycles a bit, which is comfortable. The
# calibration's real floor is far tighter, and it is worth a gate: measured
# here, the lock is solid down to ~36 cycles a bit and fails outright by ~31
# - a sharp edge, not a fraying one. This pins the tightest standard rate the
# fixture's clock reaches, so a change that raises the floor is caught.
#
# It does *not* bound what a real deployment can use. On silicon the
# oscillator's own jitter costs roughly a factor of two: an ATtiny13A on its
# factory RC trim was reliable at ~118 cycles a bit and already locking only
# 1 attempt in 5 by ~59, which no exact-clock simulation can show. The
# deployable envelope is a README matter; this is the logic's floor.
must_lock(app_hz, app_baud * 2, "tight-bit")
print("pbautobaud: calibration lock and flash/EEPROM/fuse round-trip pass at both clocks, "
"and the tight bit period still locks")
if __name__ == "__main__":
main()

88
test/pbdirty.py
View File

@@ -1,88 +0,0 @@
#!/usr/bin/env python3
"""Dirty-page-buffer acceptance test: with no discard in the loader, a page
filled over words an earlier writer left takes those instead. The whole
contract is asserted - a bare verify sees the corruption, the repairing
verify fixes it in one rewrite, and it stays fixed.
The state is reached the one way the loader cannot prevent: an application
dirties the buffer and jumps in with no reset between. Boot-sectioned megas
forbid that outright (SPM runs only from the boot section, Atmel-8271 section 26.2),
but simavr dispatches SPM from anywhere, which is what makes it constructible.
Usage: pbdirty.py <device_bin> <pureboot_elf> <mcu> <hz> <base_hex> <page>
<baud> <app_bin> <tool_py> <workdir>
"""
import os
import sys
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def main():
device_bin, elf, mcu, hz, base_hex, page, baud, app_bin, tool, workdir = sys.argv[1:]
page, baud = int(page), int(baud)
sys.path.insert(0, os.path.dirname(os.path.abspath(tool)))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pbsim
import pureboot as pb
os.makedirs(workdir, exist_ok=True)
dump = os.path.join(workdir, "dump.bin")
# Reset boots the application on a BOOTRST-unprogrammed mega; its 'L' is
# the loader entry this test needs, reached without a reset.
device = pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump, reset_hex="0")
try:
port = pb.Port(device.pty, baud)
loader = pb.Loader(port)
loader.connect(25)
# Install the application and hand over to it.
pb.op_flash(loader, app_bin, erase=False, verify=True)
loader.run_application()
if port.read_exact(3, 5.0) != b"APP":
fail("the application did not start")
port.write(b"D")
if port.read_exact(1, 5.0) != b"D":
fail("the application did not acknowledge dirtying the page buffer")
port.write(b"L")
loader = pb.Loader(port)
loader.connect(25)
# Program by hand, so the corruption is observable before anything
# repairs it.
pages = pb.plan_flash(open(app_bin, "rb").read(), loader.info)
for address in sorted(pages):
loader.write_page(address, pages[address])
try:
pb.verify_pages(loader, pages)
except pb.Error as error:
if "verify failed" not in str(error):
fail(f"the read-back failed, but not at verify: {error}")
else:
# Either the fixture no longer dirties the buffer, or the loader
# clears it again - in which case this test's premise is gone.
fail("programming over a dirty page buffer came back clean")
# What the programming path uses: one rewrite settles it, and it stays
# settled.
pb.verify_pages(loader, pages, repair=True)
pb.verify_pages(loader, pages)
# Ground truth beyond the loader's own read-back.
loader.run_application()
if port.read_exact(3, 5.0) != b"APP":
fail("the application did not start after the recovered write")
port.close()
finally:
device.stop()
print("pbdirty: a dirty page buffer is caught by verify and cleared by the retry")
if __name__ == "__main__":
main()

124
test/pbglitch.py
View File

@@ -1,124 +0,0 @@
#!/usr/bin/env python3
"""A link that damages bytes, deliberately and reproducibly, against the seal
that exists for it.
The board this was written for loses and mangles bytes on its own serial path,
and the failure that made it matter - a page-fill byte lost, the stream one
byte out, a page-address byte arriving where an SPMCSR value belongs - is not
reachable by asking a healthy link nicely. So the damage is injected here, at
a named byte index rather than a probability: a failing case is a case that
fails again.
Every check is a pair. The same bit flipped in the same field is applied on
one side of the seal and then the other: *after* the host seals the header,
which is a mangled command and must be refused, and *before*, which is a
well-formed command for something else and must be obeyed. Only the pair
proves anything - a test that showed the refusal alone would pass against a
loader that had simply stopped doing SPM, and one that showed the corruption
alone would not say what caught it.
Usage: pbglitch.py <device_bin> <pureboot_elf> <mcu> <hz> <base_hex> <page>
<baud> <tool_py> <workdir>
"""
import os
import sys
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def frame(pb, op, space, address, count, damage=None, before_seal=False):
"""A sealed command header, optionally with one byte damaged.
`damage` is (index, mask). Applied before the seal is computed it produces
a valid command for whatever the damaged fields now say; applied after, a
command whose seal no longer matches its own body - which is the shape a
link fault actually has."""
head = bytearray((op, pb.selector(space, address), address & 0xFF,
(address >> 8) & 0xFF, count & 0xFF))
if damage and before_seal:
head[damage[0]] ^= damage[1]
seal = pb.SEAL
for byte in head:
seal ^= byte
out = bytearray(head + bytes((seal,)))
if damage and not before_seal:
out[damage[0]] ^= damage[1]
return bytes(out)
def main():
device_bin, elf, mcu, hz, base_hex, page, baud, tool, workdir = sys.argv[1:]
base, page, baud = int(base_hex, 0), int(page), int(baud)
sys.path.insert(0, os.path.dirname(os.path.abspath(tool)))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pbsim
import pureboot as pb
os.makedirs(workdir, exist_ok=True)
dump = os.path.join(workdir, "dump.bin")
device = pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump)
try:
port = pb.Port(device.pty, baud)
loader = pb.Loader(port)
info = loader.connect(25)
if info.version < pb.SEALED_LOADER:
fail(f"this test is for pureboot {pb.SEALED_LOADER} and later, not {info.version}")
# A page of known bytes to watch. Everything below aims at it, so
# "nothing happened" is a readable claim rather than an absence.
marker = bytes((0x40 + (i & 0x3F)) for i in range(page))
loader.write_page(0, marker)
if loader.read_flash(0, page) != marker:
fail("the marker page did not survive an undamaged write")
# Every field of the header, one bit each. A damaged seal must be
# refused, the loader must re-prompt, and the page must be untouched -
# and it is the erase being aimed at it, so a single escape is visible.
for index in range(6):
bad = frame(pb, pb.OP_WRITE, pb.SP_SPM, 0, pb.SPM_ERASE, damage=(index, 0x01))
port.write(bad)
answer = port.read_exact(1, 5.0)
if answer != pb.NAK:
fail(f"a header damaged in byte {index} was not refused (got {answer.hex()})")
if port.read_exact(1, 5.0) != pb.PROMPT:
fail(f"no prompt after refusing a header damaged in byte {index}")
if loader.read_flash(0, page) != marker:
fail(f"damage in byte {index} reached flash - the marker page changed")
# The fill, whose payload is the protocol's one unacked burst: a
# refused fill must be refused *before* the page is sent, or the host
# is left pushing 128 bytes into a loader reading commands. Nothing is
# sent after the verdict here, and the very next command must be
# understood - that is the whole claim.
bad = frame(pb, pb.OP_FILL, pb.SP_FLASH, 0, page, damage=(3, 0x80))
port.write(bad)
if port.read_exact(1, 5.0) != pb.NAK:
fail("a damaged fill header was not refused")
if port.read_exact(1, 5.0) != pb.PROMPT:
fail("no prompt after refusing a damaged fill header")
if loader.identity().raw != info.raw:
fail("the loader was out of step after refusing a fill")
# The pair's other half. The identical flip, applied before the seal:
# a well-formed erase of the page one bit away from the one intended.
# It must be obeyed - otherwise the refusals above prove nothing about
# the seal and only that this loader stopped erasing.
port.write(frame(pb, pb.OP_WRITE, pb.SP_SPM, 0, pb.SPM_ERASE,
damage=(2, 0x01), before_seal=True))
if port.read_exact(1, 5.0) != pb.PROMPT:
fail("a correctly sealed erase was refused")
if port.read_exact(1, 5.0) != pb.PROMPT:
fail("no prompt after a correctly sealed erase")
if loader.read_flash(0, page) != b"\xff" * page:
fail("the sealed erase did not reach flash - the marker page is intact")
port.close()
finally:
device.stop()
print("pbglitch: damaged headers are refused, the identical damage sealed is obeyed")
main()

79
test/pbmute.py
View File

@@ -1,79 +0,0 @@
#!/usr/bin/env python3
"""Hand-over with a USART left enabled on the loader's own pins.
A software or autobaud link deployed on a USART's TxD is mute if an
application hands over with that USART still enabled: TXEN keeps the USART
owning the pin, so the bit-banged transmitter's port writes go nowhere and the
loader receives and obeys while answering nothing. The link's init releases it.
The state is reached the way silicon reaches it - an application that sets up
its USART and jumps in with no reset between, so nothing clears UCSRnB for it.
The pin ownership itself is modelled by the device runner: simavr wires a
USART through IRQs alone and never takes the pin from the port, so without
that the mute could not happen here at all (test/pureboot_device.cpp).
Usage: pbmute.py <device_bin> <pureboot_elf> <mcu> <hz> <base_hex> <page>
<baud> <app_bin> <tool_py> <workdir> <link>
"""
import os
import re
import sys
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def main():
device_bin, elf, mcu, hz, base_hex, page, baud, app_bin, tool, workdir, link = sys.argv[1:]
page, baud = int(page), int(baud)
sys.path.insert(0, os.path.dirname(os.path.abspath(tool)))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pbsim
import pureboot as pb
if "@" not in link:
fail(f"the link {link} names no owning USART - nothing would be under test")
# A shared line (RX == TX) echoes the host's own bytes; discard them the
# way the shipped --one-wire mode does.
one_wire = re.fullmatch(r"sw:([A-H][0-7]),\1@[01]", link) is not None
os.makedirs(workdir, exist_ok=True)
dump = os.path.join(workdir, "dump.bin")
device = pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump, link=link)
try:
port = pb.Port(device.pty, baud)
if one_wire:
port = pb.OneWirePort(port)
loader = pb.Loader(port)
loader.connect(25)
resident = loader.info.version
# Install the fixture and let it take over. It brings up the USART
# that owns these pins and jumps straight back in.
pb.op_flash(loader, app_bin, erase=False, verify=True)
loader.run_application()
# The loader is running again with that USART enabled behind it. Only
# the release makes it audible; without it the connect times out.
loader = pb.Loader(port)
try:
loader.connect(25)
except pb.Error as error:
fail(f"the loader never answered after the hand-over - the USART still owns its TX pin ({error})")
if loader.info.version != resident:
fail(f"identity changed across the hand-over: {resident} then {loader.info.version}")
# Answering is not enough: it has to still be a working loader.
pb.verify_pages(loader, pb.plan_flash(open(app_bin, "rb").read(), loader.info))
port.close()
finally:
device.stop()
print("pbmute: a loader on a USART's own pins answers after a hand-over that left it enabled")
if __name__ == "__main__":
main()

46
test/pbosccal.py
View File

@@ -1,46 +0,0 @@
#!/usr/bin/env python3
"""The build-time OSCCAL trim, observed through the wire: a loader built with
the OSCCAL axis holds the trim register at the built byte from its first
prompt on - the write sits at the top of run(), ahead of the WDRF bail, so
every path out of reset runs on the corrected clock. simavr's clock does not
follow OSCCAL, which is what makes the value assertable at all: the register
is plain state there, and the peek must return exactly what the build
declared rather than whatever the oscillator needed.
Usage: pbosccal.py <device_bin> <pureboot_elf> <mcu> <hz> <base_hex> <page>
<baud> <osccal_addr> <osccal_value> <tool_py> <workdir>
[link]
"""
import os
import sys
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def main():
args = sys.argv[1:]
link = args.pop() if len(args) == 12 else None
(device_bin, elf, mcu, hz, base_hex, page, baud, addr, value, tool, workdir) = args
addr, value, baud = int(addr, 0), int(value, 0), int(baud)
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pbsim
os.makedirs(workdir, exist_ok=True)
dump = os.path.join(workdir, "flash_dump.bin")
device = pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump, link=link)
try:
out = pbsim.run_tool(tool, device.pty, baud, "--peek", f"{addr:#x}:1")
want = f"{addr:#06x} {value:02x}"
if want not in out:
fail(f"OSCCAL at {addr:#x} did not read back {value:#04x}:\n{out}")
finally:
device.stop()
print("OK")
if __name__ == "__main__":
main()

86
test/pbrehome.py
View File

@@ -1,86 +0,0 @@
#!/usr/bin/env python3
"""Re-homing acceptance test: an image programmed somewhere other than its
canonical slot must still be a working loader, and the ordinary
--update-loader flow must put a build into the top slot from there.
Two positions. Address 0, a raw .bin handed to a programmer. And the staging
slot itself, where a loader already sitting there IS the staging copy -
recognized by its embedded block and left in place, then streaming the new
resident like any staged copy.
Usage: pbrehome.py <device_bin> <pureboot_elf> <update_bin> <mcu> <hz>
<base_hex> <page> <baud> <app_bin> <tool_py> <workdir>
"""
import os
import sys
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def rehome_from(pbsim, pb, device_bin, elf, place_hex, update_bin, base, page, baud, app_bin, workdir,
mcu, hz):
"""Place the loader at `place_hex`, heal through --update-loader, flash
the application, expect the banner."""
dump = os.path.join(workdir, f"dump-{place_hex}.bin")
state = os.path.join(workdir, f"rehome-{place_hex}.pbstate")
if os.path.exists(state):
os.unlink(state)
device = pbsim.Device(device_bin, elf, mcu, hz, place_hex, page, baud, dump, reset_hex="0")
try:
port = pb.Port(device.pty, baud)
loader = pb.Loader(port)
info = loader.connect(25)
if info.base != base:
fail(f"the misplaced copy reports base {info.base:#06x} - the info block must stay canonical")
# The ordinary update flow puts the build into the top slot.
pb.op_update_loader(loader, 25, update_bin, state, None)
update = open(update_bin, "rb").read()
if loader.read_flash(base, len(update)) != update:
fail("the canonical slot does not hold the update image")
# An application flashed through the healed resident overwrites the
# stale copy (surgery included) and launches.
pages = pb.plan_flash(open(app_bin, "rb").read(), loader.info)
for address in pb.covered(pages, loader.info, skip_blank=False):
loader.write_page(address, pages[address])
pb.verify_pages(loader, pages)
loader.run_application()
if port.read_exact(3, 5.0) != b"APP":
fail(f"application does not banner after the re-home from {place_hex}")
port.close()
finally:
device.stop()
def main():
(device_bin, elf, update_bin, mcu, hz, base_hex, page, baud, app_bin, tool, workdir) = sys.argv[1:]
base, page, baud = int(base_hex, 0), int(page), int(baud)
sys.path.insert(0, os.path.dirname(os.path.abspath(tool)))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pbsim
import pureboot as pb
os.makedirs(workdir, exist_ok=True)
# Address 0: the raw-.bin-to-a-programmer accident.
rehome_from(pbsim, pb, device_bin, elf, "0x0", update_bin, base, page, baud, app_bin, workdir, mcu, hz)
print("re-home from address 0: converged")
# The staging slot: erased flash with the loader sitting exactly where
# a staging copy would - the tool must leave it in place and let it
# stream the (different) update build into the resident slot.
stage = base - pb.SLOT
rehome_from(pbsim, pb, device_bin, elf, hex(stage), update_bin, base, page, baud, app_bin, workdir,
mcu, hz)
print("re-home from the staging slot: converged")
print("pbrehome: a misplaced loader re-homes through the ordinary update flow")
if __name__ == "__main__":
main()

91
test/pbreloc.py
View File

@@ -1,91 +0,0 @@
#!/usr/bin/env python3
"""Position-independence acceptance test: the identical binary, flashed one
slot below the resident, must serve the complete command set from there. The
info block must come back byte-identical, and the staged copy must be able to
rewrite the resident verbatim - which is the whole of what relocation is for.
Usage: pbreloc.py <device_bin> <pureboot_elf> <mcu> <hz> <base_hex> <page>
<baud> <tool_py> <workdir>
"""
import os
import subprocess
import sys
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def main():
device_bin, elf, mcu, hz, base_hex, page, baud, tool, workdir = sys.argv[1:]
base, page, baud = int(base_hex, 0), int(page), int(baud)
stage = None # derived from the device's own info (slot-sized) below
sys.path.insert(0, os.path.dirname(os.path.abspath(tool)))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pbsim
import pureboot as pb
os.makedirs(workdir, exist_ok=True)
objcopy = os.environ.get("PB_OBJCOPY", "avr-objcopy")
image_path = os.path.join(workdir, "pureboot.bin")
subprocess.run([objcopy, "-O", "binary", elf, image_path], check=True)
image = open(image_path, "rb").read()
device = pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, os.path.join(workdir, "dump.bin"))
try:
port = pb.Port(device.pty, baud)
loader = pb.Loader(port)
info = loader.connect(25)
if info.base != base:
fail(f"info reports base {info.base:#06x}")
resident_info = info.raw
# Install the staging copy exactly as the update flow would.
stage = info.stage
staged = pb.staging_content(image, info)
pb.write_differing(loader, stage, staged)
# Enter it; from here on, every command runs in the relocated copy.
staged_info = loader.enter_copy(stage, 25)
if staged_info.raw != resident_info:
fail(f"staged info {staged_info.raw.hex()} != resident info {resident_info.hex()}")
# 'R' from the staged copy already proved itself in the install
# verify; 'F' must answer 4 bytes (values are unmodeled in simavr).
if len(loader.read_fuses()) != 4:
fail("fuse read from the staged copy")
# EEPROM round-trip through the staged copy.
pattern = bytes(range(0x50, 0x60))
loader.write_eeprom(0, pattern)
if loader.read_eeprom(0, len(pattern)) != pattern:
fail("EEPROM round-trip through the staged copy")
# The resident slot, written from the copy standing beside it - the
# whole point of relocating. There is no running-slot guard to probe
# against: nothing here refuses an address, and a copy that erases the
# page it is executing from does not come back to report it.
# pbselfwrite.py gates that direction on a device it is allowed to
# destroy.
marker = bytes((i * 3) & 0xFF for i in range(page))
loader.write_page(base, marker)
if loader.read_flash(base, page) != marker:
fail("the staged copy could not write the resident slot")
# Restore the resident image through the staged copy, then 'J' back
# into it and prove it lives.
resident = image + b"\xff" * (pb.SLOT - len(image))
pb.write_differing(loader, base, resident)
back_info = loader.enter_copy(base, 25)
if back_info.raw != resident_info:
fail("the restored resident does not serve its info block")
port.close()
finally:
device.stop()
print("pbreloc: the relocated copy serves the full command set")
if __name__ == "__main__":
main()

132
test/pbselfwrite.py
View File

@@ -1,132 +0,0 @@
#!/usr/bin/env python3
"""The seal, gated on the one command that proves it: erase the page the
loader is executing from.
pureboot 9 dropped the running-slot write guard, so this command is now
permitted - that is what lets a resident copy plant something in its own slot,
which on a chip whose boot section *is* the loader slot is the only route a
self-update has. Permitted means the loader must actually do it, and the only
honest proof is the flash afterwards.
What stands in the guard's place is the seal, and the two halves are tested
against each other here: the identical destructive command, refused when its
seal is wrong and honoured when it is right. A test that only showed the
refusal would pass just as well against a loader that ignores SPM entirely.
Usage: pbselfwrite.py <device_bin> <pureboot_elf> <mcu> <hz> <base_hex> <page>
<baud> <tool_py> <workdir>
"""
import os
import sys
import time
# How long the device is left running after the sealed command, for the frame
# to arrive at the wire's rate and the erase to reach flash. It is spent in
# full on every attempt, because the only cheaper signal - the loader answering
# - is the one that precedes the SPM.
settle_seconds = 2.0
settle_attempts = 3
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def sealed_frame(pb, op, space, address, count):
"""A command header and its seal, built here rather than borrowed from the
tool: this test is about what the loader accepts, and a probe that shares
the host's frame builder cannot tell a wrong frame from a wrong loader."""
head = bytes((op, pb.selector(space, address), address & 0xFF,
(address >> 8) & 0xFF, count & 0xFF))
seal = pb.SEAL
for byte in head:
seal ^= byte
return head + bytes((seal,))
def scenario(pb, pbsim, device_bin, elf, mcu, hz, base_hex, page, baud, dump, base, settle):
"""The whole scenario once, answering with the running page as the
simulator's own flash holds it afterwards."""
device = pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump)
try:
port = pb.Port(device.pty, baud)
loader = pb.Loader(port)
info = loader.connect(25)
if info.version < pb.SEALED_LOADER:
fail(f"this test is for pureboot {pb.SEALED_LOADER} and later, not {info.version}")
# Erase the first page of the running slot: the entry stub and the
# command loop are both in it, so a loader that performs this does not
# answer again. Nothing else in the protocol is as sharp a probe.
frame = sealed_frame(pb, pb.OP_WRITE, pb.SP_SPM, base, pb.SPM_ERASE)
# Red: the same command with one bit wrong in its seal. Refused before
# anything happens, and the loader is still there to say so.
broken = bytearray(frame)
broken[-1] ^= 0x01
port.write(bytes(broken))
if port.read_exact(1, 5.0) != pb.NAK:
fail("an unsealed erase of the running page was not refused")
if port.read_exact(1, 5.0) != pb.PROMPT:
fail("the loader did not re-prompt after refusing the erase")
alive = loader.read_flash(base, 8)
if alive == b"\xff" * 8:
fail("the refused erase happened anyway - the running page reads erased")
# Green: the identical command, correctly sealed. The claim is that the
# erase reached flash, and the dump is the only witness for it - it
# tells "accepted and performed" from "merely answered".
#
# Nothing the link says may shorten the settle, because the verdict is
# issued *before* the SPM: a prompt reply means the erase has not
# happened yet, and stopping the device on it races the write. The link
# dying here is an expected outcome rather than a failure - the loader
# is erasing its own command loop - so a closing pty ends the settle
# instead of escaping it.
port.write(frame)
deadline = time.monotonic() + settle
while time.monotonic() < deadline:
try:
port.read_available(0.1)
except (pb.Error, OSError):
break
try:
port.close()
except OSError:
pass
finally:
device.stop()
# Ground truth: the simulator's flash, not the loader's opinion of it.
return open(dump, "rb").read()[base : base + page]
def main():
device_bin, elf, mcu, hz, base_hex, page, baud, tool, workdir = sys.argv[1:]
base, page, baud = int(base_hex, 0), int(page), int(baud)
sys.path.insert(0, os.path.dirname(os.path.abspath(tool)))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pbsim
import pureboot as pb
os.makedirs(workdir, exist_ok=True)
dump = os.path.join(workdir, "dump.bin")
# A settle is wall clock and the erase is the simulator's progress through
# it, so a loaded machine needs more of the first for the same amount of
# the second - which is what made a single fixed wait flaky under the gate's
# own parallelism. Doubling until the claim holds keeps the claim exact: a
# loader that does not erase fails every attempt, and only the budget moves.
settle = settle_seconds
for _ in range(settle_attempts):
if scenario(pb, pbsim, device_bin, elf, mcu, hz, base_hex, page, baud, dump, base, settle) == b"\xff" * page:
print("pbselfwrite: the running slot is refused unsealed and erased sealed")
return
settle *= 2
fail("the sealed erase did not reach flash - the running page is intact")
main()

75
test/pbsim.py
View File

@@ -1,75 +0,0 @@
"""Shared simavr harness for the pureboot tests: spawn the device runner,
hand out its pty, restart it from a flash dump (the power-fail path), and
keep its chatter out of undrained pipes."""
import os
import signal
import subprocess
class Device:
def __init__(self, binary, elf, mcu, hz, base_hex, page, baud, dump, reset_hex=None, resume=None, link=None,
window=False, fuses=None):
cmd = [binary]
if link:
cmd += ["-l", link]
if window:
cmd.append("-w") # report the first-transmit cycle, free-run idle
if fuses:
# What a fuse read answers, low,lock,extended,high. Unprogrammed
# otherwise, which is the profile every other test here runs.
cmd += ["-f", fuses]
cmd += [elf, mcu, hz, base_hex, str(page), str(baud), dump]
if reset_hex is not None or resume is not None:
# Chips without a hardware boot section - the tinies and the
# m48s - reset to address 0 like silicon; the boot-sectioned
# megas re-vector to the loader base (BOOTRST).
patch = not mcu.startswith("atmega") or mcu.startswith("atmega48")
cmd.append(reset_hex if reset_hex is not None else ("0" if patch else base_hex))
if resume is not None:
cmd.append(resume)
self.log = open(dump + ".log", "a")
self.proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=self.log, text=True)
self.dump = dump
self.pty = None
for _ in range(50):
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 reset(self):
"""The external reset line: SIGUSR1 re-enters at the reset vector."""
self.proc.send_signal(signal.SIGUSR1)
def power_fail(self):
"""SIGTERM: the runner dumps its flash and exits - the image a
restart resumes from."""
self.stop()
return self.dump
def stop(self):
self.proc.terminate()
try:
self.proc.wait(timeout=5)
except subprocess.TimeoutExpired:
self.proc.kill()
self.log.close()
def run_tool(tool, pty, baud, *args, timeout=180):
result = subprocess.run(
[os.environ.get("PYTHON", "python3"), tool, "--port", pty, "--baud", str(baud), "--wait", "25", *args],
capture_output=True,
text=True,
timeout=timeout,
)
print(result.stdout, end="")
if result.returncode != 0:
raise RuntimeError(f"tool exited {result.returncode}: {result.stderr.strip()}")
return result.stdout

210
test/pbtest.py
View File

@@ -1,210 +0,0 @@
#!/usr/bin/env python3
"""End-to-end protocol test: drive the simavr device with the real host tool
over its pty through flash, EEPROM, fuse and hand-over scenarios, and
cross-check the tool's view against the simulator's ground-truth dumps.
Usage: pbtest.py <device_bin> <pureboot_elf> <mcu> <hz> <base_hex> <page>
<baud> <eeprom_size> <app_bin> <tool_py> <workdir> [link]
The optional link is the runner's -l spec (usart1, sw:B5,B1, ...), for a
loader built off the chip's natural serial default.
"""
import os
import re
import sys
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
def main():
args = sys.argv[1:]
link = args.pop() if len(args) == 12 else None
(device_bin, elf, mcu, hz, base_hex, page, baud, eeprom_size, app_bin, tool, workdir) = args
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)))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pbsim
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:
# the boot-sectioned megas need no vector surgery (the tinies and the
# boot-section-less m48s do), the large chips speak word addresses, and
# the page byte is the wire's 0-means-256.
mega = mcu.startswith("atmega")
patch = not mega or mcu.startswith("atmega48")
# Where SRAM begins: the x8 and x4 megas push it past their extended I/O
# space, everything else starts right after the plain I/O registers. The
# loader keeps no statics and its stack sits at RAMEND, so the first SRAM
# byte is free for the data-space probe below.
classic = mcu in ("atmega8", "atmega8a", "atmega16", "atmega16a", "atmega32", "atmega32a")
ram_base = 0x0100 if mega and not classic else 0x0060
word_flash = base + pb.SLOT > 0x10000
wire_base = base // 2 if word_flash else base
flags = (1 if patch else 0) | (2 if word_flash else 0)
info = pb.Info(
bytes([ord("P"), ord("B"), pb.NEWEST_LOADER, 0, 0, 0, page & 0xFF])
+ bytes([wire_base & 0xFF, wire_base >> 8, eeprom_size & 0xFF, eeprom_size >> 8])
+ bytes([flags])
)
# A shared-line link (RX == TX in the -l spec) makes the host read every
# byte it sends back off the line; all sessions then discard the echo.
one_wire = bool(link) and re.fullmatch(r"sw:([A-H][0-7]),\1(@[01])?", link) is not None
extra = ("--one-wire",) if one_wire else ()
device = pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump, link=link)
try:
# Session 1: knock from reset, identify, program everything, stay.
out = pbsim.run_tool(tool, device.pty, baud, *extra, "--info", "--fuses", "--flash", app_bin,
"--eeprom", ee_path, "--stay")
for needed in ("version", "signature", "fuses", "verify:", "stays"):
if needed not in out:
fail(f"session 1 output lacks {needed!r}")
# Session 2: reconnect into the live session, verify, dump, exercise
# the data space; hand over is deferred - the pty must be reopened for
# the APP banner first.
probe = "c0ffee"
out = pbsim.run_tool(tool, device.pty, baud, *extra, "--verify-flash", app_bin, "--verify-eeprom", ee_path,
"--read-flash", read_flash, "--read-eeprom", read_eeprom,
"--poke", f"{ram_base:#x}:{probe}", "--peek", f"{ram_base:#x}:3", "--stay")
if out.count("verify:") != 2:
fail("session 2 did not verify both memories")
# What went into SRAM must come back out of it: the data space is one
# more selector on the same transfer as flash and EEPROM, so a wrong
# selector decode would show up here and nowhere else.
if probe not in out.replace(" ", ""):
fail(f"data-space round trip at {ram_base:#x} did not read back {probe}\n{out}")
eeprom_back = open(read_eeprom, "rb").read()
if eeprom_back[: len(ee_image)] != ee_image:
fail("EEPROM read-back mismatch")
# 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 the 'J' hand-over must
# land in the application, which banners on the same link.
device.reset()
port = pb.Port(device.pty, baud)
if one_wire:
port = pb.OneWirePort(port)
try:
loader = pb.Loader(port)
live = loader.connect(15)
# The loader built from this tree must report a version the tool
# beside it speaks - a bump the tool was never told about is a
# loader it would refuse to talk to. Not equality with the newest:
# the tool now spans two loader generations, the fixed-baud one
# here and the unified autobaud loader that follows it.
if not pb.OLDEST_LOADER <= live.version <= pb.NEWEST_LOADER:
fail(f"loader reports pureboot {live.version}, the tool speaks "
f"{pb.OLDEST_LOADER}..{pb.NEWEST_LOADER}")
# A fill addressed inside a page rather than at its base must still
# consume exactly one page and prompt. The loader's own slot is the
# target and the payload is erased-state bytes, so the probe can
# disturb neither the image nor the page buffer it leaves behind: a
# fill only loads the buffer, and nothing commits it. Hand-built
# rather than through write_page(), which would follow the fill
# with its erase and write; the point here is that the fill alone
# consumes exactly one page whatever the address's low bits say.
# Hand-sealed too - a protocol probe that borrowed the tool's own
# frame builder could not tell a wrong frame from a wrong loader.
wire = base + 1
head = bytes((pb.OP_FILL, pb.selector(pb.SP_FLASH, wire), wire & 0xFF,
(wire >> 8) & 0xFF, page & 0xFF))
seal = pb.SEAL
for byte in head:
seal ^= byte
port.write(head + bytes((seal,)))
if port.read_exact(1, 5.0) != pb.PROMPT:
fail("the loader refused a correctly sealed fill")
port.write(b"\xff" * page)
if port.read_exact(1, 5.0) != pb.PROMPT:
fail("unaligned fill did not return to the prompt")
# And the seal itself, red: one wrong bit in the address of that
# same frame must be refused outright. The verdict has to arrive
# *before* the page would have been sent - that ordering is what
# keeps a refusal from turning into a desync - so the probe sends
# no payload at all and expects the loader straight back at the
# command level.
broken = bytearray(head + bytes((seal,)))
broken[2] ^= 0x01
port.write(bytes(broken))
if port.read_exact(1, 5.0) != pb.NAK:
fail("a header with a broken seal was not refused")
if port.read_exact(1, 5.0) != pb.PROMPT:
fail("the loader did not re-prompt after refusing a broken seal")
loader.run_application()
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 (patched-vector
# chips only - a boot-sectioned mega's word 0 stays the application's).
if patch:
flash_words = (base + pb.SLOT) // 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:
fail("ground-truth EEPROM does not match what was programmed")
print("pbtest: all scenarios pass")
if __name__ == "__main__":
main()

222
test/pbupdate.py
View File

@@ -1,222 +0,0 @@
#!/usr/bin/env python3
"""Self-update end-to-end: an application is flashed, the loader replaces
itself with a re-timed build, and every power-fail phase is rehearsed by
killing the device mid-write, restarting it from its flash dump, and letting
a re-run complete the update.
The boot-sectioned megas run the BOOTRST-unprogrammed profile - reset boots
the application, whose 'L' is the application-owned loader entry - with
--assume-fuses standing in for the fuse read simavr cannot model.
Usage: pbupdate.py <device_bin> <pureboot_elf> <update_elf> <mcu> <hz>
<base_hex> <page> <baud> <app_bin> <tool_py> <workdir>
"""
import os
import subprocess
import sys
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def rjmp_decode(word, at, flash_words):
"""Written against the instruction-set definition, not with the 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 PowerFail(Exception):
pass
def assumed_fuses(pb, image):
"""Synthetic 'F' bytes for --assume-fuses: the smallest boot section
covering both the resident and the staging slot (two slots - what a
self-update needs), BOOTRST unprogrammed - the per-chip BOOTSZ ladder
and fuse byte come from the tool's own table, keyed by the update
image's embedded signature."""
info = pb.image_info(image)
which, ladder = pb.BOOT_FUSE[bytes(info.signature[1:3])]
bits = min((b for b in ladder if ladder[b] * 2 >= 2 * pb.SLOT), key=lambda b: ladder[b])
fuses = bytearray((0xFF, 0xFF, 0xFF, 0xFF))
fuses[which] = 0xF8 | (bits << 1) | 1
return bytes(fuses)
def make_fault_loader(pb, base, slot, kill_region, kill_hits, device):
"""A Loader whose write_page kills the device (or, with device=None,
just the host) at the Nth write into a region; the sequence
stage->resident->stage distinguishes the install from the restore."""
class FaultLoader(pb.Loader):
def __init__(self, port):
super().__init__(port)
self.seen_resident = False
self.hits = 0
def write_page(self, address, data):
if address >= base:
phase = "resident"
self.seen_resident = True
elif address >= base - slot:
phase = "stage_restore" if self.seen_resident else "stage"
else:
phase = "app"
if phase == kill_region:
self.hits += 1
if self.hits == kill_hits:
if device is not None:
device.power_fail()
raise PowerFail(f"{kill_region} write {kill_hits}")
super().write_page(address, data)
return FaultLoader
def main():
(device_bin, elf, update_elf, mcu, hz, base_hex, page, baud, app_bin, tool, workdir) = sys.argv[1:]
base, page, baud = int(base_hex, 0), int(page), int(baud)
mega = mcu.startswith("atmega")
# The m48s are megas without a boot section: patched vector, no fuse
# preflight, and the same reset-to-0 the tinies get.
patch = not mega or mcu.startswith("atmega48")
reset_hex = "0" if mega else None # the boot-sectioned mega runs BOOTRST-unprogrammed here
sys.path.insert(0, os.path.dirname(os.path.abspath(tool)))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pbsim
import pureboot as pb
slot = pb.SLOT
os.makedirs(workdir, exist_ok=True)
objcopy = os.environ.get("PB_OBJCOPY", "avr-objcopy")
images = {}
for name, source in (("v0", elf), ("v9", update_elf)):
path = os.path.join(workdir, name + ".bin")
subprocess.run([objcopy, "-O", "binary", source, path], check=True)
images[name] = open(path, "rb").read()
if images["v0"] == images["v9"]:
fail("the update image is byte-identical to the resident build")
dump = os.path.join(workdir, "dump.bin")
state = os.path.join(workdir, "update.pbstate")
fuses = assumed_fuses(pb, images["v0"]) if mega and not patch else None
def connect(device):
port = pb.Port(device.pty, baud)
if mega:
# Reset boots the application here; its 'L' is the loader entry.
# To a live loader the same byte is an ignored command.
port.read_available(0.5)
port.write(b"L")
loader = pb.Loader(port)
loader.connect(25)
return port, loader
def padded(image):
return image + b"\xff" * (slot - len(image))
def resident_bytes(loader):
return loader.read_flash(base, slot)
def assert_state(loader, image, app_pages):
if resident_bytes(loader) != padded(image):
fail("resident loader does not match the update image")
stage = base - slot
got = loader.read_flash(stage, slot)
for address, data in app_pages.items():
if stage <= address < base:
if got[address - stage : address - stage + page] != data:
fail(f"staging region page {address:#06x} not restored")
device = pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump, reset_hex=reset_hex)
final = "v0"
try:
# The application first - its planner output is the restore truth.
pbsim.run_tool(tool, device.pty, baud, "--flash", app_bin, "--stay")
port, loader = connect(device)
app_pages = pb.plan_flash(open(app_bin, "rb").read(), loader.info)
port.close()
# A clean CLI update, resident -> v9.
args = ["--update-loader", os.path.join(workdir, "v9.bin"), "--state", state, "--stay"]
if fuses:
args += ["--assume-fuses", fuses.hex()]
out = pbsim.run_tool(tool, device.pty, baud, *args)
if "loader updated" not in out:
fail("update did not report success")
if os.path.exists(state):
fail("state file survived a completed update")
port, loader = connect(device)
assert_state(loader, images["v9"], app_pages)
loader.run_application()
if port.read_exact(3, 5.0) != b"APP":
fail("application does not banner after the update")
port.close()
final = "v9"
print("clean update: resident replaced, staging restored, application intact")
# Power-fail rehearsal: kill mid-phase, restart from the dump,
# re-run, and the update must still complete. Each round flips the
# direction so the flash is never already at its target. The mega's
# mid-resident-rewrite loss is exercised as a host crash instead:
# with BOOTRST unprogrammed and the resident mid-erase, a power loss
# there has no reset path into the staging copy - the documented
# cost of that profile (README).
for kill_region, kill_hits, kill_device in (
("stage", 2, True),
("resident", 1, patch),
("stage_restore", 2, True),
):
device.reset() # the previous round left the application running
port, loader = connect(device)
target = "v9" if resident_bytes(loader) == padded(images["v0"]) else "v0"
image_path = os.path.join(workdir, target + ".bin")
injected = make_fault_loader(pb, base, slot, kill_region, kill_hits, device if kill_device else None)(port)
injected.info = loader.info
try:
pb.op_update_loader(injected, 25, image_path, state, fuses)
fail(f"{kill_region}: fault never triggered")
except PowerFail as event:
print(f"power fail injected: {event}")
port.close()
if kill_device:
device = pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump,
reset_hex=reset_hex, resume=dump)
port, loader = connect(device)
pb.op_update_loader(loader, 25, image_path, state, fuses)
assert_state(loader, images[target], app_pages)
loader.run_application()
if port.read_exact(3, 5.0) != b"APP":
fail(f"{kill_region}: application lost after the resumed update")
port.close()
final = target
print(f"resumed after {kill_region} loss: update completed, application intact")
finally:
device.stop()
# Ground truth: the simulator's own flash against the final state, and
# on the patched-vector chips an independent decode of the reset routing.
flash = open(dump, "rb").read()
if flash[base : base + slot] != padded(images[final]):
fail("ground-truth resident region does not match the final image")
if patch:
flash_words = (base + slot) // 2
word0 = flash[0] | (flash[1] << 8)
if rjmp_decode(word0, 0, flash_words) != base // 2:
fail("ground-truth reset vector does not land on the loader")
app = open(app_bin, "rb").read()
trampoline = flash[base - 2] | (flash[base - 1] << 8)
if rjmp_decode(trampoline, (base - 2) // 2, flash_words) != rjmp_decode(app[0] | (app[1] << 8), 0, flash_words):
fail("ground-truth trampoline does not land on the application entry")
print("pbupdate: clean update + all power-fail phases recovered")
if __name__ == "__main__":
main()

120
test/pbwalk.py
View File

@@ -1,120 +0,0 @@
#!/usr/bin/env python3
"""The reset walk region, end to end - the brick this refusal exists for.
BOOTRST programmed on a boot-sectioned mega enters reset at the boot section's
base, and the loader is reached only by walking up across erased flash. An
application that grew into that span is executed by the reset instead: it
happened to an ssd1306 board on a 328P with hfuse 0xdc, and with no reset edge
brought out, an ICE was the only way back.
The plain --flash is the path under test. It asks for no fuses of its own,
which is exactly why it once skipped the check and wrote the image anyway.
The runner still resets into the loader - the walk itself is not modeled. What
is modeled is the fuse read, so the refusal is decided from the bytes silicon
would have answered (-f, test/pureboot_device.cpp).
Usage: pbwalk.py <device_bin> <pureboot_elf> <mcu> <hz> <base_hex> <page>
<baud> <app_bin> <tool_py> <workdir>
"""
import os
import sys
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def intel_hex(chunks):
"""Intel HEX for {address: bytes}. An application that reaches the walk
region is sparse there - a .bin would have to carry every byte between."""
lines = []
for address, data in sorted(chunks.items()):
for at in range(0, len(data), 16):
row = data[at : at + 16]
here = address + at
record = bytes((len(row), here >> 8, here & 0xFF, 0)) + row
lines.append(":" + (record + bytes((-sum(record) & 0xFF,))).hex())
return "\n".join(lines + [":00000001FF"]) + "\n"
def main():
device_bin, elf, mcu, hz, base_hex, page, baud, app_bin, tool, workdir = sys.argv[1:]
page, baud = int(page), int(baud)
sys.path.insert(0, os.path.dirname(os.path.abspath(tool)))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pbsim
import pureboot as pb
os.makedirs(workdir, exist_ok=True)
dump = os.path.join(workdir, "dump.bin")
# The ssd1306 board's own fuses, as an Atmel-ICE reads them off the part,
# and the same boot section with reset left on the application - the one
# difference that decides the refusal.
bricking, harmless = "fffffddc", "fffffddd"
def session(fuses):
return pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump, fuses=fuses)
def connected(device):
port = pb.Port(device.pty, baud)
loader = pb.Loader(port)
loader.connect(25)
return port, loader
device = session(bricking)
try:
port, loader = connected(device)
info = loader.info
bootrst, bls_start = pb.mega_boot(info, bytes.fromhex(bricking))
if not bootrst or bls_start >= info.base:
fail(f"this profile cannot brick: BOOTRST {bootrst}, section at {bls_start:#06x}")
port.close()
# An application whose tail lands in the walk region, which is what the
# bricked board's dump showed: live bytes at the section base.
overgrown = os.path.join(workdir, "overgrown.hex")
with open(overgrown, "w") as image:
image.write(intel_hex({0: open(app_bin, "rb").read(), bls_start: bytes(page)}))
out = pbsim.run_tool(tool, device.pty, baud, "--fuses", "--stay")
if f"0x{bricking[6:]}" not in out:
fail(f"the device did not answer the fuses it was given ({bricking}):\n{out}")
try:
pbsim.run_tool(tool, device.pty, baud, "--flash", overgrown, "--stay")
except RuntimeError as refused:
if "walk region" not in str(refused):
fail(f"--flash failed, but not on the walk region: {refused}")
else:
fail("a plain --flash wrote an application into the reset walk region")
port, loader = connected(device)
if loader.read_flash(bls_start, page) != bytes((0xFF,)) * page:
fail(f"the refused image was written to {bls_start:#06x} anyway")
port.close()
# The override is the escape hatch, not the default.
pbsim.run_tool(tool, device.pty, baud, "--flash", overgrown, "--erase-flash", "--force", "--stay")
port, loader = connected(device)
if loader.read_flash(bls_start, page) != bytes(page):
fail(f"--force did not write {bls_start:#06x}")
port.close()
finally:
device.stop()
# An image that stays below the region flashes with no override at all,
# and so does the same overgrown image once reset boots the application.
for fuses, image in ((bricking, app_bin), (harmless, overgrown)):
device = session(fuses)
try:
pbsim.run_tool(tool, device.pty, baud, "--flash", image, "--erase-flash", "--stay")
finally:
device.stop()
print("pbwalk: the walk region is refused by default, overridden by --force, and left alone otherwise")
main()

137
test/pbwindow.py
View File

@@ -1,137 +0,0 @@
#!/usr/bin/env python3
"""The activation window as a behavioral duration gate.
The loader's window is a counted poll loop whose per-poll cost is hand-counted
in the source (`link::poll_cycles`) - but the loop compiles in consumer
context, so only the running image can prove the count. This test installs a
real application beside the loader (the host tool's own `plan_flash` supplies
the reset-vector surgery), starts the simulator with the line idle, and reads
the cycle of the first transmit activity: nothing talks until the window
closes and the application banners, so that cycle *is* the window, give or
take a banner lead measured in microseconds. Asserted at +/-2 % - one
mis-counted cycle per poll shifts a window by 10 % and more.
Fixed-baud loaders declare their window in seconds (--seconds, the build's
TIMEOUT). The autobaud loader's window is its calibration poll budget
(--autobaud-polls); the seconds it amounts to are budget x 9 / f_cpu, the
measured cost of the calibrate() wait loop this gate pins.
"""
import argparse
import importlib.util
import pathlib
import select
import sys
import time
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
from pbsim import Device
# The calibrate() budget loop's cycles per poll in the built image - what the
# README's window arithmetic rests on, verified here. A measured fact, not a
# design constant: the wait's exit branches land where the compiler's block
# layout puts them, and the bounded-calibration rework moved the loop from
# ten cycles to nine.
AUTOBAUD_POLL_CYCLES = 9
def load_tool(path):
spec = importlib.util.spec_from_file_location("pureboot", path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def compose_flash(pb, loader_bytes, app_bytes, mcu, base, page):
"""The flash image a completed programming session leaves: application
(with the tinies' vector surgery), loader at base - built through the
host tool's own planner so the surgery is the shipped one, not a copy."""
flash_size = base + pb.SLOT
patch = not mcu.startswith("atmega") or mcu.startswith("atmega48")
word_flash = flash_size > 0x10000
wire_base = base // 2 if word_flash else base
flags = (1 if patch else 0) | (2 if word_flash else 0)
raw = bytes((ord("P"), ord("B"), 5, 0, 0, 0, page & 0xFF,
wire_base & 0xFF, wire_base >> 8, 0, 0, flags))
info = pb.Info(raw)
flash = bytearray(b"\xff" * flash_size)
for address, content in pb.plan_flash(app_bytes, info).items():
flash[address:address + len(content)] = content
flash[base:base + len(loader_bytes)] = loader_bytes
return bytes(flash)
def first_tx_cycle(device, deadline):
"""The PB_WINDOW_TX report, or None. The runner prints it once."""
stream = device.proc.stdout
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
return None
ready, _, _ = select.select([stream], [], [], remaining)
if not ready:
return None
line = stream.readline()
if not line:
return None
if line.startswith("PB_WINDOW_TX"):
return int(line.split()[1])
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--device", required=True)
parser.add_argument("--loader", required=True)
parser.add_argument("--mcu", required=True)
parser.add_argument("--hz", type=int, required=True)
parser.add_argument("--base", required=True)
parser.add_argument("--page", type=int, required=True)
parser.add_argument("--baud", type=int, required=True)
parser.add_argument("--app", required=True)
parser.add_argument("--tool", required=True)
parser.add_argument("--workdir", required=True)
parser.add_argument("--link", default=None)
parser.add_argument("--seconds", type=float, default=None)
parser.add_argument("--autobaud-polls", type=int, default=None)
args = parser.parse_args()
if (args.seconds is None) == (args.autobaud_polls is None):
parser.error("exactly one of --seconds / --autobaud-polls")
pb = load_tool(args.tool)
base = int(args.base, 0)
expected = (args.seconds if args.seconds is not None
else args.autobaud_polls * AUTOBAUD_POLL_CYCLES / args.hz)
work = pathlib.Path(args.workdir)
work.mkdir(parents=True, exist_ok=True)
# Every loader target objcopies its slot content beside the ELF (.bin).
loader_bytes = pathlib.Path(args.loader + ".bin").read_bytes()
app_bytes = pathlib.Path(args.app).read_bytes()
flash_file = work / "window-flash.bin"
flash_file.write_bytes(compose_flash(pb, loader_bytes, app_bytes, args.mcu, base, args.page))
device = Device(args.device, args.loader, args.mcu, str(args.hz), args.base, args.page,
args.baud, str(work / "window-dump.bin"), resume=str(flash_file),
link=args.link, window=True)
try:
# Simulation speed is machine-dependent; a few hundred thousand
# cycles per wall second is the pessimistic floor.
budget = max(60.0, expected * args.hz / 300000)
cycle = first_tx_cycle(device, time.monotonic() + budget)
finally:
device.stop()
if cycle is None:
print(f" [FAIL] no transmit activity within {budget:.0f} s wall "
f"(expected a {expected:.2f} s window)")
return 1
measured = cycle / args.hz
error = (measured - expected) / expected
ok = abs(error) <= 0.02
print(f" [{'PASS' if ok else 'FAIL'}] window {measured:.3f} s vs declared "
f"{expected:.3f} s ({error:+.1%}, gate +/-2%)")
return 0 if ok else 1
if __name__ == "__main__":
raise SystemExit(main())

771
test/pureboot_device.cpp
View File

@@ -1,771 +0,0 @@
// simavr "device" for the pureboot protocol tests, every chip. 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:
//
// - Hardware USART builds: simavr's uart_pty on the selected instance.
// - Software UART builds: an 8N1 bridge between a pty and the GPIO pins,
// timed against the simulated cycle counter (drives the loader's RX,
// decodes its TX).
//
// The link follows the chip's natural default (USART0 on the megas, the
// software UART on PB0/PB1 elsewhere) unless -l overrides it: `-l usart1`
// for the second instance, `-l sw:B5,B1` for a software build's RX,TX pins,
// and `-l sw:D0,D1@0` where those pins are a USART's own - see the pin
// ownership the bridge models below.
//
// 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. The fuse readout's LPM diversion is missing from every core and
// is supplied too, so a fuse read answers fuses (-f) and not flash bytes.
//
// 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 <array>
#include <charconv>
#include <csignal>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <print>
#include <string_view>
#include <fcntl.h>
#include <pty.h>
#include <termios.h>
#include <unistd.h>
// The parts headers (uart_pty.h) carry no C++ linkage guards of their own,
// unlike simavr's core headers - the block covers both harmlessly.
extern "C" {
#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"
}
namespace {
avr_t *avr;
uart_pty_t uart_pty;
bool link_software;
avr_uart_t *hw_uart; // the pty-driven USART, for the datasheet-reset fix below
char uart_digit = '0';
char sw_rx_port = 'B', sw_tx_port = 'B';
int sw_rx_bit = 0, sw_tx_bit = 1;
char sw_tx_owner = 0; // the USART whose TXD the software link sits on
const char *dump_path;
std::uint32_t reset_pc;
volatile std::sig_atomic_t reset_requested;
// -w: report the cycle of the first transmit activity, once. What the
// activation-window gate reads - with an idle line and an application
// installed, the first thing that ever talks is the application's banner,
// so this cycle *is* the loader's window plus a banner lead measured in
// microseconds. Idle pacing is skipped in this mode: there is no real-time
// host in the loop, and a paced multi-second window would take hours.
bool window_report;
bool window_tx_seen;
void window_first_tx()
{
if (!window_report || window_tx_seen) {
return;
}
window_tx_seen = true;
std::println("PB_WINDOW_TX {}", avr->cycle);
std::fflush(stdout);
}
void window_uart_hook(avr_irq_t *, std::uint32_t, void *)
{
window_first_tx();
}
// One-wire (RX == TX in the link spec): both directions on one GPIO line
// idling on the firmware's pull-up. The bridge then follows the pin's
// direction the way the real wiring does: it drives only while the
// firmware's DDR bit reads input, decodes transitions as the firmware's
// transmit only while the firmware owns the line, ignores its own raises
// coming back through the shared irq - and echoes every byte it drives back
// to the pty, which is what the host-side FTDI tie does and what the host
// tool's --one-wire mode reads back and discards.
bool link_one_wire;
bool mcu_owns_line;
bool self_drive;
int parse_link(std::string_view spec)
{
if (spec == "usart0" || spec == "usart1") {
link_software = false;
uart_digit = spec[5];
return 0;
}
if (spec.starts_with("sw")) {
link_software = true;
if (spec.size() == 2) {
return 0;
}
char owner = 0;
int fields =
std::sscanf(spec.data() + 2, ":%c%d,%c%d@%c", &sw_rx_port, &sw_rx_bit, &sw_tx_port, &sw_tx_bit, &owner);
if (fields == 4 || fields == 5) {
sw_tx_owner = owner;
link_one_wire = sw_rx_port == sw_tx_port && sw_rx_bit == sw_tx_bit;
return 0;
}
}
return -1;
}
// simavr 1.6's avr_flash PGERS handler erases spm_pagesize bytes starting at
// Z & ~1 instead of the page containing Z (its PGWRT path masks correctly) -
// hardware ignores the in-page bits (section 26.8.1), so an erase issued with Z
// anywhere inside the page wipes half the neighbouring page in simulation
// only. Wrap the mega's registered flash ioctl and re-dispatch page erases
// with Z forced to the page boundary; everything else passes through.
//
// A second gap on the boot-section-less m48s: their RWWSRE bit is the
// temporary-buffer discard (Atmel-8271 section 26.2/section 26.3.1), but the stock model
// gates its RWWSRE branch on AVR_SELFPROG_HAVE_RWW - absent on the m48
// core - so the discard store falls through into the buffer-fill branch and
// plants whatever Z/R1:R0 happen to hold. Perform the silicon's discard
// here instead.
avr_flash_t *mega_flash;
int (*mega_flash_ioctl)(avr_io_t *io, std::uint32_t ctl, void *param);
int fixed_flash_ioctl(avr_io_t *io, std::uint32_t ctl, void *param)
{
if (ctl == AVR_IOCTL_FLASH_SPM && avr_regbit_get(io->avr, mega_flash->pgers)) {
auto z = static_cast<std::uint16_t>(io->avr->data[30] | (io->avr->data[31] << 8));
auto masked = static_cast<std::uint16_t>(z & ~(mega_flash->spm_pagesize - 1));
io->avr->data[30] = static_cast<std::uint8_t>(masked);
io->avr->data[31] = static_cast<std::uint8_t>(masked >> 8);
int result = mega_flash_ioctl(io, ctl, param);
io->avr->data[30] = static_cast<std::uint8_t>(z);
io->avr->data[31] = static_cast<std::uint8_t>(z >> 8);
return result;
}
if (ctl == AVR_IOCTL_FLASH_SPM && !(mega_flash->flags & AVR_SELFPROG_HAVE_RWW) &&
(io->avr->data[mega_flash->r_spm] & 0x11) == 0x11) { // RWWSRE|SELFPRGEN: the m48 buffer discard
for (int i = 0; i < mega_flash->spm_pagesize / 2; i++) {
mega_flash->tmppage[i] = 0xffff;
mega_flash->tmppage_used[i] = 0;
}
avr_regbit_clear(io->avr, mega_flash->selfprgen);
return 0;
}
return mega_flash_ioctl(io, ctl, param);
}
void fix_mega_flash_erase()
{
for (avr_io_t *io = avr->io_port; io; io = io->next) {
if (io->kind && std::string_view{io->kind} == "flash") {
mega_flash = reinterpret_cast<avr_flash_t *>(io);
mega_flash_ioctl = io->ioctl;
io->ioctl = fixed_flash_ioctl;
return;
}
}
std::println(stderr, "device: no flash module to fix - SPM page erases may misalign");
}
// --------------------------------------------------------------- fuses ---
// The fuse and lock bytes answer an LPM, not an SPM: BLBSET|SELFPRGEN in
// SPMCSR diverts the next LPM to them, selected by Z (Atmel-8271 section
// 26.8.9). simavr executes LPM straight out of avr->flash and offers no hook
// on it, so the diversion is modeled where there is one - the SPMCSR write -
// by lending the four flash bytes Z can select to the fuses for as long as the
// hardware holds SELFPRGEN. Unprogrammed until -f says otherwise, as a part
// ships and as the erased flash above is.
auto fuses = std::to_array<std::uint8_t>({0xFF, 0xFF, 0xFF, 0xFF}); // Z order: low, lock, extended, high
decltype(fuses) lent;
int parse_fuses(std::string_view spec)
{
if (spec.size() != 2 * fuses.size()) {
return -1;
}
for (std::size_t at = 0; at < fuses.size(); at++) {
const auto digits = spec.substr(2 * at, 2);
if (std::from_chars(digits.data(), digits.data() + digits.size(), fuses[at], 16).ec != std::errc{}) {
return -1;
}
}
return 0;
}
avr_cycle_count_t end_fuse_read(avr_t *mcu, avr_cycle_count_t, void *)
{
std::memcpy(mcu->flash, lent.data(), lent.size());
return 0;
}
void spmcsr_written(avr_t *mcu, avr_io_addr_t at, std::uint8_t value, void *)
{
// A registered write handler *replaces* the store simavr would have done
// (sim_core.c), so performing it is this handler's job - on the tinies
// nothing else is watching the register, and swallowing the store would
// leave every SPM command unseen by the NVM model above.
avr_core_watch_write(mcu, at, value);
constexpr std::uint8_t read_fuse_command = 0x09; // BLBSET|SELFPRGEN
// The LPM must follow within three cycles of the arming store (Atmel-8271
// section 26.8.9), and simavr runs its cycle timers between instructions -
// so the loan is returned after the LPM that took it, never during.
constexpr avr_cycle_count_t selfprgen_window = 3;
if ((value & 0x1F) != read_fuse_command) {
return;
}
std::memcpy(lent.data(), mcu->flash, lent.size());
std::memcpy(mcu->flash, fuses.data(), fuses.size());
avr_cycle_timer_register(mcu, selfprgen_window, end_fuse_read, nullptr);
}
void request_reset(int)
{
reset_requested = 1;
}
// ------------------------------------------------------------- tiny NVM ---
// Where SPMCSR sits on the cores that carry no flash module to name it.
constexpr avr_io_addr_t tiny_spmcsr = 0x57;
struct tiny_nvm_t {
avr_io_t io;
std::array<std::uint8_t, 128> buffer;
std::array<std::uint8_t, 128> used; // a buffer word loads once until erased - like silicon
unsigned page;
};
tiny_nvm_t nvm;
int nvm_ioctl(avr_io_t *io, std::uint32_t ctl, void *)
{
if (ctl != AVR_IOCTL_FLASH_SPM) {
return -1;
}
auto *n = reinterpret_cast<tiny_nvm_t *>(io);
avr_t *mcu = io->avr;
std::uint8_t command = mcu->data[tiny_spmcsr] & 0x1f;
auto z = static_cast<std::uint16_t>(mcu->data[30] | (mcu->data[31] << 8));
std::uint32_t page_base = static_cast<std::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;
if (!n->used[offset]) { // first write wins until the buffer clears
n->buffer[offset] = mcu->data[0];
n->buffer[offset + 1] = mcu->data[1];
n->used[offset] = 1;
}
} else if (command == 0x03) { // PGERS
std::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];
}
std::memset(n->buffer.data(), 0xff, n->page);
std::memset(n->used.data(), 0, n->page);
} else if (command == 0x11) { // CTPB
std::memset(n->buffer.data(), 0xff, n->page);
std::memset(n->used.data(), 0, n->page);
}
mcu->data[tiny_spmcsr] &= static_cast<std::uint8_t>(~0x1f); // the operation completes instantly
return 0;
}
// ----------------------------------------------------------- GPIO bridge ---
int pty_master = -1;
avr_irq_t *rx_pin; // the loader's RX (PB0), driven from the pty
avr_cycle_count_t bit_cycles;
int tx_level = 1, tx_active, tx_bit;
std::uint8_t tx_shift;
avr_cycle_count_t tx_sample(avr_t *, avr_cycle_count_t when, void *)
{
if (tx_bit < 0) {
// Half a bit into the start bit: a real receiver re-samples here and
// abandons a false start. The device's own init produces one - DDR
// drives the pin low for the instructions until the idle level is
// written - and without this check that glitch decodes as a stray
// byte (and would read as first transmit activity under -w).
if (tx_level) {
tx_active = 0;
return 0;
}
window_first_tx();
tx_bit = 0;
return when + bit_cycles;
}
if (tx_bit < 8) {
tx_shift = static_cast<std::uint8_t>((tx_shift >> 1) | (tx_level ? 0x80 : 0));
if (++tx_bit < 8) {
return when + bit_cycles;
}
// The byte is delivered at the stop bit's sampling point (9.5 bit
// times), where a hardware receiver raises its RXC - not sooner: a
// host answering before the stop bit would put its start bit on the
// wire while the device is still driving, which the device,
// transmitting, is not watching for.
return when + bit_cycles;
}
if (write(pty_master, &tx_shift, 1) != 1) {
std::println(stderr, "device: pty write lost a byte");
}
tx_active = 0;
return 0;
}
// A USART owns its TxD pin whenever its transmitter is enabled, and the port
// register cannot drive it (section 20.2 / Atmel-8271 section 19.2) - which is why a
// bit-banged link deployed on those pins is mute until it clears UCSRnB.
// simavr wires a USART entirely through IRQs and never touches the port pin
// model, so the ownership does not exist there and the mute cannot happen:
// supply it, or the very state this models is untestable. The link spec's
// trailing @n names the USART; without one the pins are nobody's.
avr_uart_t *tx_owner;
bool tx_pin_taken()
{
if (!tx_owner) {
return false;
}
if (avr_regbit_get(avr, tx_owner->txen)) {
return true;
}
// One-wire on the USART's RXD: RXEN forces the shared pin's direction to
// input (section 20.7.3), so the firmware's drive goes nowhere until the
// release - the receive-side twin of the TXD hold.
return link_one_wire && avr_regbit_get(avr, tx_owner->rxen);
}
// simavr leaves TXEN set in UCSRnB out of reset, where silicon clears the
// whole register (section 20.11.3) - which would hand the pin to a USART no code has
// enabled, making a freshly reset chip mute for reasons hardware does not
// have. Reset it the way the datasheet does, so the ownership starts from
// nobody's and only an application that really enables the USART takes it.
void reset_tx_owner()
{
if (tx_owner) {
avr_regbit_clear(avr, tx_owner->txen);
}
}
void find_tx_owner()
{
for (avr_io_t *io = avr->io_port; io; io = io->next) {
if (io->kind && std::string_view{io->kind} == "uart" &&
reinterpret_cast<avr_uart_t *>(io)->name == sw_tx_owner) {
tx_owner = reinterpret_cast<avr_uart_t *>(io);
reset_tx_owner();
return;
}
}
std::println(stderr, "device: no USART{} to own the software link's TX pin", sw_tx_owner);
}
void tx_hook(avr_irq_t *, std::uint32_t value, void *)
{
if (link_one_wire && (self_drive || !mcu_owns_line)) {
// The bridge's own drive coming back through the shared irq, or a
// transition while the line is the bridge's - either way not the
// firmware talking: the decoder sees an idle line.
tx_level = 1;
return;
}
if (tx_pin_taken()) { // the USART holds the line; the port write goes nowhere
tx_level = 1;
return;
}
int level = value & 1;
if (!tx_active && tx_level == 1 && level == 0) { // start edge, confirmed mid-bit
tx_active = 1;
tx_bit = -1;
avr_cycle_timer_register(avr, bit_cycles / 2, tx_sample, nullptr);
}
tx_level = level;
}
std::array<std::uint8_t, 8192> rx_queue;
unsigned rx_head, rx_tail; // ring: head = next to send
int rx_active, rx_bit;
std::uint8_t rx_byte;
void rx_start_next();
// Every level the bridge itself puts on the line goes through here, so the
// shared-pin decoder can tell its own drive from the firmware's.
void bridge_drive(int level)
{
self_drive = true;
avr_raise_irq(rx_pin, static_cast<std::uint32_t>(level));
self_drive = false;
}
avr_cycle_count_t rx_step(avr_t *, avr_cycle_count_t when, void *)
{
if (rx_bit < 8) {
bridge_drive((rx_byte >> rx_bit) & 1);
rx_bit++;
return when + bit_cycles;
}
if (rx_bit == 8) { // stop bit, plus one idle bit of margin
bridge_drive(1);
// The host-side tie: an FTDI adapter on a one-wire line reads every
// byte it transmits - supply that echo, which the host tool's
// --one-wire mode consumes as its wiring check.
if (link_one_wire && write(pty_master, &rx_byte, 1) != 1) {
std::println(stderr, "device: pty echo lost a byte");
}
rx_bit++;
return when + 2 * bit_cycles;
}
rx_active = 0;
rx_start_next();
return 0;
}
void rx_start_next()
{
if (rx_active || rx_head == rx_tail) {
return;
}
// The firmware is answering on the shared line: hold the byte - a real
// host's transmission waits out the reply on the wire too. The next
// poll_pty tick retries once the line is handed back.
if (link_one_wire && mcu_owns_line) {
return;
}
rx_byte = rx_queue[rx_head];
rx_head = (rx_head + 1) % rx_queue.size();
rx_active = 1;
rx_bit = 0;
bridge_drive(0); // start bit
avr_cycle_timer_register(avr, bit_cycles, rx_step, nullptr);
}
// The shared pin's direction is the line's ownership: DDR-out is the
// firmware driving a frame, DDR-in hands the line back to the bridge.
void on_ddr(avr_irq_t *, std::uint32_t value, void *)
{
const bool owns = (value >> sw_rx_bit) & 1;
if (mcu_owns_line && !owns) {
bridge_drive(1); // hand-back: a turn-based host idles here, and the cache stays truthful
}
mcu_owns_line = owns;
// A byte held back while the firmware answered starts from the next
// poll_pty tick, never from inside the DDR write itself - the port
// model's own pull-up re-derivation runs right after this notify and
// would erase a start edge raised here.
}
// A reset abandons whatever the bridge was mid-transfer: bytes still queued
// for a chip that no longer has the context to receive them meaningfully,
// and a decode in progress on a TX line the reset may have already changed.
// The pending cycle timers must go with the state: avr_reset drops the TX
// output latch, whose falling edge starts a spurious decode before this
// runs, and a stale tx_sample would then interleave with the loader's first
// real answer through the shared shift state, corrupting it.
void bridge_reset()
{
avr_cycle_timer_cancel(avr, tx_sample, nullptr);
avr_cycle_timer_cancel(avr, rx_step, nullptr);
rx_head = rx_tail = 0;
rx_active = 0;
tx_active = 0;
tx_level = 1;
mcu_owns_line = false; // avr_reset zeroed DDR: every pin reads input again
// Re-drive the idle line through a forced transition: ioport pin irqs are
// IRQ_FLAG_FILTERED, and avr_reset zeroes the port latch while the irq
// keeps its pre-reset cached value - so a plain raise(1) against a cached
// 1 is dropped and the device reads the line stuck low. A loader entering
// calibration on that line measures reset-to-first-edge as one giant
// pulse and mis-locks or boots the application on the first real knock.
// No cycles run between the two raises, so the device only ever sees the
// final idle-high.
bridge_drive(0);
bridge_drive(1);
}
void poll_pty()
{
std::array<std::uint8_t, 256> chunk;
ssize_t got = read(pty_master, chunk.data(), chunk.size());
for (ssize_t i = 0; i < got; i++) {
unsigned next = (rx_tail + 1) % rx_queue.size();
if (next == rx_head) {
break; // full: the host will retry on timeout
}
rx_queue[rx_tail] = chunk[i];
rx_tail = next;
}
// Unconditional: a byte held back while the firmware owned a shared
// line restarts from here once the hand-back has happened.
rx_start_next();
}
// ------------------------------------------------------------------ main ---
[[noreturn]] void finish(int)
{
if (dump_path) {
std::FILE *f = std::fopen(dump_path, "wb");
if (f) {
std::fwrite(avr->flash, 1, avr->flashend + 1, f);
std::fclose(f);
}
avr_eeprom_desc_t ee = {
.ee = nullptr,
.offset = 0,
.size = 0,
};
if (avr_ioctl(avr, AVR_IOCTL_EEPROM_GET, &ee) == 0 && ee.ee && ee.size) {
std::array<char, 512> path;
std::snprintf(path.data(), path.size(), "%s.eeprom", dump_path);
f = std::fopen(path.data(), "wb");
if (f) {
std::fwrite(ee.ee, 1, ee.size, f);
std::fclose(f);
}
}
}
if (!link_software) {
uart_pty_stop(&uart_pty);
}
_exit(0);
}
} // namespace
int main(int argc, char *argv[])
{
bool link_given = false;
for (int opt; (opt = getopt(argc, argv, "l:wf:")) != -1;) {
if (opt == 'w') {
window_report = true;
continue;
}
if (opt == 'f') {
if (parse_fuses(optarg) != 0) {
std::println(stderr, "device: -f takes 8 hex digits: low,lock,extended,high");
return 2;
}
continue;
}
if (opt != 'l' || parse_link(optarg) != 0) {
std::println(stderr, "device: bad link spec (usart0, usart1, sw, or sw:B0,B1 as RX,TX)");
return 2;
}
link_given = true;
}
int args = argc - optind;
if (args < 7 || args > 9) {
std::print(stderr,
"usage: {} [-l link] [-w] [-f fuses] <pureboot.elf> <mcu> <hz> <base_hex> <page> <baud>"
" <flash_dump> [reset_hex] [resume_flash]\n"
" -l link: usart0 | usart1 | sw[:B0,B1[@0]] (RX,TX, then the USART owning\n"
" them); default: the chip's own\n"
" -f fuses: 8 hex digits, low,lock,extended,high - what a fuse read answers;\n"
" default: unprogrammed. reset_hex stays the reset target\n"
" -w: print PB_WINDOW_TX <cycle> at the first transmit activity and\n"
" free-run idle time (window measurement mode)\n"
" reset_hex: reset vector (default: base with a boot section, else 0)\n"
" resume_flash: raw full-flash image loaded instead of the ELF - a prior\n"
" run's dump, for power-fail resume tests\n",
argv[0]);
return 2;
}
argv += optind - 1; // argv[1] is the ELF again, whatever was parsed
const std::string_view mcu_name = argv[2];
auto base = static_cast<std::uint32_t>(std::strtoul(argv[4], nullptr, 0));
auto page = static_cast<unsigned>(std::atoi(argv[5]));
auto baud = static_cast<unsigned>(std::atoi(argv[6]));
dump_path = argv[7];
const bool is_mega = mcu_name.starts_with("atmega");
if (!link_given) {
link_software = !is_mega; // the chips' natural links: USART0, or PB0/PB1
}
avr = avr_make_mcu_by_name(mcu_name.data());
if (!avr) {
std::println(stderr, "device: no {} core", mcu_name);
return 1;
}
avr_init(avr);
avr->frequency = static_cast<std::uint32_t>(std::strtoul(argv[3], nullptr, 0));
std::memset(avr->flash, 0xff, avr->flashend + 1); // real flash powers up erased
if (args > 8) {
// Resume: the full flash image of an interrupted prior run.
std::FILE *f = std::fopen(argv[9], "rb");
if (!f || std::fread(avr->flash, 1, avr->flashend + 1, f) == 0) {
std::println(stderr, "device: cannot read {}", argv[9]);
return 1;
}
std::fclose(f);
} else {
elf_firmware_t fw{};
if (elf_read_firmware(argv[1], &fw) != 0) {
std::println(stderr, "device: cannot read {}", argv[1]);
return 1;
}
// An image past flash end would smash the simulator's heap and turn
// into phantom peripheral behavior (lessons: believe the size gate
// first) - refuse it loudly instead.
if (base + fw.flashsize > avr->flashend + 1) {
std::println(stderr, "device: {} B at {:#x} runs past flash end {:#x} - image does not fit its slot",
fw.flashsize, base, avr->flashend);
return 1;
}
std::memcpy(avr->flash + base, fw.flash, fw.flashsize);
}
// The boot-sectioned megas enter the loader in hardware (BOOTRST, not
// modeled - the argument picks the modeled fuse's target); the tinies
// and the boot-section-less m48s reset to word 0 like silicon - erased
// flash walks up into the loader, and after the host's surgery the
// patched vector routes there.
const bool boot_section = is_mega && !mcu_name.starts_with("atmega48");
reset_pc = args > 7 ? static_cast<std::uint32_t>(std::strtoul(argv[8], nullptr, 0)) : (boot_section ? base : 0);
avr->pc = reset_pc;
avr->codeend = avr->flashend;
// Erased EEPROM, as hardware powers up (simavr zeroes it).
std::array<std::uint8_t, 1024> blank;
std::memset(blank.data(), 0xff, blank.size());
avr_eeprom_desc_t seed = {
.ee = blank.data(),
.offset = 0,
.size = 0,
};
if (avr_ioctl(avr, AVR_IOCTL_EEPROM_GET, &seed) == 0 && seed.size <= blank.size()) {
seed.ee = blank.data();
avr_ioctl(avr, AVR_IOCTL_EEPROM_SET, &seed);
}
// The megas carry simavr's avr_flash module (and its two gaps the wrap
// above fixes); the tinies get the NVM module simavr lacks. Which serial
// bridge runs is the link's business, not the chip class's.
if (is_mega) {
fix_mega_flash_erase();
} else {
nvm.page = page;
std::memset(nvm.buffer.data(), 0xff, nvm.buffer.size());
nvm.io.kind = "tiny_nvm";
nvm.io.ioctl = nvm_ioctl;
avr_register_io(avr, &nvm.io);
}
// Where the fuse read is armed: the mega's flash module names its SPMCSR,
// and the tinies keep theirs at the address both those cores share.
avr_register_io_write(avr, mega_flash ? mega_flash->r_spm : tiny_spmcsr, spmcsr_written, nullptr);
if (!link_software) {
// POLL_SLEEP paces an idle-polling loader in host real time (a
// no-hardware CPU-saving hack); clear it so cycles run free.
std::uint32_t flags = 0;
avr_ioctl(avr, AVR_IOCTL_UART_GET_FLAGS(uart_digit), &flags);
flags &= ~AVR_UART_FLAG_POLL_SLEEP;
avr_ioctl(avr, AVR_IOCTL_UART_SET_FLAGS(uart_digit), &flags);
// simavr leaves TXEN set out of reset where silicon clears the whole
// UCSR#B (section 20.11.3). Harmless to a loader that enables TXEN itself -
// but a half-duplex build's receiver-only init then *drops* TXEN,
// and this uart model clears UDRE on that edge and never re-raises
// it on a later enable: the first transmitter after the hand-over
// waits UDRE forever, a wedge silicon does not have. Start from the
// datasheet's zero, as the software bridge's tx-owner model does.
for (avr_io_t *io = avr->io_port; io; io = io->next) {
if (io->kind && std::string_view{io->kind} == "uart" &&
reinterpret_cast<avr_uart_t *>(io)->name == uart_digit) {
hw_uart = reinterpret_cast<avr_uart_t *>(io);
}
}
if (hw_uart) {
avr_regbit_clear(avr, hw_uart->txen);
}
uart_pty_init(avr, &uart_pty);
uart_pty_connect(&uart_pty, uart_digit);
if (window_report) {
avr_irq_register_notify(avr_io_getirq(avr, AVR_IOCTL_UART_GETIRQ(uart_digit), UART_IRQ_OUTPUT),
window_uart_hook, nullptr);
}
std::println("PB_PTY {}", uart_pty.pty.slavename);
} else {
bit_cycles = (avr->frequency + baud / 2) / baud; // matches uart.hpp's own rounding exactly
if (sw_tx_owner) {
find_tx_owner();
}
rx_pin = avr_io_getirq(avr, AVR_IOCTL_IOPORT_GETIRQ(sw_rx_port), static_cast<unsigned>(sw_rx_bit));
avr_irq_register_notify(
avr_io_getirq(avr, AVR_IOCTL_IOPORT_GETIRQ(sw_tx_port), static_cast<unsigned>(sw_tx_bit)), tx_hook,
nullptr);
if (link_one_wire) {
avr_irq_register_notify(avr_io_getirq(avr, AVR_IOCTL_IOPORT_GETIRQ(sw_rx_port), IOPORT_IRQ_DIRECTION_ALL),
on_ddr, nullptr);
}
bridge_drive(1); // idle line
int slave;
struct termios raw;
cfmakeraw(&raw);
if (openpty(&pty_master, &slave, nullptr, &raw, nullptr) != 0) {
std::println(stderr, "device: openpty failed");
return 1;
}
fcntl(pty_master, F_SETFL, O_NONBLOCK);
std::println("PB_PTY {}", ttyname(slave));
}
std::fflush(stdout);
std::signal(SIGTERM, finish);
std::signal(SIGINT, finish);
std::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 (!link_software) { // reset restores the pacing hack; re-clear it
std::uint32_t flags = 0;
avr_ioctl(avr, AVR_IOCTL_UART_GET_FLAGS(uart_digit), &flags);
flags &= ~AVR_UART_FLAG_POLL_SLEEP;
avr_ioctl(avr, AVR_IOCTL_UART_SET_FLAGS(uart_digit), &flags);
if (hw_uart) { // and simavr's bogus reset TXEN (section 20.11.3: zero)
avr_regbit_clear(avr, hw_uart->txen);
}
} else {
bridge_reset();
reset_tx_owner();
}
}
if (link_software && ++since_poll >= 2000) {
since_poll = 0;
poll_pty();
// An unthrottled idle simulation runs the activation window out
// from under the host's real-time knock cadence: a 1 MHz build's
// 8 s window is 8 M cycles - tens of wall milliseconds - so a
// first knock lost to an in-flight reset misses the window
// entirely. Pace the simulation only while the bridge is fully
// quiet (nothing decoding, nothing queued); transfers keep full
// speed, and a quiet window stretches toward real time.
if (!window_report && !rx_active && !tx_active && rx_head == rx_tail) {
usleep(200);
}
}
}
finish(0);
}

212
test/test_handshake.py
View File

@@ -1,212 +0,0 @@
#!/usr/bin/env python3
"""Host-tool activation handshake: bounded against a line that misbehaves.
`_handshake` drains the line after it sees a prompt, to absorb a real loader's
trailing bytes before it asks for the identity. That drain must be bounded: a
target that never falls quiet - a board stuck in a reset loop presents exactly
this, ~60 reboots/s of UART-reset garbage in which a stray 0x2b reads as a
prompt - otherwise spins the tool forever. Regression for that hang, plus a
control that a well-behaved loader still connects.
The handshake must also survive its own leftovers: after `--stay` the loader's
final prompt can still be in the USB pipeline when the next invocation opens
the port, and on a board wired to reset on open, that opening starts a fresh
activation window the stale prompt then betrays - the tool commits to an
identity read against a device that never heard its knock, and what it finally
collects is the application's banner. StaleDTRPort is that moment as a port.
Stdlib only, no device: host-tool logic, so it runs on every chip's preset
beside pureboot.planner.
"""
import importlib.util
import pathlib
import threading
import time
PB = pathlib.Path(__file__).resolve().parents[1] / "pureboot" / "pureboot.py"
_spec = importlib.util.spec_from_file_location("pureboot", PB)
pb = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(pb)
P = F = 0
def check(name, ok):
global P, F
P, F = P + (1 if ok else 0), F + (0 if ok else 1)
print(f" [{'PASS' if ok else 'FAIL'}] {name}")
class FloodPort:
"""A line that never falls quiet: read_available always returns bytes, and
they contain a prompt. No identity ever completes."""
def flush_input(self):
pass
def write(self, data):
pass
def read_available(self, wait):
time.sleep(0.01) # a real read waits; keep the busy loop off a core
return b"+\x00\xff"
def read_exact(self, count, timeout):
raise pb.Error("no identity")
class LoaderPort:
"""A well-behaved pureboot 5: a prompt to the knock, then quiet, then the
slim identity (version 5 + m328p signature) and a closing prompt."""
def __init__(self):
self.pending = b""
self.exacts = 0
def flush_input(self):
self.pending = b""
def write(self, data):
if b"p" in data:
self.pending = b"+" # the prompt answers the knock, nothing else
def read_available(self, wait):
data, self.pending = self.pending, b""
return data
def read_exact(self, count, timeout):
self.exacts += 1
return b"\x05\x1e\x95\x0f" if self.exacts == 1 else b"+" # identity, then prompt
class StaleDTRPort:
"""`--stay`, then a fresh invocation on a board that resets when its port
opens. Three facts of that moment, all timed from the open: the previous
session's final prompt is still in transit and lands only after the
opening flush has already run; the reset holds the device off the line
at first, eating anything written before it completes; and the fresh
window is finite - once it expires the application boots and prints a
banner whose bytes are what a pending identity read collects. A
handshake that trusts the stale prompt spends the whole window waiting
on a device that never heard its knock; one that drains the line first
knocks into the real window and connects."""
STALE_AT = 0.02 # the leftover prompt becomes visible (post-flush)
READY_AT = 0.05 # reset complete, activation window opens
WINDOW = 1.0 # window length; expiry boots the application
def __init__(self):
self.t0 = time.monotonic()
# (visible-from, bytes): the line as a timed queue.
self.queue = [(self.t0 + self.STALE_AT, b"+")]
self.armed = False # a 'p' heard inside the window arms 'b'
self.booted = False
def _boot_check(self):
if not self.booted and time.monotonic() > self.t0 + self.READY_AT + self.WINDOW:
self.booted = True
self.queue.append((self.t0 + self.READY_AT + self.WINDOW,
b"W r libavr tempmon\r\n"))
def _visible(self):
self._boot_check()
now = time.monotonic()
return b"".join(d for t, d in self.queue if t <= now)
def _consume(self, n):
now = time.monotonic()
left = []
for t, d in self.queue:
if t <= now and n:
take = min(n, len(d))
d = d[take:]
n -= take
if d:
left.append((t, d))
self.queue = left
def flush_input(self):
self._consume(len(self._visible()))
def write(self, data):
self._boot_check()
now = time.monotonic()
if now < self.t0 + self.READY_AT or self.booted:
return # still in reset, or the application owns the line
if b"p" in data:
self.armed = True
self.queue.append((now + 0.01, b"+"))
if b"b" in data and self.armed:
# The slim identity (version 5 + m328p signature) and a prompt.
self.queue.append((now + 0.01, b"\x05\x1e\x95\x0f+"))
def read_available(self, wait):
deadline = time.monotonic() + wait
while True:
data = self._visible()
if data:
self._consume(len(data))
return data
if time.monotonic() >= deadline:
return b""
time.sleep(0.005)
def read_exact(self, count, timeout):
deadline = time.monotonic() + timeout
data = b""
while len(data) < count:
visible = self._visible()
if visible:
take = visible[:count - len(data)]
self._consume(len(take))
data += take
elif time.monotonic() >= deadline:
raise pb.Error(f"timeout: got {len(data)} of {count} bytes")
else:
time.sleep(0.005)
return data
def terminates(port, wait, budget):
"""Run connect_autobaud in a thread; True if it returns/raises within
`budget` seconds rather than hanging."""
done = threading.Event()
def run():
try:
pb.Loader(port).connect_autobaud(wait)
except Exception:
pass
finally:
done.set()
threading.Thread(target=run, daemon=True).start()
return done.wait(budget)
def main():
# the hang: a flooding target must not spin the drain forever. With wait=0.5
# the whole handshake has to give up well inside a few seconds.
check("flooding target: handshake terminates, drain is bounded",
terminates(FloodPort(), wait=0.5, budget=4.0))
# the control: a real loader still connects and reads identity.
info = pb.Loader(LoaderPort()).connect_autobaud(2.0)
check("well-behaved loader still connects (version 5)", info.version == 5)
# the stale prompt: a --stay leftover plus reset-on-open must not burn the
# fresh window - the pre-knock drain absorbs it and the first real knock
# lands inside the window.
try:
stale_ok = pb.Loader(StaleDTRPort()).connect(2.5).version == 5
except pb.Error as failed:
print(f" ({failed})")
stale_ok = False
check("stale --stay prompt + reset-on-open: connects in the fresh window", stale_ok)
print(f"\n {P} passed, {F} failed")
return 1 if F else 0
if __name__ == "__main__":
raise SystemExit(main())

367
test/test_planner.py
View File

@@ -1,367 +0,0 @@
#!/usr/bin/env python3
"""Host-tool unit tests - the planning and policy logic, no simulator:
programming orders and their recovery properties, the reset-vector surgery,
the staging composition, the boot-fuse decode, and the update preflight over
fuse combinations simavr cannot model.
Usage: test_planner.py <tool_py>
"""
import sys
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def expect_error(what, fn, *needles):
try:
fn()
except Exception as error:
for needle in needles:
if needle not in str(error):
fail(f"{what}: error lacks {needle!r}: {error}")
return
fail(f"{what}: no error raised")
def info_of(pb, base, page, patch, flash, signature=(0x1E, 0x93, 0x0B), word_flash=False, version=None):
scale = 2 if word_flash else 1
wire_base = base // scale
flags = (1 if patch else 0) | (2 if word_flash else 0)
# The EEPROM size comes from the signature, as it must: pureboot 5 derives
# the whole geometry from the signature rather than sending it, so a
# synthetic block that disagreed with its own signature would describe a
# chip that cannot exist.
eeprom = pb.CHIP_GEOMETRY[signature][2]
raw = bytes((0x50, 0x42, pb.NEWEST_LOADER if version is None else version,
*signature, page & 0xFF, wire_base & 0xFF, wire_base >> 8,
eeprom & 0xFF, eeprom >> 8, flags))
info = pb.Info(raw)
if info.flash_size != flash:
fail(f"info_of({base:#x}) decodes to {info.flash_size:#x} of flash, not {flash:#x}")
return info
def rjmp_decode(word, at, flash_words):
if word & 0xF000 != 0xC000:
fail(f"not an rjmp: {word:#06x}")
offset = word & 0x0FFF
if offset >= 0x800:
offset -= 0x1000
return (at + 1 + offset) % flash_words
def main():
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(sys.argv[1])))
import pureboot as pb
tiny = info_of(pb, 0x1E00, 64, True, 0x2000)
mega = info_of(pb, 0x7E00, 128, False, 0x8000, signature=(0x1E, 0x95, 0x0F))
# Versioning: the block's third byte is the loader's version, and the tool
# speaks a window of them. Every version in the window decodes, so an older
# deployed loader stays usable; one above the window is refused by name,
# since which version changed the protocol is knowledge only the tool
# holds, and it holds none about a version it has never heard of.
for version in range(pb.OLDEST_LOADER, pb.NEWEST_LOADER + 1):
if info_of(pb, 0x1E00, 64, True, 0x2000, version=version).version != version:
fail(f"pureboot {version} does not decode")
expect_error(
"unknown loader version",
lambda: info_of(pb, 0x1E00, 64, True, 0x2000, version=pb.NEWEST_LOADER + 1),
f"pureboot {pb.NEWEST_LOADER + 1}",
"newer tool",
)
# mega_boot: BOOTSZ words and the BOOTRST sense per chip - the fuse byte
# index (EXTENDED on the x8 line except the m328s' HIGH, HIGH elsewhere)
# and the per-family ladders (Atmel-2486/2466/2503/2545/8271/DS40002065/
# 8272/8011/2593/42719). Synthetic 'F' replies: only the boot byte
# carries meaning.
cases = (
((0x1E, 0x93, 0x07), 0x2000, 3, {0b11: 0x1F00, 0b10: 0x1E00, 0b01: 0x1C00, 0b00: 0x1800}), # m8
((0x1E, 0x94, 0x03), 0x4000, 3, {0b11: 0x3F00, 0b10: 0x3E00, 0b01: 0x3C00, 0b00: 0x3800}), # m16
((0x1E, 0x95, 0x02), 0x8000, 3, {0b11: 0x7E00, 0b10: 0x7C00, 0b01: 0x7800, 0b00: 0x7000}), # m32
((0x1E, 0x93, 0x0A), 0x2000, 2, {0b11: 0x1F00, 0b10: 0x1E00, 0b01: 0x1C00, 0b00: 0x1800}), # m88
((0x1E, 0x93, 0x0F), 0x2000, 2, {0b11: 0x1F00, 0b10: 0x1E00, 0b01: 0x1C00, 0b00: 0x1800}), # m88P
((0x1E, 0x94, 0x06), 0x4000, 2, {0b11: 0x3F00, 0b10: 0x3E00, 0b01: 0x3C00, 0b00: 0x3800}), # m168/168A
((0x1E, 0x94, 0x0B), 0x4000, 2, {0b11: 0x3F00, 0b10: 0x3E00, 0b01: 0x3C00, 0b00: 0x3800}), # m168P
((0x1E, 0x95, 0x14), 0x8000, 3, {0b11: 0x7E00, 0b10: 0x7C00, 0b01: 0x7800, 0b00: 0x7000}), # m328
((0x1E, 0x95, 0x0F), 0x8000, 3, {0b11: 0x7E00, 0b10: 0x7C00, 0b01: 0x7800, 0b00: 0x7000}), # m328P
((0x1E, 0x94, 0x0F), 0x4000, 3, {0b11: 0x3F00, 0b10: 0x3E00, 0b01: 0x3C00, 0b00: 0x3800}), # m164A
((0x1E, 0x94, 0x0A), 0x4000, 3, {0b11: 0x3F00, 0b10: 0x3E00, 0b01: 0x3C00, 0b00: 0x3800}), # m164P
((0x1E, 0x95, 0x15), 0x8000, 3, {0b11: 0x7E00, 0b10: 0x7C00, 0b01: 0x7800, 0b00: 0x7000}), # m324A
((0x1E, 0x96, 0x09), 0x10000, 3, {0b11: 0xFC00, 0b10: 0xF800, 0b01: 0xF000, 0b00: 0xE000}), # m644
((0x1E, 0x96, 0x0A), 0x10000, 3, {0b11: 0xFC00, 0b10: 0xF800, 0b01: 0xF000, 0b00: 0xE000}), # m644P
((0x1E, 0x97, 0x06), 0x20000, 3, {0b11: 0x1FC00, 0b10: 0x1F800, 0b01: 0x1F000, 0b00: 0x1E000}), # 1284
((0x1E, 0x97, 0x05), 0x20000, 3, {0b11: 0x1FC00, 0b10: 0x1F800, 0b01: 0x1F000, 0b00: 0x1E000}), # 1284P
)
for signature, flash, which, ladder in cases:
chip = info_of(pb, flash - pb.SLOT, 128 if flash < 0x20000 else 0, False, flash,
signature=signature, word_flash=flash > 0x10000)
for bits, start in ladder.items():
fuses = bytearray((0xFF, 0xFF, 0xFF, 0xFF))
fuses[which] = (0xF8 | (bits << 1)) & ~1
prog, at = pb.mega_boot(chip, bytes(fuses))
if not prog or at != start:
fail(f"mega_boot {signature[1]:02x}{signature[2]:02x} BOOTSZ={bits:02b} programmed: {prog} {at:#07x}")
fuses[which] |= 1
prog, at = pb.mega_boot(chip, bytes(fuses))
if prog or at != start:
fail(f"mega_boot {signature[1]:02x}{signature[2]:02b} unprogrammed: {prog} {at:#07x}")
# Word-addressed info decode: the 1284P's base and page ride the wire
# scaled - a 17-bit base halved into the block's two bytes, a 256-byte page
# spelled 0 - and its slot is the same 512 bytes as everywhere else, so its
# staging slot lands inside the 1 KiB minimum boot section.
big = info_of(pb, 0x1FE00, 0, False, 0x20000, signature=(0x1E, 0x97, 0x05), word_flash=True)
if big.page != 256 or big.base != 0x1FE00 or big.stage != 0x1FC00:
fail(f"word-addressed info decode: page {big.page}, base {big.base:#x}, stage {big.stage:#x}")
# Surgery: word 0 lands on the loader, the trampoline on the original
# entry - checked with an independent decoder.
app = bytes((0xC0 | 0x00, 0xC0)) + bytes((0x12,)) * 300 # rjmp .+0x00C0... entry word 0xC0C0
entry = rjmp_decode(app[0] | (app[1] << 8), 0, tiny.flash_size // 2)
pages = pb.plan_flash(app, tiny)
word0 = pages[0][0] | (pages[0][1] << 8)
if rjmp_decode(word0, 0, tiny.flash_size // 2) != tiny.base // 2:
fail("surgery: patched word 0 misses the loader")
tp = pages[tiny.base - 64]
tramp = tp[62] | (tp[63] << 8)
if rjmp_decode(tramp, (tiny.base - 2) // 2, tiny.flash_size // 2) != entry:
fail("surgery: trampoline misses the original entry")
expect_error("non-rjmp vector", lambda: pb.plan_flash(bytes((0x0C, 0x94)) + app[2:], tiny), "not an rjmp")
looped = bytearray(app)
word = pb.rjmp_to(0, tiny.base // 2, tiny.flash_size // 2)
looped[0], looped[1] = word & 0xFF, word >> 8
expect_error("read-back image", lambda: pb.plan_flash(bytes(looped), tiny), "read-back")
expect_error("oversize image", lambda: pb.plan_flash(bytes(0x1DFF), tiny), "application flash ends")
# Ordering: patched vector puts page 0 first and the trampoline second;
# a boot section puts page 0 last. Blank pages drop only when erased.
order = pb.covered(pages, tiny, skip_blank=False)
if order[0] != 0 or order[1] != tiny.base - 64:
fail(f"tiny order starts {order[:2]}, want page 0 then trampoline page")
if sorted(order[2:]) != order[2:]:
fail("tiny order tail not ascending")
mega_pages = pb.plan_flash(bytes((0xFF,)) * 600, mega)
morder = pb.covered(mega_pages, mega, skip_blank=False)
if morder[-1] != 0 or sorted(morder[:-1]) != morder[:-1]:
fail(f"mega order {morder}, want ascending with page 0 last")
blanky = {0: pages[0], 64: bytes((0xFF,)) * 64, 128: pages[128], tiny.base - 64: tp}
slim = pb.covered(blanky, tiny, skip_blank=True)
if 64 in slim or 0 not in slim or tiny.base - 64 not in slim:
fail(f"skip_blank order wrong: {slim}")
# Staging content: the identical image plus the through-word on a
# patched-vector chip; hard size clamps either way.
image = bytes(range(256)) * 2 # 512 B - too big for a tiny slot
expect_error("tiny staging size", lambda: pb.staging_content(image, tiny), "510")
staged = pb.staging_content(image[:508], tiny)
through = staged[510] | (staged[511] << 8)
if rjmp_decode(through, (tiny.base - 2) // 2, tiny.flash_size // 2) != tiny.base // 2:
fail("through-word misses the resident base")
if pb.staging_content(image, mega) != image:
fail("mega staging content should be the bare image")
expect_error("mega staging size", lambda: pb.staging_content(image + b"!", mega), "512")
# The image stamp: found in a synthetic binary, absent in noise. pureboot
# 5 stamps the magic, its version and the signature, and the geometry is
# looked up from there - so what comes back must equal what a live device
# of the same chip reports.
stamp = bytes((0x50, 0x42, pb.NEWEST_LOADER)) + bytes(tiny.signature)
binary = bytes((0xAA,)) * 10 + stamp + bytes((0xBB,)) * 10
found = pb.image_info(binary)
if found is None or found.raw != tiny.raw:
fail(f"image_info misreads the v{pb.NEWEST_LOADER} stamp: "
f"{found.raw.hex() if found else None} != {tiny.raw.hex()}")
if pb.image_info(bytes((0xAA,)) * 40) is not None:
fail("image_info invents a block")
# An older loader's image stays readable, so a deployed build can be
# identified and installed like any other.
old = info_of(pb, 0x1E00, 64, True, 0x2000, version=pb.OLDEST_LOADER)
found_old = pb.image_info(bytes((0xAA,)) * 10 + old.raw)
if found_old is None or found_old.version != pb.OLDEST_LOADER:
fail("image_info misses an older loader's block")
# loader_image must peel a padded image down to the slot content: a raw
# .bin padded from address 0 (or a whole-flash read-back with the loader
# resident at base) yields the same bytes as the bare slot image.
import tempfile
slot_image = bytes((0xAA,)) * 10 + tiny.raw + bytes((0xCC,)) * 40
padded = bytes((0xFF,)) * tiny.base + slot_image
with tempfile.NamedTemporaryFile(suffix=".bin", delete=False) as f:
f.write(padded)
padded_path = f.name
try:
if pb.loader_image(padded_path) != slot_image:
fail("loader_image does not peel a padded image to the slot content")
finally:
os.unlink(padded_path)
# Update preflight: the full fuse matrix, plus target mismatch.
other = info_of(pb, 0x1E00, 32, True, 0x2000)
expect_error("wrong-target image", lambda: pb.update_preflight(binary, other, None), "another target")
expect_error("mega needs fuses", lambda: pb.update_preflight(bytes((0xAA,)) * 8 + mega.raw, mega, None),
"--assume-fuses")
mega_image = bytes((0xAA,)) * 8 + mega.raw
def fuses(high):
return bytes((0xFF, 0xFF, 0xFF, high))
expect_error("BOOTSZ 512 B", lambda: pb.update_preflight(mega_image, mega, fuses(0xFE)),
"cannot self-update", "BOOTSZ")
notes = pb.update_preflight(mega_image, mega, fuses(0xFD)) # 1 KB, BOOTRST unprogrammed
if not any("BOOTRST unprogrammed" in n for n in notes):
fail(f"1K/unprogrammed notes: {notes}")
notes = pb.update_preflight(mega_image, mega, fuses(0xFC)) # 1 KB, BOOTRST programmed
if not any("staging slot" in n for n in notes):
fail(f"1K/programmed notes: {notes}")
notes = pb.update_preflight(mega_image, mega, fuses(0xFA)) # 2 KB, BOOTRST programmed
if not any("application flash" in n for n in notes):
fail(f"2K/programmed notes: {notes}")
if pb.update_preflight(bytes((0xAA,)) * 8 + tiny.raw, tiny, None) != []:
fail("tiny preflight should pass without fuses")
# The 1284s' smallest boot section (512 words) is exactly the resident
# slot plus its staging slot, so self-update is possible at the minimum
# BOOTSZ - no fuse step up, the 644's geometry. That holds only while a
# slot is 512 B: at 1 KiB the staging slot would fall outside the section
# and the preflight would refuse.
notes = pb.update_preflight(bytes((0xAA,)) * 8 + big.raw, big, fuses(0xFE))
if not any("staging slot" in n for n in notes):
fail(f"1284 minimum-BOOTSZ notes: {notes}")
# The walk-region refusal: BOOTRST aimed below the loader plus app data
# in the walk span errors without --force; erased spans and unprogrammed
# BOOTRST pass.
deep = {0x7800: bytes((1,)) * 128}
expect_error("walk region", lambda: pb.check_walk_region(deep, mega, fuses(0xFA), False), "--force")
pb.check_walk_region(deep, mega, fuses(0xFA), True)
pb.check_walk_region(deep, mega, fuses(0xFB), False) # BOOTRST unprogrammed
pb.check_walk_region({0x7800: bytes((0xFF,)) * 128}, mega, fuses(0xFA), False)
# Fuses that could not be read leave the span unknown, and unknown is not
# empty: the check that did not run must refuse rather than pass, since the
# span it would have named is the one that costs the part.
expect_error("walk region without fuses",
lambda: pb.check_walk_region(deep, mega, None, False), "--assume-fuses", "--force")
pb.check_walk_region(deep, mega, None, True)
pb.check_walk_region(deep, tiny, None, False) # patched reset vector: no walk to guard
# The repairing verify: a mismatched page is rewritten rather than raised,
# bounded so a fault that is not self-clearing cannot spin.
class FakeLoader:
"""A device whose first `bad` writes of any page land wrong."""
def __init__(self, info, bad):
self.info = info
self.bad = bad
self.flash = {}
self.writes = 0
def write_page(self, address, data):
self.writes += 1
self.flash[address] = bytes(len(data)) if self.bad > 0 else bytes(data)
self.bad -= 1
def read_flash(self, address, count):
return self.flash.get(address, bytes(count))
want = {0: bytes((i * 5) & 0xFF for i in range(128))}
# One bad write, then good: repaired in place, and the caller never sees
# an error. The rewrite is counted, so a silent no-op cannot pass.
device = FakeLoader(info_of(pb, 0x7E00, 128, False, 0x8000), bad=1)
device.write_page(0, want[0])
pb.verify_pages(device, want, repair=True)
if device.writes != 2:
fail(f"repairing verify made {device.writes} writes, expected 2")
# Without repair the same state raises, so the repair is what fixed it.
device = FakeLoader(info_of(pb, 0x7E00, 128, False, 0x8000), bad=1)
device.write_page(0, want[0])
expect_error("verify without repair", lambda: pb.verify_pages(device, want), "verify failed")
# A page that never comes good stops after RETRIES rewrites, and says so.
device = FakeLoader(info_of(pb, 0x7E00, 128, False, 0x8000), bad=99)
device.write_page(0, want[0])
expect_error(
"unrepairable page",
lambda: pb.verify_pages(device, want, repair=True),
"verify failed",
f"after {pb.RETRIES} retries",
)
if device.writes != pb.RETRIES + 1:
fail(f"unrepairable page took {device.writes} writes, expected {pb.RETRIES + 1}")
# The knock handshake against a device that is not listening yet - the
# state a port open leaves behind: it resets the chip into a fresh
# activation window while the previous session's prompt is still in
# flight, so the first knock is lost and a prompt arrives anyway.
class FakePort:
"""A loader in its activation window, plus `lost` leading writes the
reset swallows and one stale prompt still on the wire."""
def __init__(self, info_raw, lost=0, stale=b"", active=False):
self.info_raw = info_raw
self.lost = lost
self.inflight = bytearray(stale)
self.rx = bytearray()
self.active = active
self.last = None
def flush_input(self):
self.rx.clear()
def write(self, data):
if self.lost:
self.lost -= 1
return
for byte in bytes(data):
if not self.active:
self.active = self.last == ord("p") and byte == ord("b")
self.last = byte
if self.active:
self.rx += pb.PROMPT
elif byte == ord("b"):
self.rx += self.info_raw + pb.PROMPT
else:
self.rx += pb.PROMPT
def read_available(self, wait):
self.rx = self.inflight + self.rx # the stale prompt lands late
self.inflight.clear()
out, self.rx = bytes(self.rx), bytearray()
return out
def read_exact(self, count, timeout):
if len(self.rx) < count:
raise pb.Error(f"timeout: got {len(self.rx)} of {count} bytes")
out, self.rx = bytes(self.rx[:count]), self.rx[count:]
return out
raw = info_of(pb, 0x7E00, 128, False, 0x8000).raw
for what, port in (
("clean window", FakePort(raw)),
("stale prompt over a lost knock", FakePort(raw, lost=1, stale=pb.PROMPT)),
("live session", FakePort(raw, active=True)),
):
info = pb.Loader(port).connect(5)
if info.raw != raw:
fail(f"connect ({what}) returned {info.raw.hex()}")
# A device that never answers still says so, and a version the tool cannot
# speak is reported as such rather than retried into a timeout.
expect_error("dead device", lambda: pb.Loader(FakePort(raw, lost=99)).connect(0), "no answer")
old = bytes(raw[:2]) + bytes((pb.NEWEST_LOADER + 1,)) + bytes(raw[3:])
expect_error("unspeakable version", lambda: pb.Loader(FakePort(old)).connect(5), "needs a newer tool")
print("test_planner: all planner and policy checks pass")
if __name__ == "__main__":
main()

71
test/test_scan.py
View File

@@ -1,71 +0,0 @@
#!/usr/bin/env python3
"""--scan's walk and report logic, no simulator: the probe order, the rate
arithmetic, and the advice's direction. The rate physics itself is not
sim-testable - a pty carries bytes at any termios rate - so what the wire
would arbitrate is pinned here as logic instead.
Usage: test_scan.py <tool_py>
"""
import os
import sys
def fail(message):
print(f"FAIL: {message}")
sys.exit(1)
def main():
sys.path.insert(0, os.path.dirname(os.path.abspath(sys.argv[1])))
import pureboot as pb
walk = pb.scan_ratios()
if walk != [0, -2, 2, -4, 4, -6, 6, -8, 8, -10, 10]:
fail(f"probe walk is not built-rate-first, nearest-out: {walk}")
if pb.scan_rate(9600, 4) != 9984 or pb.scan_rate(9600, -4) != 9216:
fail("probe rate arithmetic")
if pb.scan_rate(115200, 0) != 115200:
fail("the built rate must probe unchanged")
# A loader answering fast means a fast oscillator: the trim goes down.
report = "\n".join(pb.scan_report(9600, 4, 6))
for needle in ("9984", "+4 %", "--baud 9984", "4 steps lower", "pureboot 6"):
if needle not in report:
fail(f"+4 % report lacks {needle!r}:\n{report}")
report = "\n".join(pb.scan_report(9600, -6, 6))
if "6 steps higher" not in report:
fail(f"-6 % report advises the wrong direction:\n{report}")
report = "\n".join(pb.scan_report(9600, 0, 6))
if "none" not in report or "steps" in report:
fail(f"an on-rate answer must advise no trim:\n{report}")
report = "\n".join(pb.scan_report(9600, 4, 6, clock=9600000))
if "9984000" not in report:
fail(f"the absolute clock must scale with the found ratio:\n{report}")
# The walk's rates mostly have no termios B-constant, so the POSIX port
# must set them through termios2 - probed on a pty, which accepts the
# ioctl without caring about the speed. Without this every off-nominal
# probe would abort the walk on the platform --scan matters most on.
if os.name == "posix":
import pty
master, slave = pty.openpty()
try:
port = pb.Port(os.ttyname(slave), pb.scan_rate(9600, 4))
port.set_baud(pb.scan_rate(9600, -4))
port.close()
except pb.Error as error:
fail(f"PosixPort refused an off-nominal probe rate: {error}")
finally:
os.close(master)
os.close(slave)
print("OK")
if __name__ == "__main__":
main()

167
test/test_update_link.py
View File

@@ -1,167 +0,0 @@
#!/usr/bin/env python3
"""Self-update across a link change: the host must follow the staging copy.
`--update-loader` installs the new image in the staging slot and then *enters
it* to have it rewrite the resident. That copy is the new image, so it speaks the
new image's baud and backend - but the host was talking to the *resident*. Where
the two differ, the host kept knocking at the old rate in the old mode, the
staging copy never answered, and the update stranded: staging installed, resident
untouched, and on a 1 KiB tiny the application region (which *is* the staging
slot there) already gone.
The wire cannot be probed for this - 512 bytes of position-independent code carry
no header saying what rate they were built for - so the operator declares it, and
a mismatch with nothing declared has to say so instead of reporting a bare
timeout.
Stdlib only, no device: host-tool logic, so it runs on every chip's preset beside
pureboot.planner.
"""
import importlib.util
import pathlib
PB = pathlib.Path(__file__).resolve().parents[1] / "pureboot" / "pureboot.py"
_spec = importlib.util.spec_from_file_location("pureboot", PB)
pb = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(pb)
IDENTITY = b"\x05\x1e\x95\x0f" # pureboot 5 + m328p signature
P = F = 0
def check(name, ok, detail=""):
global P, F
P, F = P + (1 if ok else 0), F + (0 if ok else 1)
print(f" [{'PASS' if ok else 'FAIL'}] {name}" + (f" - {detail}" if detail else ""))
class TwoLinkPort:
"""A board whose resident and staging copy answer on different links.
Only the rate currently set decides who can be heard, which is the physical
truth: a loader's bit timing is a cycle count, so a copy built for another
rate is unreadable until the host retunes. The knock bytes carry the mode, so
a backend mismatch is caught the same way.
"""
def __init__(self, resident=(57600, False), staged=(38400, False)):
self.resident, self.staged = resident, staged
self.baud = resident[0]
self.entered = False # a 'J' has handed control to the staging copy
self.switches = [] # every retune the host asked for
self.pending = bytearray() # what the device has queued to send
# --- the part under test needs this to exist at all
def set_baud(self, baud):
self.baud = baud
self.switches.append(baud)
def flush_input(self):
self.pending.clear()
def _audible(self, knock=None):
baud, autobaud = self.staged if self.entered else self.resident
if self.baud != baud:
return False
if knock is None:
return True
return knock == (bytes((pb.CALIBRATE, ord("p"))) if autobaud else b"pb")
def write(self, data):
data = bytes(data)
if data[:1] == b"J" and len(data) == 3:
# The resident acks the jump, then control moves to the copy.
if self._audible():
self.pending += pb.PROMPT
self.entered = True
elif data in (b"pb", bytes((pb.CALIBRATE, ord("p")))):
if self._audible(data):
self.pending += pb.PROMPT
elif data == b"b":
if self._audible():
self.pending += IDENTITY + pb.PROMPT
def read_available(self, wait):
out, self.pending = bytes(self.pending), bytearray()
return out
def read_exact(self, count, timeout):
if len(self.pending) < count:
raise pb.Error(f"timeout: got {len(self.pending)} of {count} bytes")
out, self.pending = bytes(self.pending[:count]), self.pending[count:]
return out
def connected(port):
"""A Loader already in session with the resident."""
loader = pb.Loader(port)
loader.connect(2.0)
return loader
def main():
# The control first: where the staged image keeps the resident's link, the
# flow works and needs no retune. This is the case that always passed, and
# it is what made the bug look like "self-update is broken" rather than
# "self-update cannot change the link".
port = TwoLinkPort(resident=(57600, False), staged=(57600, False))
loader = connected(port)
try:
loader.enter_copy(0x7C00, 2.0)
check("same link: staging copy entered", True)
except pb.Error as error:
check("same link: staging copy entered", False, str(error))
# A baud change, declared. The host must retune before knocking.
port = TwoLinkPort(resident=(57600, False), staged=(38400, False))
loader = connected(port)
try:
loader.enter_copy(0x7C00, 2.0, link=(38400, False))
check("baud change declared: entered after retuning", 38400 in port.switches,
f"switches={port.switches}")
except (pb.Error, TypeError) as error:
check("baud change declared: entered after retuning", False, repr(error))
# A backend change, declared: the knock itself has to become the calibration
# pulse, or an autobaud staging copy never hears a thing.
port = TwoLinkPort(resident=(57600, False), staged=(57600, True))
loader = connected(port)
try:
loader.enter_copy(0x7C00, 2.0, link=(57600, True))
check("backend change declared: entered as autobaud", True)
except (pb.Error, TypeError) as error:
check("backend change declared: entered as autobaud", False, repr(error))
# Nothing declared against a changed link: it still cannot work, but the
# error has to name the cause. A bare "no answer" sent the operator looking
# at the wiring while the application region sat erased.
port = TwoLinkPort(resident=(57600, False), staged=(38400, False))
loader = connected(port)
try:
loader.enter_copy(0x7C00, 0.3)
check("undeclared mismatch: reported", False, "unexpectedly succeeded")
except pb.Error as error:
text = str(error).lower()
check("undeclared mismatch: error names the link, not just a timeout",
"link" in text or "baud" in text or "backend" in text, str(error))
except TypeError as error:
check("undeclared mismatch: error names the link, not just a timeout",
False, repr(error))
# The resident's own link must be restored for the caller: a declared
# staging link is for the copy, and the tool talks to the new resident after.
port = TwoLinkPort(resident=(57600, False), staged=(38400, False))
loader = connected(port)
try:
loader.enter_copy(0x7C00, 2.0, link=(38400, False))
check("session records the link it is now speaking", loader.baud == 38400,
f"loader.baud={getattr(loader, 'baud', None)}")
except (pb.Error, TypeError, AttributeError) as error:
check("session records the link it is now speaking", False, repr(error))
print(f"\n {P} passed, {F} failed")
return 1 if F else 0
if __name__ == "__main__":
raise SystemExit(main())