BlackMark
6f3eb06233
The info block's third byte was a protocol version that never moved in the loader's lifetime. It is the pureboot version now, and this change is version 2: the one number that says what a deployed loader is. Every loader already in the field answers 1. The protocol keeps no number of its own — a pureboot version implies it, and the host tool is what holds that map. pureboot.py states the loader-version window it speaks (OLDEST_LOADER/NEWEST_LOADER; a version that changes the protocol becomes the new floor there), so a loader newer than the tool is refused by name rather than decoded on the assumption nothing moved, while an older one is read, identified and installed like any other. The tool carries its own version, free to drift from the loader's: --version prints it and the window, --info leads with the device's, --update-loader names the version it installs. Tests: the planner unit pins the window — every version in it decodes, one above it is refused, an older loader's image is still found — and the live suite pins the built loader against the tool beside it, so a bump that reaches only one of them fails. The image is byte-identical to the previous build but for that byte. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1166 lines
50 KiB
Python
1166 lines
50 KiB
Python
#!/usr/bin/env python3
|
|
"""pureboot host tool — the smart half of the pureboot protocol (README.md).
|
|
|
|
The device exposes primitives; this tool composes them: image loading (raw
|
|
binary or Intel HEX), flash programming with read-back verification, erase as
|
|
writing 0xff, EEPROM programming, fuse and info readout, the hand-over jump,
|
|
and — on chips without a hardware boot section — the reset-vector surgery
|
|
that re-homes the application's entry through the trampoline word below the
|
|
loader. Page 0 and the trampoline are written first, so every interruption
|
|
point of a flash leaves the chip reset-recoverable into the loader.
|
|
|
|
It also updates the loader itself (--update-loader): pureboot's image is
|
|
position-independent, so the tool installs the identical binary one 512-byte
|
|
slot below the resident loader, jumps into that staging copy, lets it rewrite
|
|
the resident slot, and restores what the staging slot held — resumable at
|
|
every phase from the flash state plus a host-side state file carrying the
|
|
saved bytes.
|
|
|
|
Python standard library only; the serial port is driven with termios on POSIX
|
|
and the Win32 serial API (through ctypes) on Windows, so any tty or COM port
|
|
works — a USB adapter as well as a simavr pty.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
if os.name == "nt":
|
|
import ctypes
|
|
from ctypes import wintypes
|
|
else:
|
|
import select
|
|
import termios
|
|
|
|
PROMPT = b"+"
|
|
VERSION = 2 # this tool's own version — free to drift from a loader's
|
|
# The loader versions this tool speaks to. A pureboot version implies its wire
|
|
# protocol — the protocol carries no number of its own — so knowing which
|
|
# versions speak what is the tool's job, and this window is where it says so:
|
|
# every pureboot so far speaks this protocol, and a version that changes it
|
|
# becomes the new floor here.
|
|
OLDEST_LOADER = 1
|
|
NEWEST_LOADER = 2
|
|
SLOT = 512 # the loader slot on byte-addressed chips; word-addressed ones (>64 KiB) use 1 KiB — their own smallest boot sector
|
|
RETRIES = 3 # rewrites of a page that reads back wrong, before the run stops
|
|
|
|
VERBOSE = False
|
|
|
|
|
|
def verbose(message):
|
|
"""Detail printed only under --verbose: decisions and derived facts, not
|
|
per-byte chatter — the progress bar carries the bulk transfers."""
|
|
if VERBOSE:
|
|
print(f" {message}")
|
|
|
|
|
|
class Error(Exception):
|
|
pass
|
|
|
|
|
|
class Progress:
|
|
"""A transient in-place bar on stderr for the operations that take wire
|
|
time. Drawn only when stderr is a tty — logs, pipes and the test harness
|
|
see nothing — and erased once done; the summary line each operation
|
|
prints afterwards is the persistent record. A zero total (or no label)
|
|
disables it, so callers can pass one through unconditionally."""
|
|
|
|
def __init__(self, label, total, unit="pages"):
|
|
self.label, self.total, self.unit = label, total, unit
|
|
self.done = 0
|
|
self.width = 0
|
|
self.live = bool(label) and total > 0 and sys.stderr.isatty()
|
|
self._draw()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
if self.live:
|
|
sys.stderr.write("\r" + " " * self.width + "\r")
|
|
sys.stderr.flush()
|
|
|
|
def step(self, n=1):
|
|
self.done += n
|
|
self._draw()
|
|
|
|
def _draw(self):
|
|
if not self.live:
|
|
return
|
|
bar = 24 * self.done // self.total
|
|
line = (f"{self.label:<16} [{'#' * bar}{'-' * (24 - bar)}] "
|
|
f"{100 * self.done // self.total:3d}% {self.done}/{self.total} {self.unit}")
|
|
self.width = max(self.width, len(line))
|
|
sys.stderr.write("\r" + line)
|
|
sys.stderr.flush()
|
|
|
|
|
|
# ---------------------------------------------------------------- serial ---
|
|
|
|
|
|
class PosixPort:
|
|
"""A raw serial port with deadline-based reads, over termios."""
|
|
|
|
def __init__(self, path, baud):
|
|
self.fd = os.open(path, os.O_RDWR | os.O_NOCTTY)
|
|
attrs = termios.tcgetattr(self.fd)
|
|
attrs[0] = 0 # iflag
|
|
attrs[1] = 0 # oflag
|
|
attrs[2] = termios.CREAD | termios.CLOCAL | termios.CS8 # cflag
|
|
attrs[3] = 0 # lflag
|
|
try:
|
|
speed = getattr(termios, f"B{baud}")
|
|
except AttributeError:
|
|
raise Error(f"unsupported baud rate {baud}") from None
|
|
attrs[4] = attrs[5] = speed
|
|
attrs[6][termios.VMIN] = 0
|
|
attrs[6][termios.VTIME] = 0
|
|
termios.tcsetattr(self.fd, termios.TCSANOW, attrs)
|
|
|
|
def close(self):
|
|
os.close(self.fd)
|
|
|
|
def write(self, data):
|
|
os.write(self.fd, data)
|
|
|
|
def flush_input(self):
|
|
termios.tcflush(self.fd, termios.TCIFLUSH)
|
|
|
|
def read_available(self, wait):
|
|
"""Everything that arrives within `wait` seconds of quiet start."""
|
|
ready, _, _ = select.select([self.fd], [], [], wait)
|
|
return os.read(self.fd, 4096) if ready else b""
|
|
|
|
def read_exact(self, count, timeout):
|
|
data = b""
|
|
deadline = time.monotonic() + timeout
|
|
while len(data) < count:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise Error(f"timeout: got {len(data)} of {count} bytes")
|
|
ready, _, _ = select.select([self.fd], [], [], remaining)
|
|
if ready:
|
|
data += os.read(self.fd, count - len(data))
|
|
return data
|
|
|
|
|
|
if os.name == "nt":
|
|
# The same port, over the Win32 serial API — kernel32 through ctypes, so
|
|
# the tool stays standard-library only. Timeouts live in the driver
|
|
# (COMMTIMEOUTS) rather than in a readiness call: Windows has no select()
|
|
# for a COM handle, so each read asks the driver for its own deadline.
|
|
|
|
_GENERIC_READ, _GENERIC_WRITE = 0x80000000, 0x40000000
|
|
_OPEN_EXISTING, _PURGE_RXCLEAR = 3, 0x0008
|
|
_INVALID_HANDLE = wintypes.HANDLE(-1).value
|
|
# A gap this long ends a read_available(): longer than the coalescing a
|
|
# USB-serial adapter's latency timer imposes (16 ms on FTDI parts), so a
|
|
# burst is not split, short enough to stay responsive.
|
|
_GAP_MS = 30
|
|
|
|
class _DCB(ctypes.Structure):
|
|
_fields_ = [
|
|
("DCBlength", wintypes.DWORD),
|
|
("BaudRate", wintypes.DWORD),
|
|
("fBits", wintypes.DWORD), # the packed flag bitfield, set below
|
|
("wReserved", wintypes.WORD),
|
|
("XonLim", wintypes.WORD),
|
|
("XoffLim", wintypes.WORD),
|
|
("ByteSize", wintypes.BYTE),
|
|
("Parity", wintypes.BYTE),
|
|
("StopBits", wintypes.BYTE),
|
|
("XonChar", ctypes.c_char),
|
|
("XoffChar", ctypes.c_char),
|
|
("ErrorChar", ctypes.c_char),
|
|
("EofChar", ctypes.c_char),
|
|
("EvtChar", ctypes.c_char),
|
|
("wReserved1", wintypes.WORD),
|
|
]
|
|
|
|
class _COMMTIMEOUTS(ctypes.Structure):
|
|
_fields_ = [
|
|
("ReadIntervalTimeout", wintypes.DWORD),
|
|
("ReadTotalTimeoutMultiplier", wintypes.DWORD),
|
|
("ReadTotalTimeoutConstant", wintypes.DWORD),
|
|
("WriteTotalTimeoutMultiplier", wintypes.DWORD),
|
|
("WriteTotalTimeoutConstant", wintypes.DWORD),
|
|
]
|
|
|
|
_k32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
|
_LPDWORD = ctypes.POINTER(wintypes.DWORD)
|
|
# Declared, not inferred: a HANDLE is a pointer, and a defaulted int
|
|
# return would truncate it on 64-bit.
|
|
_k32.CreateFileW.restype = wintypes.HANDLE
|
|
_k32.CreateFileW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD,
|
|
wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE]
|
|
_k32.ReadFile.argtypes = [wintypes.HANDLE, wintypes.LPVOID, wintypes.DWORD, _LPDWORD, wintypes.LPVOID]
|
|
_k32.WriteFile.argtypes = [wintypes.HANDLE, wintypes.LPCVOID, wintypes.DWORD, _LPDWORD, wintypes.LPVOID]
|
|
_k32.GetCommState.argtypes = [wintypes.HANDLE, ctypes.POINTER(_DCB)]
|
|
_k32.SetCommState.argtypes = [wintypes.HANDLE, ctypes.POINTER(_DCB)]
|
|
_k32.SetCommTimeouts.argtypes = [wintypes.HANDLE, ctypes.POINTER(_COMMTIMEOUTS)]
|
|
_k32.PurgeComm.argtypes = [wintypes.HANDLE, wintypes.DWORD]
|
|
_k32.CloseHandle.argtypes = [wintypes.HANDLE]
|
|
|
|
def _fail(what):
|
|
code = ctypes.get_last_error()
|
|
raise Error(f"{what}: {ctypes.FormatError(code).strip()} (Windows error {code})")
|
|
|
|
class WindowsPort:
|
|
"""A raw serial port with deadline-based reads, over Win32."""
|
|
|
|
def __init__(self, path, baud):
|
|
# Win32 takes the rate as a plain integer, so unlike termios any
|
|
# rate the hardware can divide down to is available — but a driver
|
|
# may also accept one it cannot produce (an FT232R takes a baud of
|
|
# 3, reports it back, and goes on using the previous divisor).
|
|
# Only obvious nonsense is refusable; the rest is the driver's word.
|
|
if baud < 50:
|
|
raise Error(f"unsupported baud rate {baud}")
|
|
# \\.\COM6: the device-namespace form. A bare COMn resolves only
|
|
# for n < 10, and double-digit ports are routine on Windows.
|
|
if path.lower().startswith("com") and path[3:].isdigit():
|
|
path = rf"\\.\{path}"
|
|
self.handle = _k32.CreateFileW(
|
|
path, _GENERIC_READ | _GENERIC_WRITE, 0, None, _OPEN_EXISTING, 0, None
|
|
)
|
|
if self.handle == _INVALID_HANDLE:
|
|
_fail(f"cannot open {path}")
|
|
self.timeouts = None
|
|
try:
|
|
dcb = _DCB()
|
|
dcb.DCBlength = ctypes.sizeof(_DCB)
|
|
if not _k32.GetCommState(self.handle, ctypes.byref(dcb)):
|
|
_fail(f"cannot read the state of {path}")
|
|
dcb.BaudRate, dcb.ByteSize, dcb.Parity, dcb.StopBits = baud, 8, 0, 0 # 8N1
|
|
# fBinary, and DTR/RTS asserted (fDtrControl and fRtsControl,
|
|
# two bits each, = _ENABLE); every other flag clear, so no
|
|
# parity and no flow control. Raising both matches what opening
|
|
# a POSIX tty does — including the reset pulse on the boards
|
|
# that wire DTR to it.
|
|
dcb.fBits = 0x1 | (1 << 4) | (1 << 12)
|
|
if not _k32.SetCommState(self.handle, ctypes.byref(dcb)):
|
|
_fail(f"cannot configure {path} for {baud} baud 8N1")
|
|
# Arm them once here too: reads re-arm per call, but the write
|
|
# timeout would otherwise stay at the driver's default — which
|
|
# may be "wait forever" — until the first read.
|
|
self._deadline(_GAP_MS, 1000)
|
|
except Error:
|
|
# An open port outlives the exception otherwise, and a COM
|
|
# handle is exclusive: the next attempt would meet its own
|
|
# leftover as "Access is denied".
|
|
self.close()
|
|
raise
|
|
|
|
def close(self):
|
|
_k32.CloseHandle(self.handle)
|
|
|
|
def _deadline(self, interval, total):
|
|
"""Arm the driver's read timeouts: `interval` ms of quiet ends a
|
|
read once bytes have arrived, `total` ms ends it regardless."""
|
|
if self.timeouts == (interval, total):
|
|
return
|
|
spec = _COMMTIMEOUTS()
|
|
spec.ReadIntervalTimeout = interval
|
|
spec.ReadTotalTimeoutConstant = total
|
|
spec.WriteTotalTimeoutConstant = 5000
|
|
if not _k32.SetCommTimeouts(self.handle, ctypes.byref(spec)):
|
|
_fail("cannot set the port timeouts")
|
|
self.timeouts = (interval, total)
|
|
|
|
def _read(self, count):
|
|
buffer = ctypes.create_string_buffer(count)
|
|
got = wintypes.DWORD()
|
|
if not _k32.ReadFile(self.handle, buffer, count, ctypes.byref(got), None):
|
|
_fail("read failed")
|
|
return buffer.raw[: got.value]
|
|
|
|
def write(self, data):
|
|
written = wintypes.DWORD()
|
|
if not _k32.WriteFile(self.handle, data, len(data), ctypes.byref(written), None):
|
|
_fail("write failed")
|
|
if written.value != len(data):
|
|
raise Error(f"short write: {written.value} of {len(data)} bytes")
|
|
|
|
def flush_input(self):
|
|
if not _k32.PurgeComm(self.handle, _PURGE_RXCLEAR):
|
|
_fail("cannot flush the input buffer")
|
|
|
|
def read_available(self, wait):
|
|
"""Everything that arrives within `wait` seconds of quiet start."""
|
|
# A zero total means *no* timeout to the driver, so never round
|
|
# down to it — the same trap on the deadline below.
|
|
self._deadline(_GAP_MS, max(1, round(wait * 1000)))
|
|
return self._read(4096)
|
|
|
|
def read_exact(self, count, timeout):
|
|
data = b""
|
|
deadline = time.monotonic() + timeout
|
|
while len(data) < count:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise Error(f"timeout: got {len(data)} of {count} bytes")
|
|
# No interval timeout here: only the count or the deadline
|
|
# ends the read, so a gap mid-reply is simply waited out.
|
|
self._deadline(0, max(1, round(remaining * 1000)))
|
|
data += self._read(count - len(data))
|
|
return data
|
|
|
|
|
|
Port = WindowsPort if os.name == "nt" else PosixPort
|
|
|
|
|
|
# -------------------------------------------------------------- protocol ---
|
|
|
|
|
|
class Info:
|
|
"""The 12-byte info block."""
|
|
|
|
def __init__(self, raw):
|
|
if len(raw) != 12 or raw[0:2] != b"PB":
|
|
raise Error(f"bad info block: {raw.hex()}")
|
|
self.version = raw[2]
|
|
if not OLDEST_LOADER <= self.version <= NEWEST_LOADER:
|
|
raise Error(
|
|
f"pureboot {self.version}: this tool (version {VERSION}) speaks pureboot "
|
|
f"{OLDEST_LOADER}..{NEWEST_LOADER} — a newer loader needs a newer tool"
|
|
)
|
|
self.raw = bytes(raw)
|
|
self.signature = raw[3:6]
|
|
self.page = raw[6] or 256 # the wire count convention: 0 means 256
|
|
self.patch_vector = bool(raw[11] & 1)
|
|
# Large chips speak word addresses for flash (bit 1); the host keeps
|
|
# every address in bytes and converts at the wire.
|
|
self.word_flash = bool(raw[11] & 2)
|
|
scale = 2 if self.word_flash else 1
|
|
self.base = (raw[7] | (raw[8] << 8)) * scale
|
|
self.eeprom_size = raw[9] | (raw[10] << 8)
|
|
self.slot = 1024 if self.word_flash else SLOT
|
|
self.flash_size = self.base + self.slot
|
|
self.stage = self.base - self.slot # where a staging copy of the loader goes
|
|
# The hand-over target, as the word address 'J' takes: the trampoline
|
|
# below the loader (tinies), or word 0 (mega — the application's own
|
|
# reset vector; BOOTRST re-vectors a reset into the loader instead).
|
|
self.app_entry_word = (self.base - 2) // 2 if self.patch_vector else 0
|
|
|
|
def describe(self):
|
|
sig = " ".join(f"{b:02x}" for b in self.signature)
|
|
vector = "host-patched reset vector" if self.patch_vector else "hardware boot section"
|
|
return (
|
|
f"signature {sig}, page {self.page} B, "
|
|
f"app flash {self.base} B (loader at {self.base:#06x}), "
|
|
f"EEPROM {self.eeprom_size} B, {vector}"
|
|
)
|
|
|
|
def lines(self):
|
|
"""The info block as one fact per line — what --info prints."""
|
|
if self.patch_vector:
|
|
hand_over = f"host-patched reset vector, trampoline at {self.base - 2:#06x}"
|
|
else:
|
|
hand_over = "hardware boot section, jump to word 0"
|
|
return (
|
|
f"version pureboot {self.version}",
|
|
f"signature {' '.join(f'{b:02x}' for b in self.signature)}",
|
|
f"flash {self.flash_size} B, {self.page} B pages"
|
|
+ (", word-addressed wire" if self.word_flash else ""),
|
|
f"application 0x0000..{self.base - 1:#06x} ({self.base} B)",
|
|
f"loader {self.base:#06x} ({self.slot} B slot)",
|
|
f"staging {self.stage:#06x}",
|
|
f"EEPROM {self.eeprom_size} B",
|
|
f"hand-over {hand_over}",
|
|
)
|
|
|
|
|
|
class Loader:
|
|
"""A pureboot session. Between commands the loader has prompted `+` and
|
|
awaits a command byte; every method restores that invariant — except
|
|
jump(), after which the target must be knocked afresh."""
|
|
|
|
def __init__(self, port):
|
|
self.port = port
|
|
self.info = None
|
|
|
|
def connect(self, wait):
|
|
"""Knock until the activation window answers, then read the info
|
|
block. Also converges when the loader already sits in its command
|
|
loop: the knock bytes are ignored-or-executed there, and the drain
|
|
absorbs whatever they produced."""
|
|
self.port.flush_input()
|
|
deadline = time.monotonic() + wait
|
|
knocks = 0
|
|
while True:
|
|
self.port.write(b"pb")
|
|
knocks += 1
|
|
if PROMPT in self.port.read_available(0.4):
|
|
break
|
|
if time.monotonic() > deadline:
|
|
raise Error("no answer — reset the device within its activation window")
|
|
while self.port.read_available(0.3):
|
|
pass
|
|
self.port.write(b"b")
|
|
self.info = Info(self.port.read_exact(12, 2.0))
|
|
self._expect_prompt()
|
|
verbose(f"loader answered knock {knocks}; info block read")
|
|
return self.info
|
|
|
|
def _expect_prompt(self, timeout=2.0):
|
|
byte = self.port.read_exact(1, timeout)
|
|
if byte != PROMPT:
|
|
raise Error(f"expected prompt, got {byte.hex()}")
|
|
|
|
def _command(self, tx, reply_len=0, timeout=2.0):
|
|
self.port.write(tx)
|
|
reply = self.port.read_exact(reply_len, timeout) if reply_len else b""
|
|
self._expect_prompt(timeout)
|
|
return reply
|
|
|
|
def _stream_read(self, command, address, count, address_scale=1):
|
|
data = b""
|
|
while count:
|
|
chunk = min(count, 256)
|
|
wire = address // address_scale
|
|
head = bytes((ord(command), wire & 0xFF, wire >> 8, chunk & 0xFF))
|
|
data += self._command(head, chunk, 5.0)
|
|
address += chunk
|
|
count -= chunk
|
|
return data
|
|
|
|
def read_flash(self, address, count):
|
|
if not self.info.word_flash:
|
|
return self._stream_read("R", address, count)
|
|
# Word-addressed wire: widen to even bounds and never let one read
|
|
# cross a 64 KiB boundary (the device holds RAMPZ for a whole run).
|
|
start = address & ~1
|
|
span = (address + count + 1 & ~1) - start
|
|
data = b""
|
|
at = start
|
|
remaining = span
|
|
while remaining:
|
|
chunk = min(remaining, 0x10000 - (at & 0xFFFF))
|
|
data += self._stream_read("R", at, chunk, address_scale=2)
|
|
at += chunk
|
|
remaining -= chunk
|
|
return data[address - start : address - start + count]
|
|
|
|
def read_eeprom(self, address, count):
|
|
return self._stream_read("r", address, count)
|
|
|
|
def write_page(self, address, data):
|
|
assert len(data) == self.info.page and address % self.info.page == 0
|
|
wire = address // (2 if self.info.word_flash else 1)
|
|
head = bytes((ord("W"), wire & 0xFF, wire >> 8))
|
|
self._command(head + data, 0, 2.0)
|
|
|
|
def write_eeprom(self, address, data, progress=None):
|
|
offset = 0
|
|
while offset < len(data):
|
|
chunk = data[offset : offset + 256]
|
|
head = bytes((ord("w"), address & 0xFF, address >> 8, len(chunk) & 0xFF))
|
|
self.port.write(head)
|
|
for byte in chunk:
|
|
self.port.write(bytes((byte,)))
|
|
self._expect_prompt() # per-byte ack: the write has begun
|
|
if progress:
|
|
progress.step()
|
|
self._expect_prompt() # the next command prompt
|
|
address += len(chunk)
|
|
offset += len(chunk)
|
|
|
|
def read_fuses(self):
|
|
return self._command(b"F", 4, 2.0)
|
|
|
|
def jump(self, word_address):
|
|
"""'J': the device acks, then execution continues at the word
|
|
address — a loader slot's base (whose copy must then be knocked
|
|
afresh) or the application entry."""
|
|
self.port.write(bytes((ord("J"), word_address & 0xFF, word_address >> 8)))
|
|
self._expect_prompt()
|
|
|
|
def enter_copy(self, byte_address, wait):
|
|
"""Jump into the loader copy at `byte_address` and knock it. Ending
|
|
up in the copy addressed is guaranteed by construction: a jump to a
|
|
slot base lands in that slot's entry stub."""
|
|
self.jump(byte_address // 2)
|
|
return self.connect(wait)
|
|
|
|
def run_application(self):
|
|
self.jump(self.info.app_entry_word)
|
|
|
|
|
|
# ---------------------------------------------------------------- images ---
|
|
|
|
|
|
def load_image(path):
|
|
"""Raw binary, or Intel HEX by extension (.hex/.ihx/.ihex)."""
|
|
data = open(path, "rb").read()
|
|
if not path.lower().endswith((".hex", ".ihx", ".ihex")):
|
|
if not data:
|
|
raise Error(f"{path}: empty image")
|
|
return data
|
|
memory = {}
|
|
for number, line in enumerate(data.decode("ascii", "replace").splitlines(), 1):
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
if not line.startswith(":"):
|
|
raise Error(f"{path}:{number}: not an Intel HEX record")
|
|
record = bytes.fromhex(line[1:])
|
|
if sum(record) & 0xFF:
|
|
raise Error(f"{path}:{number}: checksum mismatch")
|
|
count, address, kind = record[0], (record[1] << 8) | record[2], record[3]
|
|
payload = record[4 : 4 + count]
|
|
if kind == 0:
|
|
for i, byte in enumerate(payload):
|
|
memory[address + i] = byte
|
|
elif kind == 1:
|
|
break
|
|
elif kind in (2, 4) and not any(payload):
|
|
continue # a zero base extends nothing
|
|
elif kind in (3, 5):
|
|
continue # start address: irrelevant, reset is the entry
|
|
else:
|
|
raise Error(f"{path}:{number}: record type {kind} reaches beyond the 16-bit space")
|
|
if not memory:
|
|
raise Error(f"{path}: empty image")
|
|
return bytes(memory.get(i, 0xFF) for i in range(max(memory) + 1))
|
|
|
|
|
|
# --------------------------------------------------------------- surgery ---
|
|
|
|
|
|
def rjmp_target(word_address, opcode, flash_words):
|
|
return (word_address + 1 + (opcode & 0x0FFF)) % flash_words
|
|
|
|
|
|
def rjmp_to(word_address, destination, flash_words):
|
|
return 0xC000 | ((destination - word_address - 1) % flash_words % 0x1000)
|
|
|
|
|
|
def plan_flash(image, info):
|
|
"""The pages to program, as {page_address: bytes}, already carrying the
|
|
reset-vector surgery where the chip needs it."""
|
|
page = info.page
|
|
limit = info.base - (2 if info.patch_vector else 0)
|
|
if len(image) > limit:
|
|
raise Error(f"image is {len(image)} B, application flash ends at {limit}")
|
|
final = bytearray(image) + bytearray([0xFF] * (-len(image) % page))
|
|
|
|
if info.patch_vector:
|
|
flash_words = info.flash_size // 2
|
|
word0 = final[0] | (final[1] << 8)
|
|
if word0 & 0xF000 != 0xC000:
|
|
raise Error(
|
|
"the image's reset vector is not an rjmp — pureboot's vector "
|
|
"surgery cannot re-home it (crt-less entry at address 0?)"
|
|
)
|
|
entry = rjmp_target(0, word0, flash_words)
|
|
if entry >= info.base // 2:
|
|
raise Error(
|
|
"the image's reset vector already targets the loader — this "
|
|
"is a read-back of a patched image; flash the original"
|
|
)
|
|
trampoline_word = (info.base - 2) // 2
|
|
patch = rjmp_to(0, info.base // 2, flash_words)
|
|
final[0], final[1] = patch & 0xFF, patch >> 8
|
|
trampoline_page = info.base - page
|
|
if len(final) < trampoline_page + page:
|
|
final += bytearray([0xFF] * (trampoline_page + page - len(final)))
|
|
jump = rjmp_to(trampoline_word, entry, flash_words)
|
|
final[info.base - 2], final[info.base - 1] = jump & 0xFF, jump >> 8
|
|
verbose(f"vector surgery: word 0 -> loader {info.base:#06x}, "
|
|
f"trampoline {info.base - 2:#06x} -> entry word {entry:#06x}")
|
|
|
|
pages = {a: bytes(final[a : a + page]) for a in range(0, len(final), page)}
|
|
return pages
|
|
|
|
|
|
def covered(pages, info, skip_blank):
|
|
"""Pages in programming order; optionally dropping all-0xff pages (sound
|
|
only over erased flash) — never a load-bearing one.
|
|
|
|
With a patched vector (tinies), the patched page 0 goes first and the
|
|
trampoline page second: from the first write on, a reset lands in the
|
|
loader and the loader's own fall-through lands on the application entry,
|
|
so every interruption point of the flash is recoverable. With a hardware
|
|
boot section a reset re-vectors to the loader regardless; ascending
|
|
order, page 0 last, maximizes what an interrupted image retains."""
|
|
trampoline_page = info.base - info.page if info.patch_vector else None
|
|
first = [0, trampoline_page] if info.patch_vector else []
|
|
rest = [a for a in sorted(pages) if a not in first]
|
|
if skip_blank:
|
|
rest = [a for a in rest if pages[a].count(0xFF) != len(pages[a])]
|
|
order = [a for a in first if a in pages] + rest
|
|
if not info.patch_vector:
|
|
order = [a for a in order if a != 0] + ([0] if 0 in pages else [])
|
|
return order
|
|
|
|
|
|
# ----------------------------------------------------------------- fuses ---
|
|
|
|
|
|
# Per-chip boot fuse geometry, keyed by the signature's family/part bytes:
|
|
# which byte of the 'F' reply (low, lock, extended, high) carries BOOTSZ/
|
|
# BOOTRST, and the BOOTSZ->words ladder. A die revision shares its base
|
|
# signature, so one row covers it. The m48s have no boot section and no
|
|
# row — their info block says patch-vector and this table is never
|
|
# consulted. Sources: Atmel-2486/2466/2503 (HIGH fuse), Atmel-2545/8271/
|
|
# DS40002065 (x8: EXTENDED, except the m328s' HIGH), Atmel-8272/8011/2593/
|
|
# 42719 (x4: HIGH).
|
|
_LADDER_128 = {0b11: 128, 0b10: 256, 0b01: 512, 0b00: 1024}
|
|
_LADDER_256 = {0b11: 256, 0b10: 512, 0b01: 1024, 0b00: 2048}
|
|
_LADDER_512 = {0b11: 512, 0b10: 1024, 0b01: 2048, 0b00: 4096}
|
|
BOOT_FUSE = {
|
|
bytes((0x93, 0x07)): (3, _LADDER_128), # m8/8A
|
|
bytes((0x94, 0x03)): (3, _LADDER_128), # m16/16A
|
|
bytes((0x95, 0x02)): (3, _LADDER_256), # m32/32A
|
|
bytes((0x93, 0x0A)): (2, _LADDER_128), # m88/88A
|
|
bytes((0x93, 0x0F)): (2, _LADDER_128), # m88P/88PA
|
|
bytes((0x94, 0x06)): (2, _LADDER_128), # m168/168A
|
|
bytes((0x94, 0x0B)): (2, _LADDER_128), # m168P/168PA
|
|
bytes((0x95, 0x14)): (3, _LADDER_256), # m328
|
|
bytes((0x95, 0x0F)): (3, _LADDER_256), # m328P
|
|
bytes((0x94, 0x0F)): (3, _LADDER_128), # m164A
|
|
bytes((0x94, 0x0A)): (3, _LADDER_128), # m164P/164PA
|
|
bytes((0x95, 0x15)): (3, _LADDER_256), # m324A
|
|
bytes((0x95, 0x08)): (3, _LADDER_256), # m324P
|
|
bytes((0x95, 0x11)): (3, _LADDER_256), # m324PA
|
|
bytes((0x96, 0x09)): (3, _LADDER_512), # m644/644A
|
|
bytes((0x96, 0x0A)): (3, _LADDER_512), # m644P/644PA
|
|
bytes((0x97, 0x06)): (3, _LADDER_512), # m1284
|
|
bytes((0x97, 0x05)): (3, _LADDER_512), # m1284P
|
|
}
|
|
|
|
|
|
def mega_boot(info, fuse_bytes):
|
|
"""Decode a mega's boot configuration from its fuses (the byte and the
|
|
BOOTSZ ladder are per chip): BOOTSZ1:0 in bits 2:1 select the
|
|
boot-section words, BOOTRST in bit 0 (programmed = 0) re-vectors reset
|
|
to its start. Returns (bootrst_programmed, boot_section_start_byte)."""
|
|
entry = BOOT_FUSE.get(bytes(info.signature[1:3]))
|
|
if entry is None:
|
|
raise Error(f"unknown mega signature {info.signature.hex()} — no boot fuse map")
|
|
which, ladder = entry
|
|
fuse = fuse_bytes[which]
|
|
words = ladder[(fuse >> 1) & 0x03]
|
|
return (fuse & 1) == 0, info.flash_size - words * 2
|
|
|
|
|
|
# ---------------------------------------------------------- loader update ---
|
|
|
|
|
|
def image_info(image):
|
|
"""The info block embedded in a pureboot binary, or None. Searched per
|
|
known loader version, so the magic stays three selective bytes rather than
|
|
two that code could carry by chance — and a binary this tool does not know
|
|
the version of reads as no block at all, which is what it is to the tool."""
|
|
for version in range(OLDEST_LOADER, NEWEST_LOADER + 1):
|
|
at = image.find(b"PB" + bytes((version,)))
|
|
if 0 <= at <= len(image) - 12:
|
|
return Info(image[at : at + 12])
|
|
return None
|
|
|
|
|
|
def loader_image(path):
|
|
"""A loader update image, as the slot's own content. A raw binary is that
|
|
already; an Intel HEX links the loader at its base inside an otherwise
|
|
blank flash image, and load_image() anchors every image at zero, so the
|
|
blank below the base is dropped here. The base comes from the image's own
|
|
info block rather than the device's, so an image built for somewhere else
|
|
survives intact and the preflight can say so."""
|
|
image = load_image(path)
|
|
embedded = image_info(image)
|
|
if embedded and len(image) > embedded.base:
|
|
image = image[embedded.base :]
|
|
return image
|
|
|
|
|
|
def staging_content(image, info):
|
|
"""The 512-byte staging-slot content: the image, padding, and — on
|
|
chips whose hand-over jumps through the word below the resident loader —
|
|
that word, which for a staging copy is the slot's own last word: an rjmp
|
|
to the resident base. The staging copy's fall-through and 'J'-free exit
|
|
both land in a loader instead of garbage."""
|
|
slot = info.slot
|
|
if len(image) > (slot - 2 if info.patch_vector else slot):
|
|
raise Error(f"loader image is {len(image)} B, the slot holds {slot - 2 if info.patch_vector else slot}")
|
|
content = bytearray(image) + bytearray([0xFF] * (slot - len(image)))
|
|
if info.patch_vector:
|
|
through = rjmp_to((info.base - 2) // 2, info.base // 2, info.flash_size // 2)
|
|
content[slot - 2], content[slot - 1] = through & 0xFF, through >> 8
|
|
return bytes(content)
|
|
|
|
|
|
def update_preflight(image, info, fuse_bytes):
|
|
"""Errors and warnings before any flash is touched. Returns warnings."""
|
|
embedded = image_info(image)
|
|
if embedded is None:
|
|
raise Error(
|
|
"no pureboot info block in the update image — not a pureboot binary, "
|
|
f"or a version this tool ({VERSION}) does not know"
|
|
)
|
|
if embedded.raw[3:] != info.raw[3:]:
|
|
raise Error(
|
|
f"update image is for another target: it declares "
|
|
f"[{embedded.describe()}], the device says [{info.describe()}]"
|
|
)
|
|
warnings = []
|
|
if not info.patch_vector:
|
|
if fuse_bytes is None:
|
|
raise Error("a loader update on this chip needs its fuses — unreadable? pass --assume-fuses")
|
|
bootrst, bls_start = mega_boot(info, fuse_bytes)
|
|
if info.stage < bls_start:
|
|
raise Error(
|
|
f"cannot self-update: the staging slot {info.stage:#06x} lies below the "
|
|
f"boot section ({bls_start:#06x}) where SPM is disabled "
|
|
f"— a boot section of at least two slots ({2 * info.slot} B, BOOTSZ) is "
|
|
f"required, and only an external programmer can change fuses"
|
|
)
|
|
if not bootrst:
|
|
warnings.append(
|
|
"BOOTRST unprogrammed: reset boots the application throughout the update; "
|
|
"an interruption is recovered by re-running this update"
|
|
)
|
|
elif bls_start == info.stage:
|
|
warnings.append(
|
|
"BOOTRST targets the staging slot: brief unrecoverable windows exist while "
|
|
"the staging copy itself is being installed or retired (page-write scale)"
|
|
)
|
|
else:
|
|
warnings.append(
|
|
f"BOOTRST targets {bls_start:#06x}, inside application flash: reset reaches a "
|
|
f"loader only across erased flash from there"
|
|
)
|
|
return warnings
|
|
|
|
|
|
class UpdateState:
|
|
"""The host-side memory of an update in flight: what the staging slot
|
|
held (and page 0, where the update repoints it). Losing this file after
|
|
the staging slot was overwritten loses those saved bytes — the update
|
|
still completes, but the staging region can then only be restored by
|
|
reflashing the application."""
|
|
|
|
def __init__(self, path):
|
|
self.path = path
|
|
self.data = None
|
|
|
|
def load_or_save(self, loader):
|
|
info = loader.info
|
|
if os.path.exists(self.path):
|
|
self.data = json.load(open(self.path))
|
|
if bytes.fromhex(self.data["signature"]) != info.signature or self.data["base"] != info.base:
|
|
raise Error(f"{self.path} belongs to a different device — remove it to start over")
|
|
return
|
|
self.data = {
|
|
"signature": info.signature.hex(),
|
|
"base": info.base,
|
|
"staging": loader.read_flash(info.stage, info.slot).hex(),
|
|
"page0": loader.read_flash(0, info.page).hex() if info.patch_vector else "",
|
|
}
|
|
with open(self.path, "w") as f:
|
|
json.dump(self.data, f)
|
|
|
|
@property
|
|
def staging(self):
|
|
return bytes.fromhex(self.data["staging"])
|
|
|
|
@property
|
|
def page0(self):
|
|
return bytes.fromhex(self.data["page0"])
|
|
|
|
def discard(self):
|
|
os.unlink(self.path)
|
|
|
|
|
|
def write_differing(loader, base, content, order=None, label=None):
|
|
"""Program the pages of `content` at `base` that differ from flash —
|
|
idempotent, so a resumed phase redoes only what an interruption left.
|
|
A label puts the compare-and-program loop on the progress bar."""
|
|
page = loader.info.page
|
|
offsets = list(order) if order is not None else list(range(0, len(content), page))
|
|
written = 0
|
|
with Progress(label, len(offsets)) as bar:
|
|
for offset in offsets:
|
|
want = content[offset : offset + page]
|
|
if loader.read_flash(base + offset, page) != want:
|
|
loader.write_page(base + offset, want)
|
|
written += 1
|
|
bar.step()
|
|
if label:
|
|
verbose(f"{label}: {written} of {len(offsets)} pages differed")
|
|
# Page-wise read-back with the same bounded repair as verify_pages: this
|
|
# is the loader-update path, where a page left wrong is a half-written
|
|
# loader slot.
|
|
for retry in range(RETRIES + 1):
|
|
bad = [
|
|
offset
|
|
for offset in range(0, len(content), page)
|
|
if loader.read_flash(base + offset, len(content[offset : offset + page])) != content[offset : offset + page]
|
|
]
|
|
if not bad:
|
|
break
|
|
if retry == RETRIES:
|
|
raise Error(
|
|
f"verify failed at {base + bad[0]:#06x} after programming "
|
|
f"(still wrong after {RETRIES} retries)"
|
|
)
|
|
for offset in bad:
|
|
verbose(f"rewriting page {base + offset:#06x} (retry {retry + 1})")
|
|
loader.write_page(base + offset, content[offset : offset + page])
|
|
written += 1
|
|
return written
|
|
|
|
|
|
def patch_word0(loader, page0, target_base):
|
|
"""Rewrite page 0 with its word 0 re-aimed at `target_base` — the
|
|
resume insurance around rewriting a loader slot the reset path uses."""
|
|
info = loader.info
|
|
patched = bytearray(page0)
|
|
word = rjmp_to(0, target_base // 2, info.flash_size // 2)
|
|
patched[0], patched[1] = word & 0xFF, word >> 8
|
|
write_differing(loader, 0, bytes(patched))
|
|
return bytes(patched)
|
|
|
|
|
|
def op_update_loader(loader, wait, path, state_path, fuse_bytes):
|
|
"""Replace the resident loader with `path`, using the loader itself as
|
|
its own staging loader. Every phase is idempotent and keyed off the
|
|
actual flash state, so a re-run after any interruption resumes; the
|
|
state file carries the bytes the staging slot held."""
|
|
info = loader.info
|
|
image = loader_image(path)
|
|
for warning in update_preflight(image, info, fuse_bytes):
|
|
print(f"note: {warning}")
|
|
update = image_info(image) # the preflight proved it is there
|
|
verbose(f"installing pureboot {update.version} over pureboot {info.version}")
|
|
staged = staging_content(image, info)
|
|
resident = bytes(image) + bytes([0xFF] * (info.slot - len(image)))
|
|
page = info.page
|
|
|
|
state = UpdateState(state_path)
|
|
if os.path.exists(state_path):
|
|
verbose(f"resuming the update recorded in {state_path}")
|
|
else:
|
|
verbose(f"saving the staging slot to {state_path}")
|
|
state.load_or_save(loader)
|
|
|
|
# Install the staging copy — unless a loader already sits whole in the
|
|
# staging slot (a build programmed there by hand): that copy IS the
|
|
# installed staging copy, and rewriting it would only trip its own
|
|
# running-slot guard on the composed through-word. Any pureboot with
|
|
# the device's own info block serves — the staged copy just streams
|
|
# pages, so an older build installs a newer resident all the same. Two
|
|
# checks make "already a loader" mean a *complete* one: the block must
|
|
# sit where every image carries it (within the slot's first 256 bytes
|
|
# — the build's position lint), matching the device's block byte for
|
|
# byte, and the slot must be unchanged since this update began (the
|
|
# state file's snapshot) — a resumed, half-written install differs
|
|
# from its snapshot and takes the install path below, which completes
|
|
# it page by page.
|
|
current = loader.read_flash(info.stage, info.slot)
|
|
staged_loader = image_info(current[:268])
|
|
if staged_loader is not None and staged_loader.raw == info.raw and current == state.staging:
|
|
print("staging slot already holds a loader — left in place")
|
|
else:
|
|
# On a chip whose staging slot starts at address 0 (the 1 KB
|
|
# tiny13s), its first page carries the reset vector: written last,
|
|
# so any earlier interruption still resets into the old resident,
|
|
# and from then on resets enter the staging copy.
|
|
order = list(range(0, info.slot, page))
|
|
if info.stage == 0:
|
|
order = order[1:] + [0]
|
|
if write_differing(loader, info.stage, staged, order, label="staging copy"):
|
|
print(f"staging copy installed at {info.stage:#06x}")
|
|
|
|
# Enter it and let it rewrite the resident slot. Where a patched reset
|
|
# vector routes through the resident (a tiny with the staging slot away
|
|
# from page 0), word 0 is re-aimed at the staging copy around the
|
|
# rewrite, so a power failure mid-rewrite still resets into a loader.
|
|
verbose(f"entering the staging copy at {info.stage:#06x}")
|
|
loader.enter_copy(info.stage, wait)
|
|
redirect = info.patch_vector and info.stage != 0
|
|
if redirect:
|
|
verbose("word 0 re-aimed at the staging copy for the rewrite")
|
|
patch_word0(loader, state.page0, info.stage)
|
|
if write_differing(loader, info.base, resident, label="resident"):
|
|
print(f"resident loader rewritten at {info.base:#06x}")
|
|
|
|
# Enter the new resident and put the staging region back: page 0 first
|
|
# where it lives in that region (word 0 then points at the new resident
|
|
# for the rest of the restore), the saved trampoline with the rest.
|
|
verbose(f"entering the new resident at {info.base:#06x}")
|
|
loader.enter_copy(info.base, wait)
|
|
if redirect:
|
|
verbose("word 0 restored")
|
|
write_differing(loader, 0, state.page0)
|
|
order = list(range(0, info.slot, page))
|
|
if info.stage == 0:
|
|
order = [0] + order[1:]
|
|
write_differing(loader, info.stage, state.staging, order, label="staging restore")
|
|
|
|
state.discard()
|
|
print(f"loader updated: pureboot {update.version}, {len(image)} B at {info.base:#06x}, staging region restored")
|
|
|
|
|
|
def check_walk_region(pages, info, fuse_bytes, force):
|
|
"""With BOOTRST programmed but targeting below the loader, reset reaches
|
|
the loader only by walking across erased flash from the boot-section
|
|
start; application data in that span would divert reset into itself.
|
|
Only checkable when the fuses are known (--fuses or --assume-fuses)."""
|
|
if info.patch_vector or fuse_bytes is None:
|
|
return
|
|
bootrst, bls_start = mega_boot(info, fuse_bytes)
|
|
if not bootrst or bls_start >= info.base:
|
|
return
|
|
overlap = [a for a in sorted(pages) if a >= bls_start and pages[a].count(0xFF) != len(pages[a])]
|
|
if overlap and not force:
|
|
raise Error(
|
|
f"the image writes {overlap[0]:#06x}.. inside the reset walk region "
|
|
f"[{bls_start:#06x}, {info.base:#06x}) (BOOTRST programmed): reset could no "
|
|
f"longer reach the loader — --force to flash it anyway"
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------ operations ---
|
|
|
|
|
|
def op_erase_flash(loader):
|
|
"""0xff over the whole application area. Descending on a patched-vector
|
|
chip: page 0 — the patched reset vector — goes last, so an interrupted
|
|
erase still resets into the loader, and once it is gone the whole area
|
|
is erased and the reset walk reaches the loader anyway."""
|
|
blank = bytes([0xFF] * loader.info.page)
|
|
addresses = range(0, loader.info.base, loader.info.page)
|
|
with Progress("erase", len(addresses)) as bar:
|
|
for address in reversed(addresses) if loader.info.patch_vector else addresses:
|
|
loader.write_page(address, blank)
|
|
bar.step()
|
|
print(f"erase: {loader.info.base // loader.info.page} pages")
|
|
|
|
|
|
def op_erase_eeprom(loader):
|
|
with Progress("erase EEPROM", loader.info.eeprom_size, "B") as bar:
|
|
loader.write_eeprom(0, bytes([0xFF] * loader.info.eeprom_size), progress=bar)
|
|
print(f"erase: {loader.info.eeprom_size} B of EEPROM")
|
|
|
|
|
|
def op_flash(loader, path, erase, verify, fuse_bytes=None, force=False):
|
|
image = load_image(path)
|
|
verbose(f"{path}: {len(image)} B image")
|
|
pages = plan_flash(image, loader.info)
|
|
check_walk_region(pages, loader.info, fuse_bytes, force)
|
|
if erase:
|
|
op_erase_flash(loader)
|
|
order = covered(pages, loader.info, skip_blank=erase)
|
|
if len(order) != len(pages):
|
|
verbose(f"{len(pages) - len(order)} blank pages skipped (erased flash underneath)")
|
|
with Progress("flash", len(order)) as bar:
|
|
for address in order:
|
|
loader.write_page(address, pages[address])
|
|
bar.step()
|
|
print(f"flash: {path}: {len(order)} pages")
|
|
if verify:
|
|
verify_pages(loader, pages, repair=True)
|
|
|
|
|
|
def verify_pages(loader, pages, repair=False):
|
|
"""Read every page back and compare. With `repair`, a mismatched page is
|
|
rewritten and re-read, up to RETRIES times before it is raised: a page
|
|
filled over a dirty SPM buffer takes stale words, and the write that took
|
|
them cleared the buffer, so one rewrite settles it. Anything still wrong
|
|
after three is not that, and stops the run."""
|
|
repaired = 0
|
|
with Progress("verify", len(pages)) as bar:
|
|
for address in sorted(pages):
|
|
for retry in range(RETRIES + 1):
|
|
got = loader.read_flash(address, loader.info.page)
|
|
if got == pages[address]:
|
|
break
|
|
first = next(i for i in range(len(got)) if got[i] != pages[address][i])
|
|
detail = (
|
|
f"verify failed at {address + first:#06x}: "
|
|
f"wrote {pages[address][first]:02x}, read {got[first]:02x}"
|
|
)
|
|
if not repair:
|
|
raise Error(detail)
|
|
if retry == RETRIES:
|
|
raise Error(f"{detail} (still wrong after {RETRIES} retries)")
|
|
verbose(f"{detail} — rewriting page {address:#06x} (retry {retry + 1})")
|
|
loader.write_page(address, pages[address])
|
|
repaired += 1
|
|
bar.step()
|
|
note = f", {repaired} page rewrite(s)" if repaired else ""
|
|
print(f"verify: {len(pages)} pages ok{note}")
|
|
|
|
|
|
def op_verify_flash(loader, path):
|
|
verify_pages(loader, plan_flash(load_image(path), loader.info))
|
|
|
|
|
|
def read_progress(reader, total, label):
|
|
"""A bulk read in 256-byte wire chunks under a progress bar."""
|
|
data = b""
|
|
with Progress(label, total, "B") as bar:
|
|
while len(data) < total:
|
|
chunk = min(256, total - len(data))
|
|
data += reader(len(data), chunk)
|
|
bar.step(chunk)
|
|
return data
|
|
|
|
|
|
def op_read_flash(loader, path):
|
|
data = read_progress(loader.read_flash, loader.info.base, "read flash")
|
|
open(path, "wb").write(data)
|
|
print(f"read flash: {len(data)} B -> {path}")
|
|
|
|
|
|
def op_eeprom(loader, path, erase, verify):
|
|
image = load_image(path)
|
|
if len(image) > loader.info.eeprom_size:
|
|
raise Error(f"EEPROM image is {len(image)} B, device has {loader.info.eeprom_size}")
|
|
if erase:
|
|
op_erase_eeprom(loader)
|
|
with Progress("eeprom", len(image), "B") as bar:
|
|
loader.write_eeprom(0, image, progress=bar)
|
|
print(f"eeprom: {path}: {len(image)} B")
|
|
if verify:
|
|
got = read_progress(loader.read_eeprom, len(image), "verify EEPROM")
|
|
if got != image:
|
|
first = next(i for i in range(len(got)) if got[i] != image[i])
|
|
raise Error(f"verify failed at EEPROM {first:#06x}: wrote {image[first]:02x}, read {got[first]:02x}")
|
|
print(f"verify: {len(image)} B ok")
|
|
|
|
|
|
def op_verify_eeprom(loader, path):
|
|
image = load_image(path)
|
|
got = read_progress(loader.read_eeprom, len(image), "verify EEPROM")
|
|
if got != image:
|
|
first = next(i for i in range(len(got)) if got[i] != image[i])
|
|
raise Error(f"verify failed at EEPROM {first:#06x}: expected {image[first]:02x}, read {got[first]:02x}")
|
|
print(f"verify: {len(image)} B of EEPROM ok")
|
|
|
|
|
|
def op_read_eeprom(loader, path):
|
|
data = read_progress(loader.read_eeprom, loader.info.eeprom_size, "read EEPROM")
|
|
open(path, "wb").write(data)
|
|
print(f"read EEPROM: {len(data)} B -> {path}")
|
|
|
|
|
|
def op_fuses(loader):
|
|
low, lock, extended, high = loader.read_fuses()
|
|
print("fuses:")
|
|
print(f" low 0x{low:02x}")
|
|
print(f" high 0x{high:02x}")
|
|
print(f" extended 0x{extended:02x}")
|
|
print(f" lock 0x{lock:02x}")
|
|
fuse_bytes = bytes((low, lock, extended, high))
|
|
# On a boot-sectioned mega the BOOTSZ/BOOTRST decode is the fuse fact the
|
|
# loader's whole deployment hangs on — say it in words.
|
|
if not loader.info.patch_vector:
|
|
try:
|
|
bootrst, bls_start = mega_boot(loader.info, fuse_bytes)
|
|
reset = "reset enters it" if bootrst else "reset boots the application"
|
|
print(f" boot section at {bls_start:#06x} ({loader.info.flash_size - bls_start} B), "
|
|
f"BOOTRST {'programmed' if bootrst else 'unprogrammed'} — {reset}")
|
|
except Error:
|
|
pass # unknown signature: the raw bytes above still stand
|
|
return fuse_bytes
|
|
|
|
|
|
# -------------------------------------------------------------------- cli ---
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="pureboot host tool", epilog="operations run in the order listed above"
|
|
)
|
|
parser.add_argument("--version", action="version", version=f"%(prog)s {VERSION} "
|
|
f"(speaks pureboot {OLDEST_LOADER}..{NEWEST_LOADER})")
|
|
parser.add_argument("--port", required=True, help="serial device: COM6, /dev/ttyUSB0, or a simavr pty")
|
|
parser.add_argument("--baud", type=int, default=115200, help="115200 mega, 57600 tinies")
|
|
parser.add_argument("--wait", type=float, default=30.0, help="seconds to keep knocking")
|
|
parser.add_argument("--info", action="store_true", help="print the device info block")
|
|
parser.add_argument("--fuses", action="store_true", help="read the fuse and lock bytes")
|
|
parser.add_argument("--update-loader", metavar="FILE", help="replace the loader with this pureboot binary")
|
|
parser.add_argument("--state", metavar="FILE", help="update state file (default: FILE.pbstate)")
|
|
parser.add_argument("--assume-fuses", metavar="HEX8", help="fuse bytes low,lock,ext,high as 8 hex digits "
|
|
"(overrides reading them — e.g. under a simulator that cannot)")
|
|
parser.add_argument("--erase-flash", action="store_true", help="0xff over the application flash")
|
|
parser.add_argument("--flash", metavar="FILE", help="program an application (bin or ihex)")
|
|
parser.add_argument("--no-verify", action="store_true", help="skip read-back after writes")
|
|
parser.add_argument("--read-flash", metavar="FILE", help="dump the application flash")
|
|
parser.add_argument("--verify-flash", metavar="FILE", help="compare flash against an image")
|
|
parser.add_argument("--erase-eeprom", action="store_true", help="0xff over the EEPROM")
|
|
parser.add_argument("--eeprom", metavar="FILE", help="program the EEPROM (bin or ihex)")
|
|
parser.add_argument("--read-eeprom", metavar="FILE", help="dump the EEPROM")
|
|
parser.add_argument("--verify-eeprom", metavar="FILE", help="compare EEPROM against an image")
|
|
parser.add_argument("--force", action="store_true", help="override refusable safety checks")
|
|
parser.add_argument("--stay", action="store_true", help="leave the loader in its session")
|
|
parser.add_argument("-v", "--verbose", action="store_true",
|
|
help="print decisions and derived facts as operations run")
|
|
args = parser.parse_args()
|
|
global VERBOSE
|
|
VERBOSE = args.verbose
|
|
|
|
if args.update_loader and (args.flash or args.erase_flash):
|
|
parser.error("--update-loader does not combine with application flash operations")
|
|
fuse_override = None
|
|
if args.assume_fuses:
|
|
try:
|
|
fuse_override = bytes.fromhex(args.assume_fuses)
|
|
assert len(fuse_override) == 4
|
|
except (ValueError, AssertionError):
|
|
parser.error("--assume-fuses takes 8 hex digits: low,lock,extended,high")
|
|
|
|
port = Port(args.port, args.baud)
|
|
verbose(f"{args.port}: {args.baud} Bd 8N1, DTR/RTS asserted")
|
|
try:
|
|
loader = Loader(port)
|
|
info = loader.connect(args.wait)
|
|
if args.info:
|
|
print("device:")
|
|
for line in info.lines():
|
|
print(f" {line}")
|
|
fuse_bytes = fuse_override
|
|
if args.fuses or (args.update_loader and not info.patch_vector and fuse_bytes is None):
|
|
read = op_fuses(loader)
|
|
if fuse_bytes is None:
|
|
fuse_bytes = read
|
|
if args.update_loader:
|
|
state = args.state or args.update_loader + ".pbstate"
|
|
op_update_loader(loader, args.wait, args.update_loader, state, fuse_bytes)
|
|
if args.flash:
|
|
op_flash(loader, args.flash, args.erase_flash, not args.no_verify, fuse_bytes, args.force)
|
|
elif args.erase_flash:
|
|
op_erase_flash(loader)
|
|
if args.read_flash:
|
|
op_read_flash(loader, args.read_flash)
|
|
if args.verify_flash:
|
|
op_verify_flash(loader, args.verify_flash)
|
|
if args.eeprom:
|
|
op_eeprom(loader, args.eeprom, args.erase_eeprom, not args.no_verify)
|
|
elif args.erase_eeprom:
|
|
op_erase_eeprom(loader)
|
|
if args.read_eeprom:
|
|
op_read_eeprom(loader, args.read_eeprom)
|
|
if args.verify_eeprom:
|
|
op_verify_eeprom(loader, args.verify_eeprom)
|
|
if args.stay:
|
|
print("loader stays in its session (reset to leave)")
|
|
else:
|
|
loader.run_application()
|
|
print("application running")
|
|
finally:
|
|
port.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Error as error:
|
|
print(f"error: {error}", file=sys.stderr)
|
|
sys.exit(1)
|
|
except KeyboardInterrupt:
|
|
sys.exit(130)
|