BlackMark
37f9747e26
check_walk_region() refuses an image writing into the span reset crosses to reach the loader, and a plain --flash never ran it: the fuses were read for --fuses and for --update-loader, so the tool read them to protect the loader and never to protect the reset path. It cost an ssd1306 board - an application grown through 0x7c00 on a 328P with hfuse 0xdc, reset landing mid-function, an ICE the only way back. The check now fetches its own input, so the operation that asks for no fuses cannot skip it and neither can a direct API caller: pbdirty and pbmute call op_flash() as a library and are guarded without a line changing in them. Fuses that cannot be read are a refusal naming --assume-fuses and --force, because unknown is not empty. The rig had to stop lying first. The fuse read is an LPM diverted by BLBSET, which simavr executes straight out of flash with no hook, so a fuse read answered flash bytes and --fuses had been printing them on every chip. The runner models the diversion at the SPMCSR write, -f states the profile - and stores the register itself, since a registered handler replaces simavr's store and would otherwise swallow every SPM command on the cores where nothing else watches it. pureboot.walk reproduces the brick: the unfixed tool writes 249 pages through 0x7c00 in silence, the fixed one refuses. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
368 lines
18 KiB
Python
368 lines
18 KiB
Python
#!/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()
|