BlackMark
11707e0b06
A prompt byte alone does not: one left over from a previous session can still be in the pipeline while the port opening resets the device into a fresh window, where the bare command that follows is discarded. Each attempt is now the whole handshake, retried until the block comes back. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
350 lines
17 KiB
Python
350 lines
17 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)
|
|
raw = bytes((0x50, 0x42, pb.NEWEST_LOADER if version is None else version,
|
|
*signature, page & 0xFF, wire_base & 0xFF, wire_base >> 8, 0, 2, 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 embedded info block: found in a synthetic binary, absent in noise.
|
|
binary = bytes((0xAA,)) * 10 + tiny.raw + bytes((0xBB,)) * 10
|
|
found = pb.image_info(binary)
|
|
if found is None or found.raw != tiny.raw:
|
|
fail("image_info misses the embedded block")
|
|
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)
|
|
pb.check_walk_region(deep, mega, None, False) # fuses unknown: no check
|
|
|
|
# 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()
|