tsb: reimplement TinySafeBoot on libavr in three size tiers
The native-UART fixed-baud TinySafeBoot protocol, ported onto libavr as a
crt-free boot-section loader, in three variants that trade clarity for size:
tsb_pure 740 B idiomatic C++: SRAM page buffer, separate flash/EEPROM
leaves, shared framing; the polled `unused` guard posture.
tsb_tricks 658 B unified runtime-flag paths (noinline/noclone), call-saved
global-register page walk — attributes only, no asm.
tsb_asm 508 B streaming store + hand-rolled UART/SPM/EEPROM/erase loops;
fits the 512 B boot section (BOOTSZ=11). Trims the optional
password gate and WDT-reset bail — unreachable in C++ with
both (hand-asm is ~15 % denser). Tiers 1-2 keep them and
live in the 1 KB section they fit.
All three are .text byte-identical across libavr's generated and reflect modes.
The CMake build strips the leaked -O3 (a Release build is silently -O3, not the
-Os this loader is measured against) and gates each variant's size against its
section. A simavr harness (test/device.c + test/tsbtest.py) drives the real wire
protocol over a pty and flashes the device; the size and protocol tests run in
ctest. Verified byte-for-byte against the reference tsbloader_adv (C#/mono):
activate, read info, flash write + verify.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
11
test/check_size.cmake
Normal file
11
test/check_size.cmake
Normal file
@@ -0,0 +1,11 @@
|
||||
execute_process(COMMAND ${SIZE_TOOL} ${ELF} OUTPUT_VARIABLE _out RESULT_VARIABLE _res)
|
||||
if(NOT _res EQUAL 0)
|
||||
message(FATAL_ERROR "avr-size failed")
|
||||
endif()
|
||||
# avr-size line 2 is "<text> <data> <bss> <dec> <hex> <file>".
|
||||
string(REGEX MATCH "\n[ \t]*([0-9]+)" _m "${_out}")
|
||||
set(_text ${CMAKE_MATCH_1})
|
||||
if(_text GREATER LIMIT)
|
||||
message(FATAL_ERROR ".text is ${_text} bytes, over the ${LIMIT}-byte boot section")
|
||||
endif()
|
||||
message(STATUS ".text ${_text} <= ${LIMIT} (boot section budget)")
|
||||
86
test/device.c
Normal file
86
test/device.c
Normal file
@@ -0,0 +1,86 @@
|
||||
// simavr "device" for the TSB bootloader: load the boot-linked ELF into the
|
||||
// ATmega328P boot section, enter it (BOOTRST is not modelled, so we set PC to
|
||||
// the boot base, exactly as simavr's own board_simduino does), and expose
|
||||
// UART0 as a pty. A host client (Python pyserial, or the real tsbloader) then
|
||||
// speaks the TSB protocol over that pty and actually flashes the device.
|
||||
//
|
||||
// SPM genuinely writes avr->flash on the mega cores, so on exit (or SIGTERM)
|
||||
// we dump the flash image to a file for a ground-truth cross-check against
|
||||
// what the client read back through the bootloader.
|
||||
#include <signal.h>
|
||||
#include <stdint.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include "sim_avr.h"
|
||||
#include "sim_elf.h"
|
||||
#include "uart_pty.h"
|
||||
|
||||
static avr_t *avr;
|
||||
static uart_pty_t uart_pty;
|
||||
static const char *dump_path;
|
||||
|
||||
static void finish(int sig)
|
||||
{
|
||||
(void)sig;
|
||||
if (dump_path) {
|
||||
FILE *f = fopen(dump_path, "wb");
|
||||
if (f) {
|
||||
fwrite(avr->flash, 1, avr->flashend + 1, f);
|
||||
fclose(f);
|
||||
}
|
||||
}
|
||||
uart_pty_stop(&uart_pty);
|
||||
_exit(0);
|
||||
}
|
||||
|
||||
int main(int argc, char *argv[])
|
||||
{
|
||||
if (argc < 3) {
|
||||
fprintf(stderr, "usage: %s <tsb.elf> <boot_base_hex> [flash_dump.bin]\n", argv[0]);
|
||||
return 2;
|
||||
}
|
||||
uint32_t boot_base = (uint32_t)strtoul(argv[2], NULL, 0);
|
||||
dump_path = argc >= 4 ? argv[3] : NULL;
|
||||
|
||||
avr = avr_make_mcu_by_name("atmega328p");
|
||||
if (!avr) {
|
||||
fprintf(stderr, "device: no ATmega328P core\n");
|
||||
return 1;
|
||||
}
|
||||
avr_init(avr);
|
||||
avr->frequency = 16000000;
|
||||
// Real flash powers up erased (0xff); the app region must look erased
|
||||
// before the bootloader programs it.
|
||||
memset(avr->flash, 0xff, avr->flashend + 1);
|
||||
|
||||
// simavr's ELF loader flattens the flash base to 0 (it expects an app at
|
||||
// 0x0), but it hands back the boot code in fw.flash; place it at the boot
|
||||
// section base ourselves and enter there (BOOTRST is not modelled).
|
||||
elf_firmware_t fw = {0};
|
||||
if (elf_read_firmware(argv[1], &fw) != 0) {
|
||||
fprintf(stderr, "device: cannot read %s\n", argv[1]);
|
||||
return 1;
|
||||
}
|
||||
memcpy(avr->flash + boot_base, fw.flash, fw.flashsize);
|
||||
avr->pc = boot_base;
|
||||
avr->codeend = avr->flashend;
|
||||
|
||||
uart_pty_init(avr, &uart_pty);
|
||||
uart_pty_connect(&uart_pty, '0');
|
||||
printf("TSB_PTY %s\n", uart_pty.pty.slavename);
|
||||
fflush(stdout);
|
||||
|
||||
signal(SIGTERM, finish);
|
||||
signal(SIGINT, finish);
|
||||
|
||||
for (;;) {
|
||||
int state = avr_run(avr);
|
||||
if (state == cpu_Done || state == cpu_Crashed)
|
||||
break;
|
||||
}
|
||||
finish(0);
|
||||
return 0;
|
||||
}
|
||||
195
test/tsbtest.py
Normal file
195
test/tsbtest.py
Normal file
@@ -0,0 +1,195 @@
|
||||
#!/usr/bin/env python3
|
||||
"""End-to-end TSB protocol test: spawn the simavr device, speak the TinySafeBoot
|
||||
wire protocol over its pty (as the real host tools do), and actually flash it.
|
||||
|
||||
Usage: tsbtest.py <device_binary> <tsb.elf> <boot_base_hex>
|
||||
Exits 0 if every scenario passes.
|
||||
"""
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
import serial
|
||||
|
||||
CONFIRM = 0x21 # '!'
|
||||
REQUEST = 0x3F # '?'
|
||||
PAGE = 128 # ATmega328P: 64 words
|
||||
|
||||
|
||||
class Device:
|
||||
"""The simavr runner, exposing UART0 as a pty."""
|
||||
|
||||
def __init__(self, binary, elf, boot_base, dump="/tmp/tsb_dump.bin"):
|
||||
self.proc = subprocess.Popen(
|
||||
[binary, elf, boot_base, dump],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
|
||||
self.dump = dump
|
||||
self.pty = None
|
||||
deadline = time.time() + 5
|
||||
while time.time() < deadline:
|
||||
line = self.proc.stdout.readline()
|
||||
if not line:
|
||||
break
|
||||
if line.startswith("TSB_PTY"):
|
||||
self.pty = line.split()[1]
|
||||
break
|
||||
if not self.pty:
|
||||
self.stop()
|
||||
raise RuntimeError("device did not report a pty")
|
||||
|
||||
def stop(self):
|
||||
self.proc.terminate()
|
||||
try:
|
||||
self.proc.wait(timeout=3)
|
||||
except subprocess.TimeoutExpired:
|
||||
self.proc.kill()
|
||||
|
||||
|
||||
class Host:
|
||||
"""A faithful TSB host, per the wire protocol."""
|
||||
|
||||
def __init__(self, pty):
|
||||
self.s = serial.Serial(pty, 115200, timeout=1.5)
|
||||
self.info = None
|
||||
|
||||
def _read(self, n):
|
||||
data = self.s.read(n)
|
||||
if len(data) != n:
|
||||
raise AssertionError(f"expected {n} bytes, got {len(data)}: {data.hex()}")
|
||||
return data
|
||||
|
||||
def activate(self):
|
||||
self.s.reset_input_buffer()
|
||||
self.s.write(b"@@@")
|
||||
reply = self._read(17)
|
||||
if reply[16] != CONFIRM:
|
||||
raise AssertionError(f"activation reply not '!'-terminated: {reply.hex()}")
|
||||
self.info = reply[:16]
|
||||
return self.info
|
||||
|
||||
# Parsed info-block fields (host math from the spec).
|
||||
@property
|
||||
def pagesize(self):
|
||||
return self.info[9] * 2
|
||||
|
||||
@property
|
||||
def appflash(self):
|
||||
return (self.info[10] | (self.info[11] << 8)) * 2
|
||||
|
||||
@property
|
||||
def eeprom_size(self):
|
||||
return (self.info[12] | (self.info[13] << 8)) + 1
|
||||
|
||||
def _expect(self, byte, what):
|
||||
r = self._read(1)
|
||||
if r[0] != byte:
|
||||
raise AssertionError(f"{what}: expected {byte:#x}, got {r.hex()}")
|
||||
|
||||
# Host-paced page read ('f'/'e'): send '!', take a page, repeat; stop with
|
||||
# anything else, then the Mainloop '!'.
|
||||
def _read_pages(self, cmd, npages):
|
||||
self.s.write(cmd.encode())
|
||||
data = b""
|
||||
for _ in range(npages):
|
||||
self.s.write(bytes([CONFIRM]))
|
||||
data += self._read(PAGE)
|
||||
self.s.write(bytes([REQUEST])) # stop
|
||||
self._expect(CONFIRM, f"{cmd} end")
|
||||
return data
|
||||
|
||||
# Device-paced page write ('F'/'E'): device offers '?', host sends '!'+page,
|
||||
# or anything else to stop.
|
||||
def _write_pages(self, cmd, data):
|
||||
if len(data) % PAGE:
|
||||
data += b"\xff" * (PAGE - len(data) % PAGE)
|
||||
self.s.write(cmd.encode())
|
||||
for off in range(0, len(data), PAGE):
|
||||
self._expect(REQUEST, f"{cmd} '?'")
|
||||
self.s.write(bytes([CONFIRM]) + data[off:off + PAGE])
|
||||
self._expect(REQUEST, f"{cmd} trailing '?'")
|
||||
self.s.write(bytes([REQUEST])) # stop
|
||||
self._expect(CONFIRM, f"{cmd} end")
|
||||
|
||||
def write_flash(self, data):
|
||||
self._write_pages("F", data)
|
||||
|
||||
def read_flash(self, npages):
|
||||
return self._read_pages("f", npages)
|
||||
|
||||
def write_eeprom(self, data):
|
||||
self._write_pages("E", data)
|
||||
|
||||
def read_eeprom(self, npages):
|
||||
return self._read_pages("e", npages)
|
||||
|
||||
def read_config(self):
|
||||
self.s.write(b"c")
|
||||
page = self._read(PAGE)
|
||||
self._expect(CONFIRM, "c end")
|
||||
return page
|
||||
|
||||
def write_config(self, data):
|
||||
assert len(data) == PAGE
|
||||
self.s.write(b"C")
|
||||
self._expect(REQUEST, "C '?'")
|
||||
self.s.write(bytes([CONFIRM]) + data)
|
||||
echo = self._read(PAGE) # device echoes what it programmed
|
||||
self._expect(CONFIRM, "C end")
|
||||
return echo
|
||||
|
||||
|
||||
def check(cond, msg):
|
||||
if not cond:
|
||||
raise AssertionError(msg)
|
||||
print(f" ok: {msg}")
|
||||
|
||||
|
||||
def main():
|
||||
binary, elf, boot_base = sys.argv[1], sys.argv[2], sys.argv[3]
|
||||
dev = Device(binary, elf, boot_base)
|
||||
failures = []
|
||||
try:
|
||||
host = Host(dev.pty)
|
||||
|
||||
# --- activation ---
|
||||
info = host.activate()
|
||||
check(info[0:3] == b"TSB", f"magic 'TSB' (got {info[0:3]!r})")
|
||||
check(info[6:9] == bytes([0x1E, 0x95, 0x0F]), f"signature 1E 95 0F (got {info[6:9].hex()})")
|
||||
check(info[14] == info[15], f"device-type bytes 14==15 (got {info[14]:#x},{info[15]:#x})")
|
||||
check(host.pagesize == PAGE, f"page size {PAGE} (got {host.pagesize})")
|
||||
check(host.eeprom_size == 1024, f"eeprom size 1024 (got {host.eeprom_size})")
|
||||
print(f" info: {info.hex()} appflash={host.appflash} eeprom={host.eeprom_size}")
|
||||
|
||||
# --- flash write / read round-trip (actual self-programming) ---
|
||||
app = bytes(range(256)) # two pages of known data
|
||||
host.write_flash(app)
|
||||
back = host.read_flash(2)
|
||||
check(back == app, f"flash round-trip 2 pages ({'match' if back == app else 'MISMATCH'})")
|
||||
|
||||
# --- EEPROM write / read round-trip ---
|
||||
edata = bytes((i * 7) & 0xFF for i in range(PAGE))
|
||||
host.write_eeprom(edata)
|
||||
eback = host.read_eeprom(1)
|
||||
check(eback == edata, "eeprom round-trip 1 page")
|
||||
|
||||
# --- config page write / read round-trip ---
|
||||
cfg = bytes([0x00, 0x00, 0x40]) + b"\xff" * (PAGE - 3) # appjump 0, timeout 0x40, no password
|
||||
echo = host.write_config(cfg)
|
||||
check(echo == cfg, "config write echoes the programmed page")
|
||||
check(host.read_config() == cfg, "config read-back matches")
|
||||
|
||||
except AssertionError as e:
|
||||
failures.append(str(e))
|
||||
print(f" FAIL: {e}")
|
||||
finally:
|
||||
dev.stop()
|
||||
|
||||
if failures:
|
||||
print(f"FAILED ({len(failures)})")
|
||||
return 1
|
||||
print("ALL PASS")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user