avr/bootloader
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

tool: survive our own leftovers — drain the fresh port, shorten the identity read

Browse Source 
--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 <noreply@anthropic.com>
This commit is contained in:
BlackMark
2026-07-30 16:05:16 +02:00
parent 4362886c39
commit c8ac61779e
2 changed files with 145 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
pureboot/pureboot.py
View File

@@ -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:

121
test/test_handshake.py
View File

@@ -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