From c8ac61779e704bba1b21f9f6c5d1b6b58a169a41 Mon Sep 17 00:00:00 2001 From: BlackMark Date: Thu, 30 Jul 2026 16:05:16 +0200 Subject: [PATCH] =?UTF-8?q?tool:=20survive=20our=20own=20leftovers=20?= =?UTF-8?q?=E2=80=94=20drain=20the=20fresh=20port,=20shorten=20the=20ident?= =?UTF-8?q?ity=20read?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --stay leaves the loader's final prompt in the USB pipeline; a fresh invocation on a board that resets when its port opens then flushes too early, trusts the stale prompt, and spends the new activation window on a 2-second identity read against a device that never heard its knock — collecting the application's banner as an unknown signature. Three host-side moves, no device bytes: the line is drained until quiet (bounded, 250 ms) before the port's first knock — once per port, since a mid-session re-knock faces no foreign bytes and its own window is already burning; the identity read_exact drops 2.0 to 0.5 s, dozens of times the worst real answer, so any false prompt match leaves room for the retry that already works; and the tool version drifts to 8. The StaleDTRPort fixture models the whole moment — stale prompt in transit, reset holding the device off the line, a finite window, the banner — red against the old tool in exactly the field shape (unknown signature from banner bytes), green now; LoaderPort answers its prompt to the knock rather than to a read count, which the drain exposed as a call-order coupling. Co-Authored-By: Claude Fable 5 --- pureboot/pureboot.py | 36 ++++++++++-- test/test_handshake.py | 121 ++++++++++++++++++++++++++++++++++++++--- 2 files changed, 145 insertions(+), 12 deletions(-) diff --git a/pureboot/pureboot.py b/pureboot/pureboot.py index ec3126e..17c2f22 100644 --- a/pureboot/pureboot.py +++ b/pureboot/pureboot.py @@ -26,7 +26,7 @@ else: import termios PROMPT = b"+" -VERSION = 7 # this tool's own version — free to drift from a loader's +VERSION = 8 # this tool's own version — free to drift from a loader's # The loader versions this tool can drive. A pureboot version implies its wire # protocol, which carries no number of its own, so this window is where that # map lives: the tool keeps a decoder for every generation in it (1–4 speak @@ -546,6 +546,11 @@ 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 pre-knock drain runs once per port: the bytes it exists for are + # leftovers from before this process opened the port. Re-knocks later + # in the same session must not pay it — a fresh activation window is + # already burning while they wait. + self._line_drained = 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) @@ -555,10 +560,16 @@ class Loader: """The 'b' reply, in either of the two layouts a loader may send. pureboot 5 answers with its version and the signature; older loaders answer with a 12-byte block. The version byte cannot be mistaken for - the older block's 'P', so four bytes are enough to tell them apart.""" - head = self.port.read_exact(4, 2.0) + the older block's 'P', so four bytes are enough to tell them apart. + + The timeout is short on purpose: a real answer follows the prompt + within a frame time or two, so half a second is dozens of times the + worst case — while a *false* prompt match (a stale byte, reset + garbage) makes this read collect noise, and every second spent on it + comes out of the activation window the retry needs.""" + head = self.port.read_exact(4, 0.5) if head[0:2] == b"PB": - return Info(head + self.port.read_exact(8, 2.0)) + return Info(head + self.port.read_exact(8, 0.5)) return Info.from_identity(head) def _handshake(self, wait, knock, what): @@ -569,8 +580,23 @@ class Loader: into a fresh window, where a command without its knock is discarded. Each attempt is therefore the whole handshake. This also converges into an already-live session: the knock bytes are ignored there and the - drain absorbs whatever they produced.""" + drain absorbs whatever they produced. + + Before the port's first knock ever, the line is drained until quiet: a + prompt from a previous session (`--stay`) can still be in the USB + pipeline when the port opens, where a flush cannot clear what has not + arrived yet — and on a board that resets when its port opens, trusting + that stale byte would spend the fresh activation window reading noise + from a device that never heard the knock. Once only, and bounded: + later re-knocks in this session face no foreign leftovers, and their + own window is already burning.""" deadline = time.monotonic() + wait + if not self._line_drained: + self._line_drained = True + drain = time.monotonic() + 0.25 + while self.port.read_available(0.05): + if time.monotonic() > drain: + break knocks = 0 refusal = None while True: diff --git a/test/test_handshake.py b/test/test_handshake.py index e045579..97af003 100644 --- a/test/test_handshake.py +++ b/test/test_handshake.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Host-tool activation handshake: it must not hang on a flooding target. +"""Host-tool activation handshake: bounded against a line that misbehaves. `_handshake` drains the line after it sees a prompt, to absorb a real loader's trailing bytes before it asks for the identity. That drain must be bounded: a @@ -8,6 +8,13 @@ this, ~60 reboots/s of UART-reset garbage in which a stray 0x2b reads as a prompt — otherwise spins the tool forever. Regression for that hang, plus a control that a well-behaved loader still connects. +The handshake must also survive its own leftovers: after `--stay` the loader's +final prompt can still be in the USB pipeline when the next invocation opens +the port, and on a board wired to reset on open, that opening starts a fresh +activation window the stale prompt then betrays — the tool commits to an +identity read against a device that never heard its knock, and what it finally +collects is the application's banner. StaleDTRPort is that moment as a port. + Stdlib only, no device: host-tool logic, so it runs on every chip's preset beside pureboot.planner. """ @@ -49,27 +56,117 @@ class FloodPort: class LoaderPort: - """A well-behaved pureboot 5: one prompt to the knock, then quiet, then the + """A well-behaved pureboot 5: a prompt to the knock, then quiet, then the slim identity (version 5 + m328p signature) and a closing prompt.""" def __init__(self): - self.reads = self.exacts = 0 + self.pending = b"" + self.exacts = 0 def flush_input(self): - pass + self.pending = b"" def write(self, data): - pass + if b"p" in data: + self.pending = b"+" # the prompt answers the knock, nothing else def read_available(self, wait): - self.reads += 1 - return b"+" if self.reads == 1 else b"" # prompt once, then settle quiet + data, self.pending = self.pending, b"" + return data def read_exact(self, count, timeout): self.exacts += 1 return b"\x05\x1e\x95\x0f" if self.exacts == 1 else b"+" # identity, then prompt +class StaleDTRPort: + """`--stay`, then a fresh invocation on a board that resets when its port + opens. Three facts of that moment, all timed from the open: the previous + session's final prompt is still in transit and lands only after the + opening flush has already run; the reset holds the device off the line + at first, eating anything written before it completes; and the fresh + window is finite — once it expires the application boots and prints a + banner whose bytes are what a pending identity read collects. A + handshake that trusts the stale prompt spends the whole window waiting + on a device that never heard its knock; one that drains the line first + knocks into the real window and connects.""" + + STALE_AT = 0.02 # the leftover prompt becomes visible (post-flush) + READY_AT = 0.05 # reset complete, activation window opens + WINDOW = 1.0 # window length; expiry boots the application + + def __init__(self): + self.t0 = time.monotonic() + # (visible-from, bytes): the line as a timed queue. + self.queue = [(self.t0 + self.STALE_AT, b"+")] + self.armed = False # a 'p' heard inside the window arms 'b' + self.booted = False + + def _boot_check(self): + if not self.booted and time.monotonic() > self.t0 + self.READY_AT + self.WINDOW: + self.booted = True + self.queue.append((self.t0 + self.READY_AT + self.WINDOW, + b"W r libavr tempmon\r\n")) + + def _visible(self): + self._boot_check() + now = time.monotonic() + return b"".join(d for t, d in self.queue if t <= now) + + def _consume(self, n): + now = time.monotonic() + left = [] + for t, d in self.queue: + if t <= now and n: + take = min(n, len(d)) + d = d[take:] + n -= take + if d: + left.append((t, d)) + self.queue = left + + def flush_input(self): + self._consume(len(self._visible())) + + def write(self, data): + self._boot_check() + now = time.monotonic() + if now < self.t0 + self.READY_AT or self.booted: + return # still in reset, or the application owns the line + if b"p" in data: + self.armed = True + self.queue.append((now + 0.01, b"+")) + if b"b" in data and self.armed: + # The slim identity (version 5 + m328p signature) and a prompt. + self.queue.append((now + 0.01, b"\x05\x1e\x95\x0f+")) + + def read_available(self, wait): + deadline = time.monotonic() + wait + while True: + data = self._visible() + if data: + self._consume(len(data)) + return data + if time.monotonic() >= deadline: + return b"" + time.sleep(0.005) + + def read_exact(self, count, timeout): + deadline = time.monotonic() + timeout + data = b"" + while len(data) < count: + visible = self._visible() + if visible: + take = visible[:count - len(data)] + self._consume(len(take)) + data += take + elif time.monotonic() >= deadline: + raise pb.Error(f"timeout: got {len(data)} of {count} bytes") + else: + time.sleep(0.005) + return data + + def terminates(port, wait, budget): """Run connect_autobaud in a thread; True if it returns/raises within `budget` seconds rather than hanging.""" @@ -97,6 +194,16 @@ def main(): info = pb.Loader(LoaderPort()).connect_autobaud(2.0) check("well-behaved loader still connects (version 5)", info.version == 5) + # the stale prompt: a --stay leftover plus reset-on-open must not burn the + # fresh window — the pre-knock drain absorbs it and the first real knock + # lands inside the window. + try: + stale_ok = pb.Loader(StaleDTRPort()).connect(2.5).version == 5 + except pb.Error as failed: + print(f" ({failed})") + stale_ok = False + check("stale --stay prompt + reset-on-open: connects in the fresh window", stale_ok) + print(f"\n {P} passed, {F} failed") return 1 if F else 0