fix(cli): fence OSC 11 background query with DA1 so late replies can't leak into the prompt
The classic CLI's light-mode detection sends an OSC 11 background-color query and blind-waits 100ms. Terminal managers that swallow OSC 11 (herdr) made every startup pay the full 100ms for nothing, and any in-order relay that answers slower than 100ms (SSH bridges, WSL, loaded tmux servers) delivered the reply AFTER prompt_toolkit owned the tty — the rgb:.../escape payload leaked into the input line as gibberish characters. Fix: send the OSC 11 query followed by a DA1 sentinel (ESC [ c) in one write — the same fence pattern the Ink TUI's TerminalQuerier uses. Terminals answer queries in order and effectively all of them answer DA1, so the DA1 reply proves the terminal has already processed (or ignored) our OSC 11. Fast terminals and herdr-style multiplexers now resolve in ~1ms; slow relays get their reply consumed instead of leaked; a hypothetical DA1-mute terminal falls back at a 1s safety net, same clean timeout path as before. Adds real-PTY regression tests covering the herdr-style (DA1-only), slow-relay (+300ms reply), and fully mute emulator behaviors, each asserting zero leftover bytes in the tty buffer. Sabotage-verified: the slow-relay test fails against the old un-fenced code with the exact leak payload in LEFTOVER.
This commit is contained in:
Teknium
parent
54641186ff
commit
851f23ebc6
46
cli.py
46
cli.py
|
|
@ -2574,18 +2574,31 @@ def _luminance_from_hex(hex_str: str) -> float | None:
|
|||
return (0.2126 * r + 0.7152 * g + 0.0722 * b) / 255.0
|
||||
|
||||
|
||||
_DA1_REPLY_RE = re.compile(rb"\x1b\[\?[0-9;]*c")
|
||||
|
||||
|
||||
def _query_osc11_background() -> str | None:
|
||||
"""Ask the terminal for its background color via OSC 11.
|
||||
|
||||
Most modern terminals reply with \\x1b]11;rgb:RRRR/GGGG/BBBB\\x1b\\\\
|
||||
within a few ms. We wait up to 100ms total before giving up.
|
||||
Returns "#RRGGBB" or None on timeout / non-tty.
|
||||
Most modern terminals reply with \x1b]11;rgb:RRRR/GGGG/BBBB\x1b\\
|
||||
within a few ms. Returns "#RRGGBB" or None on timeout / non-tty.
|
||||
|
||||
Skipped over SSH: the round-trip routinely exceeds our 100ms budget, so a
|
||||
The OSC 11 query is fenced with a DA1 sentinel (\x1b[c) — the same
|
||||
pattern the Ink TUI's TerminalQuerier uses. Terminals answer queries
|
||||
in order and virtually every terminal answers DA1, so seeing the DA1
|
||||
reply proves the terminal already ignored our OSC 11 (multiplexers
|
||||
like herdr answer DA1 in <1ms while swallowing OSC 11). Without the
|
||||
fence we can only wait out a blind timeout, and a reply that arrives
|
||||
AFTER we stop listening leaks into prompt_toolkit's stdin as typed
|
||||
text — the "gibberish ANSI characters" seen inside terminal managers
|
||||
that relay color queries slowly (herdr, WSL bridges, some tmux
|
||||
setups).
|
||||
|
||||
Skipped over SSH: the round-trip routinely exceeds our budget, so a
|
||||
late reply lands after prompt_toolkit has grabbed the tty — its payload
|
||||
leaks in as typed text and the BEL terminator reads as Ctrl+G (open
|
||||
editor), trapping the user in a stray editor. Remote sessions fall back to
|
||||
COLORFGBG / env hints / the dark default instead.
|
||||
editor), trapping the user in a stray editor. Remote sessions fall back
|
||||
to COLORFGBG / env hints / the dark default instead.
|
||||
|
||||
After the main read + TCSAFLUSH, a short drain window (50 ms) catches
|
||||
late-arriving bytes that slipped past the flush — a race observed on VPS
|
||||
|
|
@ -2608,13 +2621,26 @@ def _query_osc11_background() -> str | None:
|
|||
except Exception:
|
||||
return None
|
||||
try:
|
||||
sys.stdout.write("\x1b]11;?\x1b\\")
|
||||
# OSC 11 query + DA1 sentinel fence, in one write so no
|
||||
# reordering is possible.
|
||||
sys.stdout.write("\x1b]11;?\x1b\\\x1b[c")
|
||||
sys.stdout.flush()
|
||||
except Exception:
|
||||
return None
|
||||
# Read up to ~50ms for the response
|
||||
# Read until the DA1 fence closes — proof the terminal has processed
|
||||
# everything up to and including our OSC 11, so nothing can arrive
|
||||
# late and leak into prompt_toolkit's stdin. DA1 is answered by
|
||||
# effectively every terminal ever made (it predates color), and on
|
||||
# real terminals the fence closes in single-digit milliseconds
|
||||
# (herdr: <1ms, xterm/kitty/tmux: <5ms). The 1s deadline is a
|
||||
# safety net for a hypothetical terminal that ignores DA1 — not a
|
||||
# window we ever expect to wait out. A slow in-order relay that
|
||||
# delivers the OSC 11 reply at e.g. 400ms is handled correctly:
|
||||
# we keep listening until its DA1 reply follows, so the payload is
|
||||
# consumed here instead of leaking as typed input (the "gibberish
|
||||
# ANSI characters" seen inside terminal managers).
|
||||
import select
|
||||
deadline = time.monotonic() + 0.1
|
||||
deadline = time.monotonic() + 1.0
|
||||
buf = b""
|
||||
while time.monotonic() < deadline:
|
||||
r, _, _ = select.select([fd], [], [], deadline - time.monotonic())
|
||||
|
|
@ -2627,7 +2653,7 @@ def _query_osc11_background() -> str | None:
|
|||
if not chunk:
|
||||
break
|
||||
buf += chunk
|
||||
if b"\x1b\\" in buf or b"\x07" in buf:
|
||||
if _DA1_REPLY_RE.search(buf):
|
||||
break
|
||||
# Parse: \x1b]11;rgb:RRRR/GGGG/BBBB\x1b\\
|
||||
m = re.search(rb"rgb:([0-9a-fA-F]+)/([0-9a-fA-F]+)/([0-9a-fA-F]+)", buf)
|
||||
|
|
|
|||
|
|
@ -180,14 +180,17 @@ class TestOsc11DrainGuard:
|
|||
"""Regression: a late-arriving OSC 11 reply must not leak into
|
||||
prompt_toolkit's input buffer (#40250).
|
||||
|
||||
The drain loop in the ``finally`` block of ``_query_osc11_background``
|
||||
reads (and discards) any bytes that arrive after TCSAFLUSH completes.
|
||||
Two layers guard against this: the DA1 fence keeps the main read loop
|
||||
listening until the terminal proves it has processed our query, and
|
||||
the drain loop in the ``finally`` block reads (and discards) any
|
||||
stragglers that slip past TCSAFLUSH.
|
||||
"""
|
||||
|
||||
def test_finally_drain_discards_late_bytes(self, cli_mod, monkeypatch):
|
||||
"""Simulate a terminal that sends the OSC 11 reply after the main
|
||||
read loop's deadline — the drain window must eat it."""
|
||||
import io, os, termios, tty as _tty
|
||||
def test_late_reply_is_consumed_not_leaked(self, cli_mod, monkeypatch):
|
||||
"""Simulate a terminal that sends the OSC 11 reply 150ms after the
|
||||
query. With the DA1 fence the main loop is still listening, so the
|
||||
reply is consumed AND used; nothing remains for prompt_toolkit."""
|
||||
import os, termios, tty as _tty
|
||||
|
||||
# Create a pipe pair to fake stdin
|
||||
read_fd, write_fd = os.pipe()
|
||||
|
|
@ -208,28 +211,204 @@ class TestOsc11DrainGuard:
|
|||
for v in ("SSH_CONNECTION", "SSH_CLIENT", "SSH_TTY"):
|
||||
monkeypatch.delenv(v, raising=False)
|
||||
|
||||
# Write a delayed OSC 11 reply — the main select() loop will time out
|
||||
# (nothing in the pipe during the 100ms window), then the drain loop
|
||||
# should read and discard it.
|
||||
# Write a delayed OSC 11 reply (then the DA1 fence reply) — the
|
||||
# fenced main loop must still be listening and consume both.
|
||||
import threading
|
||||
|
||||
def delayed_write():
|
||||
import time
|
||||
time.sleep(0.15) # after the 100ms main deadline
|
||||
os.write(write_fd, b"\x1b]11;rgb:0c0c/0c0c/0c0c\x1b\\")
|
||||
time.sleep(0.15)
|
||||
os.write(write_fd, b"\x1b]11;rgb:0c0c/0c0c/0c0c\x1b\\\x1b[?62;22c")
|
||||
|
||||
t = threading.Thread(target=delayed_write, daemon=True)
|
||||
t.start()
|
||||
|
||||
# The function should return None (no valid response during main window)
|
||||
# and the drain loop should eat the late bytes.
|
||||
# The late reply is consumed by the fenced read loop and used.
|
||||
result = cli_mod._query_osc11_background()
|
||||
assert result is None
|
||||
assert result == "#0C0C0C"
|
||||
|
||||
# Verify the pipe is drained — a non-blocking read should return empty
|
||||
import select
|
||||
r, _, _ = select.select([read_fd], [], [], 0)
|
||||
assert not r, "drain loop should have consumed late OSC 11 bytes"
|
||||
assert not r, "late OSC 11 bytes must be consumed, not left to leak"
|
||||
|
||||
os.close(read_fd)
|
||||
os.close(write_fd)
|
||||
|
||||
def test_post_deadline_straggler_is_drained(self, cli_mod, monkeypatch):
|
||||
"""Bytes that arrive after the main loop has already finished (DA1
|
||||
answered instantly, reply straggles in during teardown) are eaten
|
||||
by the post-flush drain window instead of leaking (#40250)."""
|
||||
import os, termios, tty as _tty
|
||||
|
||||
read_fd, write_fd = os.pipe()
|
||||
fake_attrs = [0, 0, 0, 0, 0, 0, [b'\x00'] * 32]
|
||||
monkeypatch.setattr(termios, "tcgetattr", lambda fd: fake_attrs)
|
||||
monkeypatch.setattr(termios, "tcsetattr", lambda fd, when, attrs: None)
|
||||
monkeypatch.setattr(_tty, "setcbreak", lambda fd: None)
|
||||
monkeypatch.setattr(cli_mod.sys.stdin, "isatty", lambda: True, raising=False)
|
||||
monkeypatch.setattr(cli_mod.sys.stdout, "isatty", lambda: True, raising=False)
|
||||
monkeypatch.setattr(cli_mod.sys.stdin, "fileno", lambda: read_fd, raising=False)
|
||||
for v in ("SSH_CONNECTION", "SSH_CLIENT", "SSH_TTY"):
|
||||
monkeypatch.delenv(v, raising=False)
|
||||
|
||||
# DA1 answered immediately (herdr-style: OSC 11 swallowed) — main
|
||||
# loop exits fast — then a straggler payload lands during teardown.
|
||||
os.write(write_fd, b"\x1b[?62;22c")
|
||||
|
||||
import threading
|
||||
|
||||
def straggler():
|
||||
import time
|
||||
time.sleep(0.02) # inside the 50ms drain window
|
||||
os.write(write_fd, b"\x1b]11;rgb:0c0c/0c0c/0c0c\x1b\\")
|
||||
|
||||
t = threading.Thread(target=straggler, daemon=True)
|
||||
t.start()
|
||||
|
||||
result = cli_mod._query_osc11_background()
|
||||
assert result is None # OSC 11 was swallowed; only DA1 answered
|
||||
|
||||
import select
|
||||
r, _, _ = select.select([read_fd], [], [], 0)
|
||||
assert not r, "drain loop should have consumed the straggler bytes"
|
||||
|
||||
os.close(read_fd)
|
||||
os.close(write_fd)
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────────────────
|
||||
# OSC 11 query — DA1 fence behavior.
|
||||
#
|
||||
# The query is fenced with a DA1 sentinel so a terminal manager that
|
||||
# swallows OSC 11 (herdr) or relays it slowly (SSH bridges, some tmux
|
||||
# setups) can never leave reply bytes in the tty buffer for
|
||||
# prompt_toolkit to read as typed input. These tests run the real
|
||||
# function in a child on a real PTY and play the terminal's role from
|
||||
# the parent side.
|
||||
|
||||
import os as _os
|
||||
import sys as _sys
|
||||
|
||||
|
||||
_CHILD_SRC = r"""
|
||||
import sys, os
|
||||
sys.path.insert(0, os.environ["HERMES_REPO"])
|
||||
import cli
|
||||
bg = cli._query_osc11_background()
|
||||
print("RESULT:" + repr(bg), flush=True)
|
||||
# Drain anything left in the tty buffer — must be empty (no leak).
|
||||
import termios, tty, select, time
|
||||
fd = sys.stdin.fileno()
|
||||
old = termios.tcgetattr(fd)
|
||||
tty.setcbreak(fd)
|
||||
buf = b""
|
||||
deadline = time.monotonic() + 0.6
|
||||
while time.monotonic() < deadline:
|
||||
r, _, _ = select.select([fd], [], [], 0.1)
|
||||
if r:
|
||||
buf += os.read(fd, 256)
|
||||
termios.tcsetattr(fd, termios.TCSADRAIN, old)
|
||||
print("LEFTOVER:" + repr(buf), flush=True)
|
||||
"""
|
||||
|
||||
|
||||
def _run_osc11_child(reply_fn, repo_root, timeout=8.0):
|
||||
"""Fork a PTY child running _query_osc11_background().
|
||||
|
||||
reply_fn(query_age_seconds) -> bytes to write once, or None to wait.
|
||||
Returns (result_line, leftover_line).
|
||||
"""
|
||||
import pty
|
||||
import time as _time
|
||||
|
||||
env = dict(_os.environ, HERMES_REPO=str(repo_root))
|
||||
for var in ("SSH_CONNECTION", "SSH_CLIENT", "SSH_TTY"):
|
||||
env.pop(var, None)
|
||||
pid, master = pty.fork()
|
||||
if pid == 0: # child
|
||||
_os.execvpe(_sys.executable, [_sys.executable, "-c", _CHILD_SRC], env)
|
||||
_os.set_blocking(master, False)
|
||||
out = b""
|
||||
answered = 0
|
||||
query_at = None
|
||||
t0 = _time.monotonic()
|
||||
try:
|
||||
while _time.monotonic() - t0 < timeout:
|
||||
try:
|
||||
chunk = _os.read(master, 1024)
|
||||
if chunk:
|
||||
out += chunk
|
||||
except (BlockingIOError, OSError):
|
||||
pass
|
||||
# importing cli primes _detect_light_mode() which issues its own
|
||||
# OSC 11 query before the explicit call — answer each query.
|
||||
if query_at is None and out.count(b"\x1b]11;?") > answered:
|
||||
query_at = _time.monotonic()
|
||||
if query_at is not None:
|
||||
payload = reply_fn(_time.monotonic() - query_at)
|
||||
if payload is not None:
|
||||
_os.write(master, payload)
|
||||
answered += 1
|
||||
query_at = None
|
||||
if b"LEFTOVER:" in out and out.rstrip().endswith(b"'"):
|
||||
break
|
||||
_time.sleep(0.005)
|
||||
finally:
|
||||
try:
|
||||
_os.close(master)
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
_os.waitpid(pid, 0)
|
||||
except ChildProcessError:
|
||||
pass
|
||||
text = out.decode("utf-8", "replace")
|
||||
print("child raw output:", repr(text)) # aids debugging on failure
|
||||
result = leftover = None
|
||||
for line in text.splitlines():
|
||||
# The PTY echoes the query bytes onto the same line as the first
|
||||
# print, so match anywhere in the line rather than at the start.
|
||||
if "RESULT:" in line and result is None:
|
||||
result = line.split("RESULT:", 1)[1]
|
||||
elif "LEFTOVER:" in line and leftover is None:
|
||||
leftover = line.split("LEFTOVER:", 1)[1]
|
||||
return result, leftover
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def repo_root():
|
||||
import pathlib
|
||||
return pathlib.Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
@pytest.mark.skipif(_sys.platform == "win32", reason="POSIX PTY test")
|
||||
class TestOsc11Da1Fence:
|
||||
def test_herdr_style_da1_only_returns_none_without_leak(self, repo_root):
|
||||
"""Terminal answers DA1 instantly but swallows OSC 11 (herdr)."""
|
||||
result, leftover = _run_osc11_child(
|
||||
lambda age: b"\x1b[?62;22c", repo_root
|
||||
)
|
||||
assert result == "None"
|
||||
assert leftover == "b''"
|
||||
|
||||
def test_slow_inorder_reply_is_consumed_not_leaked(self, repo_root):
|
||||
"""OSC 11 reply arrives at +300ms (past the old 100ms budget),
|
||||
DA1 right behind it. The fence keeps us listening, so the color
|
||||
is detected and nothing leaks into the tty buffer."""
|
||||
result, leftover = _run_osc11_child(
|
||||
lambda age: (
|
||||
b"\x1b]11;rgb:1e1e/1e1e/2e2e\x1b\\\x1b[?62;22c"
|
||||
if age > 0.3 else None
|
||||
),
|
||||
repo_root,
|
||||
)
|
||||
assert result == "'#1E1E2E'"
|
||||
assert leftover == "b''"
|
||||
|
||||
def test_mute_terminal_times_out_clean(self, repo_root):
|
||||
"""Terminal that answers nothing: give up at the safety-net
|
||||
deadline with no leftovers."""
|
||||
result, leftover = _run_osc11_child(lambda age: None, repo_root)
|
||||
assert result == "None"
|
||||
assert leftover == "b''"
|
||||
|
|
|
|||
Loading…
Reference in New Issue