diff --git a/README.md b/README.md index feb85b2..e118254 100644 --- a/README.md +++ b/README.md @@ -191,7 +191,9 @@ socat -,raw,echo=0 TCP:172.17.32.17:35240 > on the same host as defib (or close to it on a LAN) is recommended. The `-t` flag auto-detects the post-boot mode: -- **Normal U-Boot shell** (e.g. hi3516ev300): raw terminal passthrough — type commands directly +- **Normal U-Boot shell** (e.g. hi3516ev300): a two-way serial terminal — your + keystrokes go to the board and its output comes back. Ctrl-C exits the + terminal rather than being sent on to U-Boot. - **Download command mode** (e.g. hi3516av200): interactive `defib>` prompt that wraps commands in HiSilicon's XHEAD/XCMD protocol, enabling flash operations on devices that enter `download_process()` after serial boot ```bash diff --git a/src/defib/cli/app.py b/src/defib/cli/app.py index 5ab3cdf..638226e 100644 --- a/src/defib/cli/app.py +++ b/src/defib/cli/app.py @@ -396,7 +396,16 @@ def on_sigint(*_: object) -> None: if output == "human": console.print("[dim]--- Session closed ---[/dim]") else: - # Normal U-Boot shell — raw terminal passthrough + # Normal U-Boot shell — a real terminal, in both directions. + # + # It used to be one direction. The loop drained the port onto the + # screen and nothing read the keyboard, so `-t` handed you a live + # U-Boot prompt that ignored everything you typed + # (OpenIPC/firmware#2381). Keys are polled between serial reads + # rather than awaited on their own task: no thread, no executor, + # and nothing that can stall the loop holding the serial link. + from defib.cli.keyboard import raw_terminal, read_available_keys + if output == "human": console.print("[dim]--- Terminal mode (Ctrl-C to exit) ---[/dim]") @@ -409,13 +418,20 @@ def on_sigint(*_: object) -> None: signal.signal(signal.SIGINT, on_sigint) try: - while not stop: - try: - data = await transport.read(256, timeout=0.1) - _sys.stdout.buffer.write(data) - _sys.stdout.buffer.flush() - except Exception: - pass + with raw_terminal(): + while not stop: + typed = read_available_keys() + if typed: + try: + await transport.write(typed) + except Exception: + break + try: + data = await transport.read(256, timeout=0.1) + _sys.stdout.buffer.write(data) + _sys.stdout.buffer.flush() + except Exception: + pass finally: signal.signal(signal.SIGINT, signal.SIG_DFL) if output == "human": diff --git a/src/defib/cli/keyboard.py b/src/defib/cli/keyboard.py new file mode 100644 index 0000000..d872c10 --- /dev/null +++ b/src/defib/cli/keyboard.py @@ -0,0 +1,157 @@ +"""Keystrokes for the serial terminal. + +`defib burn -t` advertised a U-Boot console and delivered a viewer: the +terminal-mode loop read the port and wrote the screen, and nothing anywhere +read the keyboard. The reporter in OpenIPC/firmware#2381 reached a live +`OpenIPC #` prompt on a camera whose flash they still had to rewrite, and it +answered none of what they typed. + +Both readers here are non-blocking on purpose. They are polled from the same +loop that drains the serial port, so no thread and no executor is involved and +nothing can stall the event loop -- which during a recovery is also holding the +serial link to the board. +""" + +from __future__ import annotations + +import logging +import os +import sys +from contextlib import contextmanager +from typing import Any, Iterator, Protocol + +logger = logging.getLogger(__name__) + +# Windows reports a special key (arrows, function keys, keypad) as a marker +# byte followed by a scan code. Neither means anything to U-Boot, and passing +# them through types garbage at the prompt. +_WINDOWS_SPECIAL_PREFIXES = (b"\x00", b"\xe0") + +# Most one poll may collect before handing the loop back. A person types a +# few bytes; a pipe never stops, and draining it until it pauses would keep +# the serial read and the stop flag waiting for as long as the producer runs. +_MAX_BYTES_PER_POLL = 4096 + + +class _HasFileno(Protocol): + """Anything a terminal can be configured on -- in practice sys.stdin.""" + + def fileno(self) -> int: + ... + + +@contextmanager +def raw_terminal(stream: _HasFileno | None = None) -> Iterator[None]: + """Send keystrokes as they are typed, not a line at a time. + + Also turns off local echo: a serial console echoes back what it received, + so echoing locally as well shows every character twice. + + Ctrl-C is deliberately left as the interrupt (cbreak keeps ISIG), because + that is how terminal mode has always been exited and the banner says so. + A no-op where there is no terminal to configure -- Windows, a pipe, a + captured stdin under pytest -- so callers need no platform branch. + """ + stream = stream if stream is not None else sys.stdin + if sys.platform == "win32": + yield + return + try: + import termios + import tty + + fd = stream.fileno() + saved = termios.tcgetattr(fd) + except Exception: + yield + return + try: + tty.setcbreak(fd) + yield + finally: + _restore_terminal(fd, saved) + + +def read_available_keys() -> bytes: + """Whatever has been typed since the last call, or b"" if nothing has. + + Never waits. A caller polling this between serial reads keeps typing + responsive without giving up the loop. + """ + if sys.platform == "win32": + return _read_windows() + return _read_posix() + + +def _read_windows() -> bytes: + try: + import msvcrt + except ImportError: # pragma: no cover - only absent off Windows + return b"" + + out = bytearray() + while len(out) < _MAX_BYTES_PER_POLL and msvcrt.kbhit(): # type: ignore[attr-defined] + char = msvcrt.getch() # type: ignore[attr-defined] + if char in _WINDOWS_SPECIAL_PREFIXES: + msvcrt.getch() # type: ignore[attr-defined] # drop the scan code + continue + out += char + return bytes(out) + + +def _read_posix() -> bytes: + import select + + try: + fd = sys.stdin.fileno() + except (AttributeError, OSError, ValueError): + return b"" + + out = bytearray() + while len(out) < _MAX_BYTES_PER_POLL: + try: + ready, _, _ = select.select([fd], [], [], 0) + except (OSError, ValueError): + break + if not ready: + break + try: + chunk = os.read(fd, min(256, _MAX_BYTES_PER_POLL - len(out))) + except OSError: + break + if not chunk: # EOF on a pipe + break + out += chunk + return bytes(out) + + +def _restore_terminal(fd: int, saved: Any) -> None: + """Put the terminal back the way it was, and say so if that fails. + + This is the only thing between a failed restore and a shell with no echo + and no line editing. Discarding the failure -- which is what this used to + do -- leaves the operator with a broken terminal and nothing to explain it. + + TCSADRAIN waits for pending output and so can be the half that fails; it is + the right first choice because it does not truncate what the board was + printing, and TCSANOW is worth trying before giving up. + + Never raises. The body may already be unwinding with the error that + actually matters, and a cleanup failure must not replace it. + """ + import termios + + for when in (termios.TCSADRAIN, termios.TCSANOW): + try: + termios.tcsetattr(fd, when, saved) + return + except Exception as exc: # noqa: PERF203 - the retry is the point + logger.debug("tcsetattr(%s) failed: %s", when, exc) + + # stderr rather than logging alone: this has to reach someone whose + # terminal has just stopped echoing, whatever their logging setup is. + print( + "defib could not restore your terminal settings. " + "Run `stty sane` (you may have to type it blind) to get echo back.", + file=sys.stderr, + ) diff --git a/src/defib/network/ip_manager.py b/src/defib/network/ip_manager.py index e65de7a..efff0a5 100644 --- a/src/defib/network/ip_manager.py +++ b/src/defib/network/ip_manager.py @@ -273,6 +273,13 @@ def _windows_interfaces() -> list[str]: logger.warning("Could not list adapters with netsh: %s", exc) return [] + # Check the exit status, as the async twin does. Without this a netsh that + # failed left us parsing empty stdout and reporting "no adapters", which + # looks like a host with no network rather than a command that did not run. + if proc.returncode != 0: + logger.warning("netsh could not list adapters: %s", proc.stderr.strip()) + return [] + return _parse_netsh_interfaces(proc.stdout) diff --git a/src/defib/profiles/loader.py b/src/defib/profiles/loader.py index 748fe70..0f76f97 100644 --- a/src/defib/profiles/loader.py +++ b/src/defib/profiles/loader.py @@ -68,7 +68,7 @@ def load_profile(chip_name: str, profiles_dir: Path | None = None) -> SoCProfile if not profile_path.exists(): raise FileNotFoundError(f"No profile found for chip: {current}") - content = profile_path.read_text().strip() + content = profile_path.read_text(encoding="utf-8").strip() # Check if this is an alias (single token ending in .json) tokens = content.split() @@ -141,7 +141,7 @@ def list_variants(chip_name: str, profiles_dir: Path | None = None) -> list[str] profile_path = profiles_dir / f"{current}.json" if not profile_path.exists(): return [] - content = profile_path.read_text().strip() + content = profile_path.read_text(encoding="utf-8").strip() tokens = content.split() if len(tokens) == 1 and tokens[0].endswith(".json"): current = tokens[0][:-5] diff --git a/tests/test_agent_binary_lookup.py b/tests/test_agent_binary_lookup.py index 251a20d..55ad1da 100644 --- a/tests/test_agent_binary_lookup.py +++ b/tests/test_agent_binary_lookup.py @@ -61,7 +61,7 @@ def test_every_chip_the_agent_makefile_builds_is_reachable(self): """The map is what `defib agent` will accept; the Makefile is the truth.""" makefile = ( Path(__file__).parent.parent / "agent" / "Makefile" - ).read_text() + ).read_text(encoding="utf-8") built = { line.split("$(SOC),")[1].split(")")[0] for line in makefile.splitlines() diff --git a/tests/test_ip_manager_windows.py b/tests/test_ip_manager_windows.py index 9cf26fb..bb3e587 100644 --- a/tests/test_ip_manager_windows.py +++ b/tests/test_ip_manager_windows.py @@ -92,16 +92,21 @@ def test_windows_does_not_go_through_if_nameindex(self, monkeypatch): @pytest.mark.skipif( not sys.platform.startswith("win"), reason="needs a real netsh", ) - def test_real_netsh_yields_at_least_one_adapter(self): - """The mocks above pin the parser; this pins it against the real thing. - - CI runs this matrix on windows-latest, so the one claim the fix rests - on -- that netsh can be asked for names netsh will accept -- is checked - on a real Windows host rather than only against a captured table. + def test_real_netsh_runs_and_parses_cleanly(self): + """The mocks above pin the parser; this runs it against the real thing. + + What is asserted is ours: netsh is callable, its output parses, and + nothing that is not an adapter name comes back. Whether this host has + adapters to list is not. An earlier version of this test demanded at + least one and duly failed on a runner that listed none -- which said + nothing about the parser and blocked a merge for it. """ - assert _windows_interfaces(), ( - "netsh listed no adapters; the parser or the command has drifted" - ) + names = _windows_interfaces() + assert isinstance(names, list) + assert all(isinstance(n, str) and n.strip() for n in names) + # A header row surviving the shape test is the failure mode that + # matters, and it would show up here whatever the host's adapters are. + assert not any("Interface Name" in n for n in names) class TestAddressAlreadyThere: diff --git a/tests/test_profiles_usb_recovery.py b/tests/test_profiles_usb_recovery.py index 2fe2f70..4b1b390 100644 --- a/tests/test_profiles_usb_recovery.py +++ b/tests/test_profiles_usb_recovery.py @@ -118,7 +118,7 @@ def test_idblock_stays_at_its_fixed_offset(self): def test_profile_json_is_minimal(self): """A USB profile carrying UART bytecode would mean someone copied the wrong template.""" - data = json.loads((PROFILES_DIR / "rv1106.json").read_text()) + data = json.loads((PROFILES_DIR / "rv1106.json").read_text(encoding="utf-8")) assert not {"DDRSTEP0", "PRESTEP0", "ADDRESS", "FILELEN"} & set(data) diff --git a/tests/test_terminal_keyboard.py b/tests/test_terminal_keyboard.py new file mode 100644 index 0000000..eba7eca --- /dev/null +++ b/tests/test_terminal_keyboard.py @@ -0,0 +1,284 @@ +"""Terminal mode has to carry keystrokes, not just serial output. + +OpenIPC/firmware#2381: `defib burn -b -t` put a reporter at a live `OpenIPC #` +prompt on a camera whose flash still had to be rewritten, and the prompt +answered nothing they typed. The loop read the port and wrote the screen; no +code path anywhere in the CLI read stdin. README.md advertised the opposite -- +"raw terminal passthrough — type commands directly". +""" +from __future__ import annotations + +import os +import sys +from pathlib import Path + +import pytest + +from defib.cli import keyboard +from defib.cli.keyboard import ( + _MAX_BYTES_PER_POLL, + _read_posix, + _restore_terminal, + raw_terminal, + read_available_keys, +) + + +class TestNothingTypedIsNothingSent: + def test_an_idle_keyboard_reads_empty(self, monkeypatch): + monkeypatch.setattr(keyboard.sys, "platform", "linux") + monkeypatch.setattr("select.select", lambda *a, **k: ([], [], [])) + assert read_available_keys() == b"" + + def test_a_stdin_without_a_descriptor_is_not_an_error(self, monkeypatch): + """pytest replaces stdin with an object that has no fileno().""" + monkeypatch.setattr(keyboard.sys, "platform", "linux") + + class NoFileno: + def fileno(self): + raise OSError("captured") + + monkeypatch.setattr(keyboard.sys, "stdin", NoFileno()) + assert read_available_keys() == b"" + + +@pytest.mark.skipif( + sys.platform == "win32", + reason="select() takes only sockets on Windows, and this path never runs there", +) +class TestPosixReading: + """The POSIX reader, on the platforms it actually runs on. + + `read_available_keys` dispatches on the platform, so `_read_posix` is never + reached on Windows. Forcing it there tests nothing and fails for a reason + that is not a defect: Windows `select()` accepts sockets only, so the pipe + these tests use is rejected and the read comes back empty. + """ + + def test_what_was_typed_comes_back(self, monkeypatch): + r, w = os.pipe() + os.write(w, b"sf probe 0\n") + os.close(w) + monkeypatch.setattr(keyboard.sys, "platform", "linux") + monkeypatch.setattr(keyboard.sys, "stdin", _Fd(r)) + try: + assert _read_posix() == b"sf probe 0\n" + finally: + os.close(r) + + def test_eof_on_a_pipe_ends_the_read(self, monkeypatch): + r, w = os.pipe() + os.close(w) # immediate EOF + monkeypatch.setattr(keyboard.sys, "platform", "linux") + monkeypatch.setattr(keyboard.sys, "stdin", _Fd(r)) + try: + assert _read_posix() == b"" + finally: + os.close(r) + + +class TestWindowsReading: + def test_keys_are_drained_while_the_buffer_has_any(self, monkeypatch): + monkeypatch.setattr(keyboard.sys, "platform", "win32") + monkeypatch.setitem(sys.modules, "msvcrt", _FakeMsvcrt(list(b"reset\r"))) + assert read_available_keys() == b"reset\r" + + def test_a_special_key_is_dropped_with_its_scan_code(self, monkeypatch): + """An arrow key is a marker plus a scan code; U-Boot wants neither.""" + monkeypatch.setattr(keyboard.sys, "platform", "win32") + keys = [b"a"[0], 0xE0, 0x48, b"b"[0]] # a, Up, b + monkeypatch.setitem(sys.modules, "msvcrt", _FakeMsvcrt(keys)) + assert read_available_keys() == b"ab" + + def test_the_other_special_prefix_is_dropped_too(self, monkeypatch): + monkeypatch.setattr(keyboard.sys, "platform", "win32") + keys = [b"x"[0], 0x00, 0x3B] # x, F1 + monkeypatch.setitem(sys.modules, "msvcrt", _FakeMsvcrt(keys)) + assert read_available_keys() == b"x" + + +class TestRawTerminal: + def test_it_restores_what_it_changed(self, monkeypatch): + pytest.importorskip("termios") + import termios + + monkeypatch.setattr(keyboard.sys, "platform", "linux") + restored = [] + monkeypatch.setattr(termios, "tcgetattr", lambda fd: ["saved"]) + monkeypatch.setattr( + termios, "tcsetattr", lambda fd, when, attrs: restored.append(attrs), + ) + monkeypatch.setattr("tty.setcbreak", lambda fd: None) + with raw_terminal(_Fd(0)): + pass + assert restored == [["saved"]] + + def test_it_restores_even_when_the_body_raises(self, monkeypatch): + pytest.importorskip("termios") + import termios + + monkeypatch.setattr(keyboard.sys, "platform", "linux") + restored = [] + monkeypatch.setattr(termios, "tcgetattr", lambda fd: ["saved"]) + monkeypatch.setattr( + termios, "tcsetattr", lambda fd, when, attrs: restored.append(attrs), + ) + monkeypatch.setattr("tty.setcbreak", lambda fd: None) + with pytest.raises(RuntimeError): + with raw_terminal(_Fd(0)): + raise RuntimeError("serial went away") + assert restored == [["saved"]] + + def test_a_stdin_that_cannot_be_configured_is_not_fatal(self, monkeypatch): + """Piped or captured stdin still has to run, just without raw mode.""" + monkeypatch.setattr(keyboard.sys, "platform", "linux") + + class NoFileno: + def fileno(self): + raise OSError("captured") + + with raw_terminal(NoFileno()): + pass # must not raise + + def test_windows_needs_no_termios(self, monkeypatch): + monkeypatch.setattr(keyboard.sys, "platform", "win32") + with raw_terminal(None): + pass + + +class TestTheDocsMatchTheCode: + def test_terminal_mode_actually_reads_the_keyboard(self): + """The regression itself: `-t` promised a console and gave a viewer.""" + # encoding is not optional here: read_text() uses the locale codec, + # which on Windows is cp1252 and cannot decode this file. + app = (Path(__file__).parent.parent / "src/defib/cli/app.py").read_text( + encoding="utf-8", + ) + block = app.split("Normal U-Boot shell", 1)[1][:2000] + assert "read_available_keys" in block + assert "transport.write(typed)" in block + + def test_the_readme_does_not_promise_what_the_code_cannot_do(self): + readme = (Path(__file__).parent.parent / "README.md").read_text( + encoding="utf-8", + ) + assert "type commands directly" not in readme or "keystrokes" in readme + + +class _Fd: + def __init__(self, fd: int) -> None: + self._fd = fd + + def fileno(self) -> int: + return self._fd + + +class _FakeMsvcrt: + """Just enough msvcrt to drive `_read_windows` off Windows.""" + + def __init__(self, keys: list[int]) -> None: + self._keys = list(keys) + + def kbhit(self) -> bool: + return bool(self._keys) + + def getch(self) -> bytes: + return bytes([self._keys.pop(0)]) + + +@pytest.mark.skipif( + sys.platform == "win32", + reason="select() takes only sockets on Windows, and this path never runs there", +) +class TestOnePollCannotRunAway: + """Qodo review on OpenIPC/defib#136. + + The reader drained stdin for as long as it stayed readable. A person types + a few bytes and it returns at once, but a pipe that keeps producing never + stops being readable -- so the serial read, the stop flag and the transport + cleanup all waited for the producer rather than the board. + """ + + def test_a_flood_returns_after_a_bounded_read(self, monkeypatch): + # The cap is lowered rather than the write enlarged: filling a pipe + # past its buffer would block this test on the platform with the + # smallest one, and pipe capacity is not what is under test. + monkeypatch.setattr(keyboard, "_MAX_BYTES_PER_POLL", 64) + r, w = os.pipe() + os.write(w, b"x" * 1024) + monkeypatch.setattr(keyboard.sys, "platform", "linux") + monkeypatch.setattr(keyboard.sys, "stdin", _Fd(r)) + try: + got = _read_posix() + finally: + os.close(w) + os.close(r) + assert len(got) == 64 + + def test_the_default_cap_is_far_above_human_typing(self): + assert _MAX_BYTES_PER_POLL >= 1024 + + def test_a_short_burst_is_not_truncated(self, monkeypatch): + """The bound must be invisible to anyone actually typing.""" + r, w = os.pipe() + os.write(w, b"sf erase 0x0 0x1000000\n") + os.close(w) + monkeypatch.setattr(keyboard.sys, "platform", "linux") + monkeypatch.setattr(keyboard.sys, "stdin", _Fd(r)) + try: + assert _read_posix() == b"sf erase 0x0 0x1000000\n" + finally: + os.close(r) + + +class TestABrokenRestoreIsNeverSilent: + """Qodo review on OpenIPC/defib#136. + + cbreak leaves the shell with no echo and no line editing. If putting it + back fails and we say nothing, the operator is left with a terminal that + appears dead and no idea why. + """ + + def test_it_falls_back_to_tcsanow(self, monkeypatch): + termios = pytest.importorskip("termios") + attempts = [] + + def tcsetattr(fd, when, attrs): + attempts.append(when) + if when == termios.TCSADRAIN: + raise OSError("interrupted") + + monkeypatch.setattr(termios, "tcsetattr", tcsetattr) + _restore_terminal(0, ["saved"]) + assert attempts == [termios.TCSADRAIN, termios.TCSANOW] + + def test_a_total_failure_tells_the_operator_how_to_recover( + self, monkeypatch, capsys, + ): + termios = pytest.importorskip("termios") + + def always_fails(fd, when, attrs): + raise OSError("gone") + + monkeypatch.setattr(termios, "tcsetattr", always_fails) + _restore_terminal(0, ["saved"]) + err = capsys.readouterr().err + assert "stty sane" in err + + def test_it_does_not_raise_and_so_cannot_mask_the_real_error( + self, monkeypatch, capsys, + ): + termios = pytest.importorskip("termios") + monkeypatch.setattr(keyboard.sys, "platform", "linux") + monkeypatch.setattr(termios, "tcgetattr", lambda fd: ["saved"]) + monkeypatch.setattr( + termios, "tcsetattr", + lambda fd, when, attrs: (_ for _ in ()).throw(OSError("gone")), + ) + monkeypatch.setattr("tty.setcbreak", lambda fd: None) + + # The serial failure is what the caller needs to see, not our cleanup. + with pytest.raises(RuntimeError, match="serial went away"): + with raw_terminal(_Fd(0)): + raise RuntimeError("serial went away") + assert "stty sane" in capsys.readouterr().err