diff --git a/CMakeLists.txt b/CMakeLists.txt index da41176..c94039c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -162,6 +162,8 @@ if(PROJECT_IS_TOP_LEVEL) ${CMAKE_CURRENT_SOURCE_DIR}/pureboot/pureboot.py) add_test(NAME pureboot.handshake COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test/test_handshake.py) + add_test(NAME pureboot.updatelink + COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test/test_update_link.py) endif() # The protocol test flashes this fixture through the loader with the real diff --git a/pureboot/README.md b/pureboot/README.md index c5b4ecd..9388ac8 100644 --- a/pureboot/README.md +++ b/pureboot/README.md @@ -328,6 +328,23 @@ with any pureboot build — a re-timed window, a newer version — using the loader itself as its own staging loader. The image is the loader's own 512 bytes as a raw binary, or the Intel HEX the build emits beside it. +One thing the image cannot tell the host: **which link it speaks.** The update +works by entering copies of the *new* image (steps 3 and 4 below), so a build +made for another baud or another backend answers on that one and not on the +session's — and 512 bytes of position-independent code carry no header to read +it from. Where the new image's link differs, name it: + +```sh +# a 57600 fixed-baud resident, replaced by an autobaud build +pureboot.py --port … --baud 57600 --update-loader ab.bin --staged-autobaud +# …or by a 38400 build of the same backend +pureboot.py --port … --baud 57600 --update-loader sw38400.bin --staged-baud 38400 +``` + +The host retunes on the open port, so no DTR pulse resets the copy it is talking +to. Omit them against a changed link and the update stops after installing the +staging copy, saying so and naming this as the cause. + The preflight refuses an image built for another chip: the stamp every pureboot binary carries must resolve to the device's own geometry, and the error names both. Die revisions share their base signature and geometry, so their images @@ -348,10 +365,15 @@ are interchangeable — as the silicon is. content, and the state file is discarded. Every phase is idempotent and keyed off the actual flash state, so re-running -the same command after any interruption resumes and completes. The state file -carries the only bytes not recoverable from the device; losing it mid-update -still completes the update, and the staging region comes back by reflashing -the application. A boot-sectioned mega needs its fuses for the preflight — read +the same command after any interruption resumes and completes — with one +qualification, which is the link again: from step 2 on, the copy the re-run has +to reach is the *new* image, so a resumed run needs the same `--staged-*` as the +first one. On a patched-vector part step 3 also re-aims word 0 at the staging +copy, so after that point a reset reaches the new image's link and **only** that +one; a re-run on the resident's link finds nothing at all. The state file carries +the only bytes not recoverable from the device; losing it mid-update still +completes the update, and the staging region comes back by reflashing the +application. A boot-sectioned mega needs its fuses for the preflight — read from the device, or supplied with `--assume-fuses` where reading is impossible (simulators). @@ -425,6 +447,11 @@ Per chip preset, `ctest` runs: - `pureboot.handshake` — the host tool's activation must not hang on a target that never falls quiet: the drain after a prompt is bounded by the handshake deadline, and a well-behaved loader still connects; +- `pureboot.updatelink` — an update whose image changes the baud or the backend + must follow the staging copy onto *its* link, since that copy is the new image; + and where nothing was declared, the failure must name the link rather than + report a bare activation timeout, because by then the staging slot is written + and on a 1 KiB tiny that was the application; - `pureboot.planner` — the host tool's pure logic: programming orders and their recovery properties, the surgery, the staging composition, the boot-fuse decode, the update preflight over synthetic fuse bytes, and the repairing diff --git a/pureboot/pureboot.py b/pureboot/pureboot.py index 0d9d561..cf3f12e 100644 --- a/pureboot/pureboot.py +++ b/pureboot/pureboot.py @@ -146,6 +146,13 @@ class Progress: class PosixPort: """A raw serial port with deadline-based reads, over termios.""" + @staticmethod + def _speed(baud): + try: + return getattr(termios, f"B{baud}") + except AttributeError: + raise Error(f"unsupported baud rate {baud}") from None + def __init__(self, path, baud): self.fd = os.open(path, os.O_RDWR | os.O_NOCTTY) attrs = termios.tcgetattr(self.fd) @@ -153,14 +160,20 @@ class PosixPort: 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[4] = attrs[5] = self._speed(baud) attrs[6][termios.VMIN] = 0 attrs[6][termios.VTIME] = 0 termios.tcsetattr(self.fd, termios.TCSANOW, attrs) + self.baud = baud + + def set_baud(self, baud): + """Retune the port without closing it — the fd stays open, so no DTR + pulse and no reset. That matters: the only caller is mid-session with a + loader copy that a reset would throw away.""" + attrs = termios.tcgetattr(self.fd) + attrs[4] = attrs[5] = self._speed(baud) + termios.tcsetattr(self.fd, termios.TCSANOW, attrs) + self.baud = baud def close(self): os.close(self.fd) @@ -289,6 +302,7 @@ if os.name == "nt": # timeout would otherwise stay at the driver's default — which # may be "wait forever" — until the first read. self._deadline(_GAP_MS, 1000) + self.baud = baud except Error: # An open port outlives the exception otherwise, and a COM # handle is exclusive: the next attempt would meet its own @@ -296,6 +310,21 @@ if os.name == "nt": self.close() raise + def set_baud(self, baud): + """Retune the port on its live handle — SetCommState only, so the + handle is never reopened and DTR never drops. That matters: the only + caller is mid-session with a loader copy a reset would throw away.""" + if baud < 50: + raise Error(f"unsupported baud rate {baud}") + dcb = _DCB() + dcb.DCBlength = ctypes.sizeof(_DCB) + if not _k32.GetCommState(self.handle, ctypes.byref(dcb)): + _fail("cannot read the port state") + dcb.BaudRate = baud + if not _k32.SetCommState(self.handle, ctypes.byref(dcb)): + _fail(f"cannot retune the port to {baud} baud") + self.baud = baud + def close(self): _k32.CloseHandle(self.handle) @@ -457,6 +486,10 @@ class Loader: # Set once a session is established over an autobaud link, so a # re-entry after 'J' repeats the handshake that worked. self.autobaud = False + # The link this session is speaking. It moves when the host follows a + # staging copy built for another one (enter_copy). + self.baud = getattr(port, "baud", None) + self._link_declared = False def _read_identity(self): """The 'b' reply, in either of the two layouts a loader may send. @@ -664,12 +697,41 @@ class Loader: self.port.write(bytes((ord("J"), word_address & 0xFF, word_address >> 8))) self._expect_prompt() - def enter_copy(self, byte_address, wait): + def enter_copy(self, byte_address, wait, link=None): """Jump into the loader copy at `byte_address` and knock it — a slot - base is that copy's entry stub, so it can only land there.""" - autobaud = self.autobaud + base is that copy's entry stub, so it can only land there. + + `link` is that copy's own `(baud, autobaud)`, for when it is not this + session's. A staging copy *is* the new image, so it speaks the rate and + backend it was built for; the host has to be told which, because 512 + bytes of position-independent code carry no header to read it from. + Retuning goes through the open port, so no DTR pulse resets the copy that + is now running — and the session keeps the new link afterwards, since + every later jump lands in the same image. + """ + baud, autobaud = link if link is not None else (self.baud, self.autobaud) + if link is not None: + self._link_declared = True self.jump(byte_address // 2) - return self.connect_autobaud(wait) if autobaud else self.connect(wait) + if baud is not None and baud != self.baud: + self.port.set_baud(baud) + self.baud = baud + self.autobaud = autobaud + try: + return self.connect_autobaud(wait) if autobaud else self.connect(wait) + except Error as unheard: + if self._link_declared: + raise + # The bare activation timeout sends the operator to look at wiring, + # while on a patched-vector part the application region is already + # gone. Name the one cause that fits: the copy answers on its own + # link, not the resident's. + raise Error( + f"the copy at {byte_address:#06x} did not answer on this session's " + f"link ({baud} Bd, {'autobaud' if autobaud else 'fixed baud'}). An " + f"image built for another baud or backend speaks that one instead — " + f"say which with --staged-baud / --staged-autobaud" + ) from unheard def run_application(self): self.jump(self.info.app_entry_word) @@ -1017,10 +1079,16 @@ def patch_word0(loader, page0, target_base): return bytes(patched) -def op_update_loader(loader, wait, path, state_path, fuse_bytes): +def op_update_loader(loader, wait, path, state_path, fuse_bytes, staged_link=None): """Replace the resident loader with `path`, using the loader as its own staging loader. Every phase is idempotent and keyed off the flash state, - so a re-run resumes; the state file carries what the staging slot held.""" + so a re-run resumes; the state file carries what the staging slot held. + + `staged_link` is the new image's own `(baud, autobaud)` where it differs from + this session's — the copies the host enters *are* that image, so they answer + on its link and not the resident's. Note what this does to the idempotence + above: once the staging copy is installed, the resumable state is only + reachable on the new link, so a re-run has to name it too.""" info = loader.info image = loader_image(path) for warning in update_preflight(image, info, fuse_bytes): @@ -1065,7 +1133,7 @@ def op_update_loader(loader, wait, path, state_path, fuse_bytes): # routes through the resident, word 0 is re-aimed at the staging copy for # the rewrite, so a power loss mid-rewrite still resets into a loader. verbose(f"entering the staging copy at {info.stage:#06x}") - loader.enter_copy(info.stage, wait) + loader.enter_copy(info.stage, wait, link=staged_link) redirect = info.patch_vector and info.stage != 0 if redirect: verbose("word 0 re-aimed at the staging copy for the rewrite") @@ -1302,6 +1370,15 @@ def main(): 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)") + # The update enters the staging copy, which is the new image and so speaks + # the link *it* was built for. Nothing in the image says which, so where it + # differs from this session's these name it and the host follows. + parser.add_argument("--staged-baud", metavar="BD", type=int, + help="the baud the --update-loader image was built for, where it " + "differs from --baud") + parser.add_argument("--staged-autobaud", action=argparse.BooleanOptionalAction, default=None, + help="whether that image is an autobaud build, where it differs " + "from --autobaud") 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") @@ -1351,7 +1428,14 @@ def main(): 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) + staged_link = None + if args.staged_baud is not None or args.staged_autobaud is not None: + staged_link = ( + args.staged_baud if args.staged_baud is not None else args.baud, + args.staged_autobaud if args.staged_autobaud is not None else args.autobaud, + ) + op_update_loader(loader, args.wait, args.update_loader, state, fuse_bytes, + staged_link) if args.flash: op_flash(loader, args.flash, args.erase_flash, not args.no_verify, fuse_bytes, args.force) elif args.erase_flash: diff --git a/test/test_update_link.py b/test/test_update_link.py new file mode 100755 index 0000000..bfb29c5 --- /dev/null +++ b/test/test_update_link.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +"""Self-update across a link change: the host must follow the staging copy. + +`--update-loader` installs the new image in the staging slot and then *enters +it* to have it rewrite the resident. That copy is the new image, so it speaks the +new image's baud and backend — but the host was talking to the *resident*. Where +the two differ, the host kept knocking at the old rate in the old mode, the +staging copy never answered, and the update stranded: staging installed, resident +untouched, and on a 1 KiB tiny the application region (which *is* the staging +slot there) already gone. + +The wire cannot be probed for this — 512 bytes of position-independent code carry +no header saying what rate they were built for — so the operator declares it, and +a mismatch with nothing declared has to say so instead of reporting a bare +timeout. + +Stdlib only, no device: host-tool logic, so it runs on every chip's preset beside +pureboot.planner. +""" +import importlib.util +import pathlib + +PB = pathlib.Path(__file__).resolve().parents[1] / "pureboot" / "pureboot.py" +_spec = importlib.util.spec_from_file_location("pureboot", PB) +pb = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(pb) + +IDENTITY = b"\x05\x1e\x95\x0f" # pureboot 5 + m328p signature +P = F = 0 + + +def check(name, ok, detail=""): + global P, F + P, F = P + (1 if ok else 0), F + (0 if ok else 1) + print(f" [{'PASS' if ok else 'FAIL'}] {name}" + (f" — {detail}" if detail else "")) + + +class TwoLinkPort: + """A board whose resident and staging copy answer on different links. + + Only the rate currently set decides who can be heard, which is the physical + truth: a loader's bit timing is a cycle count, so a copy built for another + rate is unreadable until the host retunes. The knock bytes carry the mode, so + a backend mismatch is caught the same way. + """ + + def __init__(self, resident=(57600, False), staged=(38400, False)): + self.resident, self.staged = resident, staged + self.baud = resident[0] + self.entered = False # a 'J' has handed control to the staging copy + self.switches = [] # every retune the host asked for + self.pending = bytearray() # what the device has queued to send + + # --- the part under test needs this to exist at all + def set_baud(self, baud): + self.baud = baud + self.switches.append(baud) + + def flush_input(self): + self.pending.clear() + + def _audible(self, knock=None): + baud, autobaud = self.staged if self.entered else self.resident + if self.baud != baud: + return False + if knock is None: + return True + return knock == (bytes((pb.CALIBRATE, ord("p"))) if autobaud else b"pb") + + def write(self, data): + data = bytes(data) + if data[:1] == b"J" and len(data) == 3: + # The resident acks the jump, then control moves to the copy. + if self._audible(): + self.pending += pb.PROMPT + self.entered = True + elif data in (b"pb", bytes((pb.CALIBRATE, ord("p")))): + if self._audible(data): + self.pending += pb.PROMPT + elif data == b"b": + if self._audible(): + self.pending += IDENTITY + pb.PROMPT + + def read_available(self, wait): + out, self.pending = bytes(self.pending), bytearray() + return out + + def read_exact(self, count, timeout): + if len(self.pending) < count: + raise pb.Error(f"timeout: got {len(self.pending)} of {count} bytes") + out, self.pending = bytes(self.pending[:count]), self.pending[count:] + return out + + +def connected(port): + """A Loader already in session with the resident.""" + loader = pb.Loader(port) + loader.connect(2.0) + return loader + + +def main(): + # The control first: where the staged image keeps the resident's link, the + # flow works and needs no retune. This is the case that always passed, and + # it is what made the bug look like "self-update is broken" rather than + # "self-update cannot change the link". + port = TwoLinkPort(resident=(57600, False), staged=(57600, False)) + loader = connected(port) + try: + loader.enter_copy(0x7C00, 2.0) + check("same link: staging copy entered", True) + except pb.Error as error: + check("same link: staging copy entered", False, str(error)) + + # A baud change, declared. The host must retune before knocking. + port = TwoLinkPort(resident=(57600, False), staged=(38400, False)) + loader = connected(port) + try: + loader.enter_copy(0x7C00, 2.0, link=(38400, False)) + check("baud change declared: entered after retuning", 38400 in port.switches, + f"switches={port.switches}") + except (pb.Error, TypeError) as error: + check("baud change declared: entered after retuning", False, repr(error)) + + # A backend change, declared: the knock itself has to become the calibration + # pulse, or an autobaud staging copy never hears a thing. + port = TwoLinkPort(resident=(57600, False), staged=(57600, True)) + loader = connected(port) + try: + loader.enter_copy(0x7C00, 2.0, link=(57600, True)) + check("backend change declared: entered as autobaud", True) + except (pb.Error, TypeError) as error: + check("backend change declared: entered as autobaud", False, repr(error)) + + # Nothing declared against a changed link: it still cannot work, but the + # error has to name the cause. A bare "no answer" sent the operator looking + # at the wiring while the application region sat erased. + port = TwoLinkPort(resident=(57600, False), staged=(38400, False)) + loader = connected(port) + try: + loader.enter_copy(0x7C00, 0.3) + check("undeclared mismatch: reported", False, "unexpectedly succeeded") + except pb.Error as error: + text = str(error).lower() + check("undeclared mismatch: error names the link, not just a timeout", + "link" in text or "baud" in text or "backend" in text, str(error)) + except TypeError as error: + check("undeclared mismatch: error names the link, not just a timeout", + False, repr(error)) + + # The resident's own link must be restored for the caller: a declared + # staging link is for the copy, and the tool talks to the new resident after. + port = TwoLinkPort(resident=(57600, False), staged=(38400, False)) + loader = connected(port) + try: + loader.enter_copy(0x7C00, 2.0, link=(38400, False)) + check("session records the link it is now speaking", loader.baud == 38400, + f"loader.baud={getattr(loader, 'baud', None)}") + except (pb.Error, TypeError, AttributeError) as error: + check("session records the link it is now speaking", False, repr(error)) + + print(f"\n {P} passed, {F} failed") + return 1 if F else 0 + + +if __name__ == "__main__": + raise SystemExit(main())