From 07d110a06496ec66f58a2c4b0b9b961296da0741 Mon Sep 17 00:00:00 2001 From: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com> Date: Sat, 25 Jul 2026 00:30:05 +0200 Subject: [PATCH] =?UTF-8?q?fix(terminal):=20close=20the=20reaper's=20selec?= =?UTF-8?q?tion=E2=86=92reconnect=E2=86=92close=20TOCTOU?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_terminals_to_reap()` snapshots victims under `_LOCK` and then releases it, but `subscribe()` clears `unwatched_since` and appends to `_subscribers` under the terminal's own `_sub_lock`, which the selection pass never held. A viewer reconnecting in that gap left object identity — all `close_terminal(expected=…)` ever checked — still true, so the reaper killed a terminal that now had a live viewer. - `_claim_reap_victim()` re-establishes the *whole* selection predicate while holding `_LOCK` and the terminal's `_sub_lock`, and removes the registry entry in that same critical section: same object, no subscribers, and `unwatched_since` still set and still expired. A dead process is still reaped unconditionally — it cannot come back, and a viewer watching a corpse must not pin the entry forever. - `attach_terminal()` does the registry lookup and the subscribe atomically under `_LOCK`, replacing `get_terminal()` + `term.subscribe()` in the SSE route. It takes the same two locks in the same order as the claim, so exactly one side wins: the viewer attaches (and the reap is refused), or the entry is already gone (and the route answers "terminal not running" before committing a 200, instead of streaming from a corpse). Registration is the authority — a registered-but-`closed` terminal still attaches so the viewer receives `terminal_closed`. - Teardown moved to `_teardown_terminal()` and runs after both locks are released; `killpg`+`wait` can block ~2.5s and must not pin the registry. - Dropped the duplicated `self.last_activity = time.time()` the rebase left in `put_output()`. Tests: a barrier regression that pauses the reaper after selection, attaches a viewer through the production path, resumes, and asserts the terminal stays registered/live and still receives output; the inverse where the reap claims first and the attach reports not-running; refreshed-grace and replaced-entry claims; dead-with-viewer; and a probe that `_LOCK` is free during teardown. Three of them fail against an identity-only claim. --- api/routes.py | 32 +++-- api/terminal.py | 100 +++++++++++++++- tests/test_terminal_idle_reaper.py | 186 +++++++++++++++++++++++++++++ 3 files changed, 299 insertions(+), 19 deletions(-) diff --git a/api/routes.py b/api/routes.py index d97aa979..de708399 100644 --- a/api/routes.py +++ b/api/routes.py @@ -17842,18 +17842,7 @@ def _handle_terminal_output(handler, parsed): sid = qs.get("session_id", [""])[0] if not sid: return bad(handler, "session_id required") - from api.terminal import get_terminal - term = get_terminal(sid) - if term is None: - return j(handler, {"error": "terminal not running"}, status=404) - - handler.send_response(200) - handler.send_header("Content-Type", "text/event-stream; charset=utf-8") - handler.send_header("Cache-Control", "no-cache") - handler.send_header("X-Accel-Buffering", "no") - handler.send_header("Connection", "close") - end_sse_headers(handler) - _sse_set_write_deadline(handler) # Defect A: slow tab can't pin this thread + from api.terminal import attach_terminal # EventSource automatically returns the last received SSE id on transport # reconnect. Seed only newer backlog entries in that case so already-rendered # terminal bytes (including ANSI cursor controls) are not written twice. A @@ -17865,7 +17854,24 @@ def _handle_terminal_output(handler, parsed): after_seq = max(0, int(last_event_id)) except ValueError: pass - output = term.subscribe(after_seq=after_seq) + # Look up and subscribe in one atomic step. A separate `get_terminal()` then + # `term.subscribe()` leaves a window in which the idle reaper can claim and + # tear the terminal down, leaving this stream attached to a corpse after we + # already committed a 200. Attaching atomically means we either hold a live + # viewer (which makes the terminal un-reapable) or learn it is gone in time + # to answer 404. + attached = attach_terminal(sid, after_seq=after_seq) + if attached is None: + return j(handler, {"error": "terminal not running"}, status=404) + term, output = attached + + handler.send_response(200) + handler.send_header("Content-Type", "text/event-stream; charset=utf-8") + handler.send_header("Cache-Control", "no-cache") + handler.send_header("X-Accel-Buffering", "no") + handler.send_header("Connection", "close") + end_sse_headers(handler) + _sse_set_write_deadline(handler) # Defect A: slow tab can't pin this thread try: while True: try: diff --git a/api/terminal.py b/api/terminal.py index c668f2d6..538509a2 100644 --- a/api/terminal.py +++ b/api/terminal.py @@ -132,7 +132,6 @@ class TerminalSession: self.unwatched_since = time.time() def put_output(self, event: str, payload: dict) -> None: - self.last_activity = time.time() self.last_activity = time.time() with self._sub_lock: item = (self._next_output_seq, event, payload) @@ -455,13 +454,55 @@ def _terminals_to_reap(now: float) -> list[tuple[str, TerminalSession]]: return victims +def _claim_reap_victim(sid: str, term: TerminalSession, now: float) -> TerminalSession | None: + """Atomically remove *term* from the registry, or refuse. + + ``_terminals_to_reap`` snapshots victims and then releases ``_LOCK``, so by + the time we get here a viewer may have reconnected: ``subscribe()`` appends + to ``_subscribers`` and clears ``unwatched_since`` under the terminal's + ``_sub_lock``, which the selection pass never held. Object identity alone — + what ``close_terminal(expected=…)`` checks — is still true in that case, so + the reaper would kill a terminal that now has a live viewer. + + The claim therefore re-establishes the *whole* selection predicate while + holding both locks, and takes the entry out of ``_TERMINALS`` in the same + critical section. A concurrent ``attach_terminal()`` acquires the same two + locks in the same order, so exactly one of the two wins: either the viewer + is attached (and we refuse) or the entry is already gone (and the attach + reports "not running"). + + Returns the claimed terminal — the caller owns its teardown — or ``None``. + """ + with _LOCK: + if _TERMINALS.get(sid) is not term: + return None + # A dead process is reaped unconditionally: it cannot come back to life, + # and an attached viewer only means someone is watching a corpse. + if term.is_alive(): + with term._sub_lock: + if term._subscribers: + return None + unwatched = term.unwatched_since + if unwatched is None: + return None + if (now - unwatched) < _TERMINAL_IDLE_GRACE_SECONDS: + return None + del _TERMINALS[sid] + return term + + def _reap_idle_terminals(now: float) -> int: - """Close every terminal selected by ``_terminals_to_reap``. Returns the count - closed. ``expected=term`` guards against closing a restart replacement.""" + """Close every terminal selected by ``_terminals_to_reap`` that is *still* + idle when claimed. Returns the count closed.""" reaped = 0 for sid, term in _terminals_to_reap(now): - if close_terminal(sid, expected=term): - reaped += 1 + claimed = _claim_reap_victim(sid, term, now) + if claimed is None: + continue + # Process/fd teardown runs after both locks are released: killpg + wait + # can take seconds and must not block attaches or spawns. + _teardown_terminal(claimed) + reaped += 1 return reaped @@ -601,6 +642,42 @@ def get_terminal(session_id: str) -> TerminalSession | None: return term +def attach_terminal( + session_id: str, after_seq: int | None = None +) -> tuple[TerminalSession, queue.Queue] | None: + """Attach a viewer atomically against the idle reaper. + + ``get_terminal()`` followed by ``term.subscribe()`` leaves a window: the + reaper can claim and tear the terminal down in between, so the viewer ends + up subscribed to a corpse and the caller reports a live stream that will + never produce output. Doing the lookup and the subscribe inside the same + ``_LOCK`` section — the same lock, in the same order, that + ``_claim_reap_victim`` takes — makes the two mutually exclusive: + + * attach wins → ``_subscribers`` is non-empty and ``unwatched_since`` is + ``None`` before the reaper can revalidate, so the reap is refused; + * reap wins → the entry is already out of ``_TERMINALS``, so this returns + ``None`` and the route answers "terminal not running" instead of hanging. + + Registration is the authority, deliberately: both the reaper and + ``close_terminal()`` remove the entry *before* tearing the terminal down, so + "still in ``_TERMINALS``" is exactly the condition that cannot race. A + terminal that is registered but already flagged ``closed`` (its shell exited + and the reader loop has not retired it yet) still attaches, so the viewer + receives the ``terminal_closed`` event instead of a bare 404. + + Returns ``(term, queue)`` or ``None``. + """ + if not _TERMINAL_SUPPORTED: + return None + sid = str(session_id or "") + with _LOCK: + term = _TERMINALS.get(sid) + if term is None: + return None + return term, term.subscribe(after_seq=after_seq) + + def write_terminal(session_id: str, data: str) -> None: if not _TERMINAL_SUPPORTED: raise NotImplementedError("Embedded terminal is not supported on Windows") @@ -644,6 +721,18 @@ def close_terminal(session_id: str, *, expected: TerminalSession | None = None) term = _TERMINALS.pop(sid, None) if not term: return False + _teardown_terminal(term) + return True + + +def _teardown_terminal(term: TerminalSession) -> None: + """Kill the shell, close the pty master fd and reap descendants. + + Split out of ``close_terminal`` so a caller that has already claimed the + registry entry (the idle reaper) can run the teardown without re-entering + the registry lookup — and, importantly, without holding any lock while + ``killpg``/``wait`` block for up to ~2.5s. + """ term.closed.set() try: if term.proc.poll() is None: @@ -671,7 +760,6 @@ def close_terminal(session_id: str, *, expected: TerminalSession | None = None) except OSError: pass _reap_terminal_descendants(term.proc.pid) - return True def close_all_terminals() -> None: diff --git a/tests/test_terminal_idle_reaper.py b/tests/test_terminal_idle_reaper.py index 81568b55..da14ce0d 100644 --- a/tests/test_terminal_idle_reaper.py +++ b/tests/test_terminal_idle_reaper.py @@ -9,6 +9,8 @@ terminal as a belt-and-suspenders. A tab refresh / brief drop re-attaches within the grace and keeps the session. """ import os +import threading +import time import pytest @@ -154,3 +156,187 @@ def test_reaper_thread_starts_idempotently(monkeypatch): terminal._ensure_terminal_reaper() terminal._ensure_terminal_reaper() # second call is a no-op (already alive) assert started == [1] + + +# ── Selection → reconnect → close TOCTOU ──────────────────────────────────── +# `_terminals_to_reap()` snapshots victims under `_LOCK` and then RELEASES it. +# `subscribe()` clears `unwatched_since` under the terminal's own `_sub_lock`, +# which the selection pass never held, so a viewer can attach in that gap. The +# old close path re-checked object identity only — still true — and killed a +# terminal that now had a live viewer. The claim must re-establish the whole +# selection predicate under both locks. + + +def _expired_since(): + """A wall-clock stamp old enough to select a terminal for reaping.""" + return time.time() - terminal._TERMINAL_IDLE_GRACE_SECONDS - 5 + + +def test_reconnect_between_selection_and_close_keeps_the_terminal(monkeypatch): + """Attach a viewer after selection but before the claim: no reap. + + Drives the claim with the *stale snapshot* the selection pass produced. + Re-running the full `_reap_idle_terminals()` here would not pin anything: + its own second selection pass no longer sees the terminal as idle, so the + test would pass even with an identity-only claim — the exact defect. + """ + sid = "toctou-reconnect" + now = time.time() + term = _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since()) + + victims = terminal._terminals_to_reap(now) + assert [s for s, _ in victims] == [sid], "the terminal was not even selected" + + # The gap: a reconnecting EventSource attaches through the production path. + attached = terminal.attach_terminal(sid) + assert attached is not None + _term, output = attached + assert term.unwatched_since is None + + # The claim runs against the snapshot taken BEFORE the reconnect. + for victim_sid, victim_term in victims: + assert terminal._claim_reap_victim(victim_sid, victim_term, now) is None + + with terminal._LOCK: + assert terminal._TERMINALS.get(sid) is term, "a watched terminal was reaped" + assert not term.closed.is_set() + + # And it still receives output on that viewer's queue. + term.put_output("terminal_output", {"data": "hello"}) + assert output.get(timeout=1)[1] == "terminal_output" + + +def test_reconnect_racing_the_claim_under_a_barrier_keeps_the_terminal(monkeypatch): + """Same race, driven deterministically across two threads. + + The reaper is paused *after* it has selected its victim and is about to + claim it; the viewer attaches through the production path in that window; + then the reaper resumes. + """ + sid = "toctou-barrier" + term = _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since()) + + selected = threading.Event() + attached_done = threading.Event() + real_select = terminal._terminals_to_reap + + def paused_select(now): + victims = real_select(now) + selected.set() + # Hold the reaper here until the viewer is attached. + assert attached_done.wait(timeout=5) + return victims + + monkeypatch.setattr(terminal, "_terminals_to_reap", paused_select) + + result = {} + + def reap(): + result["reaped"] = terminal._reap_idle_terminals(time.time()) + + reaper = threading.Thread(target=reap) + reaper.start() + try: + assert selected.wait(timeout=5) + attached = terminal.attach_terminal(sid) + assert attached is not None, "attach lost to a reaper that had not claimed yet" + _term, output = attached + finally: + attached_done.set() + reaper.join(timeout=10) + assert not reaper.is_alive() + + assert result["reaped"] == 0 + with terminal._LOCK: + assert terminal._TERMINALS.get(sid) is term + assert not term.closed.is_set() + term.put_output("terminal_output", {"data": "still alive"}) + assert output.get(timeout=1)[1] == "terminal_output" + + +def test_reap_that_claims_first_makes_attach_report_not_running(monkeypatch): + """The inverse: once the reaper has claimed, an attach must fail cleanly. + + It must not hand back a queue for a terminal that is being torn down — + the caller would commit a 200 and stream from a corpse forever. + """ + sid = "toctou-reap-wins" + term = _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since()) + + assert terminal._reap_idle_terminals(time.time()) == 1 + assert term.closed.is_set() + with terminal._LOCK: + assert sid not in terminal._TERMINALS + + assert terminal.attach_terminal(sid) is None + + +def test_claim_refuses_a_terminal_whose_grace_was_refreshed(monkeypatch): + """`unwatched_since` moved forward after selection → not idle any more.""" + sid = "toctou-refreshed" + term = _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since()) + now = time.time() + assert [s for s, _ in terminal._terminals_to_reap(now)] == [sid] + + # A viewer attached and detached again inside the gap: no subscribers, but + # the grace restarted. + term.unwatched_since = time.time() + + assert terminal._claim_reap_victim(sid, term, now) is None + with terminal._LOCK: + assert terminal._TERMINALS.get(sid) is term + + +def test_claim_refuses_a_replaced_registry_entry(monkeypatch): + """A restart replaced the sid: the old snapshot must not kill the new one.""" + sid = "toctou-replaced" + old = _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since()) + now = time.time() + assert [s for s, _ in terminal._terminals_to_reap(now)] == [sid] + + replacement = _make_registered_term(monkeypatch, sid, unwatched_since=None) + assert terminal._claim_reap_victim(sid, old, now) is None + with terminal._LOCK: + assert terminal._TERMINALS.get(sid) is replacement + + +def test_dead_terminal_is_reaped_even_with_an_attached_viewer(monkeypatch): + """A viewer watching a corpse must not keep the entry alive forever.""" + sid = "toctou-dead" + term = _make_registered_term(monkeypatch, sid, alive=False, unwatched_since=None) + term.subscribe() + + assert terminal._reap_idle_terminals(time.time()) == 1 + with terminal._LOCK: + assert sid not in terminal._TERMINALS + + +def test_teardown_runs_without_holding_the_registry_lock(monkeypatch): + """Process teardown can block for seconds; it must not pin `_LOCK`. + + Otherwise a single hung shell stalls every attach and spawn for the whole + kill/wait budget. + """ + sid = "toctou-teardown-lock" + _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since()) + + observed = {} + + def slow_killpg(*_a, **_k): + # `_LOCK` is an RLock, so probing it from the reaping thread itself + # would succeed even while held. Probe from a separate thread, and + # acquire/release there — an RLock may only be released by its owner. + def probe(): + acquired = terminal._LOCK.acquire(timeout=0.5) + observed["free"] = acquired + if acquired: + terminal._LOCK.release() + + holder = threading.Thread(target=probe) + holder.start() + holder.join(timeout=5) + + monkeypatch.setattr(terminal.os, "killpg", slow_killpg) + terminal._reap_idle_terminals(time.time()) + + assert observed.get("free") is True, "_LOCK was held across process teardown"