pureboot.py: bound the activation drain against a flooding target
The post-prompt settle loop in _handshake had no deadline, so a target that never falls quiet — a board stuck in a reset loop, whose UART-reset garbage carries a stray prompt byte — spun the tool forever. Bound it by the handshake deadline; a real loader still settles on its first quiet read. Regression: test/test_handshake.py (flood terminates, valid loader still connects). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -484,8 +484,13 @@ class Loader:
|
|||||||
self.port.write(knock)
|
self.port.write(knock)
|
||||||
knocks += 1
|
knocks += 1
|
||||||
if PROMPT in self.port.read_available(0.4):
|
if PROMPT in self.port.read_available(0.4):
|
||||||
|
# Settle: absorb a real loader's trailing bytes before asking
|
||||||
|
# for the identity. Bounded by the deadline so a target that
|
||||||
|
# never falls quiet — a board stuck in a reset loop, whose
|
||||||
|
# garbage carries a stray prompt — cannot spin here forever.
|
||||||
while self.port.read_available(0.3):
|
while self.port.read_available(0.3):
|
||||||
pass
|
if time.monotonic() > deadline:
|
||||||
|
break
|
||||||
self.port.write(b"b")
|
self.port.write(b"b")
|
||||||
try:
|
try:
|
||||||
# A version the tool cannot speak is the loader's own
|
# A version the tool cannot speak is the loader's own
|
||||||
|
|||||||
104
pureboot/test/test_handshake.py
Normal file
104
pureboot/test/test_handshake.py
Normal file
@@ -0,0 +1,104 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Host-tool activation handshake: it must not hang on a flooding target.
|
||||||
|
|
||||||
|
`_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
|
||||||
|
target that never falls quiet — a board stuck in a reset loop presents exactly
|
||||||
|
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.
|
||||||
|
|
||||||
|
Stdlib only; run with `python test/test_handshake.py`.
|
||||||
|
"""
|
||||||
|
import importlib.util
|
||||||
|
import pathlib
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
|
||||||
|
PB = pathlib.Path(__file__).resolve().parents[1] / "pureboot.py"
|
||||||
|
_spec = importlib.util.spec_from_file_location("pureboot", PB)
|
||||||
|
pb = importlib.util.module_from_spec(_spec)
|
||||||
|
_spec.loader.exec_module(pb)
|
||||||
|
|
||||||
|
P = F = 0
|
||||||
|
|
||||||
|
|
||||||
|
def check(name, ok):
|
||||||
|
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}")
|
||||||
|
|
||||||
|
|
||||||
|
class FloodPort:
|
||||||
|
"""A line that never falls quiet: read_available always returns bytes, and
|
||||||
|
they contain a prompt. No identity ever completes."""
|
||||||
|
|
||||||
|
def flush_input(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def write(self, data):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def read_available(self, wait):
|
||||||
|
time.sleep(0.01) # a real read waits; keep the busy loop off a core
|
||||||
|
return b"+\x00\xff"
|
||||||
|
|
||||||
|
def read_exact(self, count, timeout):
|
||||||
|
raise pb.Error("no identity")
|
||||||
|
|
||||||
|
|
||||||
|
class LoaderPort:
|
||||||
|
"""A well-behaved pureboot 5: one 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
|
||||||
|
|
||||||
|
def flush_input(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def write(self, data):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def read_available(self, wait):
|
||||||
|
self.reads += 1
|
||||||
|
return b"+" if self.reads == 1 else b"" # prompt once, then settle quiet
|
||||||
|
|
||||||
|
def read_exact(self, count, timeout):
|
||||||
|
self.exacts += 1
|
||||||
|
return b"\x05\x1e\x95\x0f" if self.exacts == 1 else b"+" # identity, then prompt
|
||||||
|
|
||||||
|
|
||||||
|
def terminates(port, wait, budget):
|
||||||
|
"""Run connect_autobaud in a thread; True if it returns/raises within
|
||||||
|
`budget` seconds rather than hanging."""
|
||||||
|
done = threading.Event()
|
||||||
|
|
||||||
|
def run():
|
||||||
|
try:
|
||||||
|
pb.Loader(port).connect_autobaud(wait)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
finally:
|
||||||
|
done.set()
|
||||||
|
|
||||||
|
threading.Thread(target=run, daemon=True).start()
|
||||||
|
return done.wait(budget)
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
# the hang: a flooding target must not spin the drain forever. With wait=0.5
|
||||||
|
# the whole handshake has to give up well inside a few seconds.
|
||||||
|
check("flooding target: handshake terminates, drain is bounded",
|
||||||
|
terminates(FloodPort(), wait=0.5, budget=4.0))
|
||||||
|
|
||||||
|
# the control: a real loader still connects and reads identity.
|
||||||
|
info = pb.Loader(LoaderPort()).connect_autobaud(2.0)
|
||||||
|
check("well-behaved loader still connects (version 5)", info.version == 5)
|
||||||
|
|
||||||
|
print(f"\n {P} passed, {F} failed")
|
||||||
|
return 1 if F else 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
raise SystemExit(main())
|
||||||
Reference in New Issue
Block a user