From 5fa18d9c0c760e1b141e8f06b5d1b54596cc4539 Mon Sep 17 00:00:00 2001 From: jaylfc Date: Tue, 1 Sep 2026 22:39:17 +0000 Subject: [PATCH 1/3] feat(agent-terminal): tuiui apphost protocol client (D1) Adds tinyagentos.tuiui_conduit, a synchronous client for the tuiui apphost Unix socket. Speaks newline-delimited externally-tagged JSON per docs/design/taos-tuiui-spike-findings.md. Operations: connect (default $XDG_RUNTIME_DIR/tuiui-$USER/apphost.sock), Spawn (cmd/args/cwd/cols/rows -> AppId + pid), send_input (raw bytes serialised as integer array, NOT base64), list_apps (Roster), kill, set_meta, shutdown, iter_frames, frame_lines (ANSI-free CellBuffer grid -> plain text), and rebind_by_meta for meta-based AppId recovery across daemon restarts. Session-identity contract per the spike: AppId is transient within one apphost lifetime; the SetMeta blob (typically {title, rect, z, minimized, app_key}) is the persistent identifier. rebind_by_meta rosters and matches by title (+ optional app_key) so the caller can recover the new AppId after a daemon restart (counter resets to 1). Out of scope for D1: bind-mounting the socket out of agent LXCs, opencode wiring, UI. Gated on this client. Tests: tests/test_tuiui_conduit.py runs against an in-test StubApphost that speaks the same newline-JSON. Covers spawn round-trip, input byte encoding on the wire, frame-to-text reconstruction, meta-based rebind across a simulated apphost restart, and the default socket path. CI does not require the real tuiui binary. Proof: uv run pytest tests/test_tuiui_conduit.py -q -> 9 passed. --- changelog.d/tsk-6oaua3-tuiui-conduit.md | 3 + tests/test_tuiui_conduit.py | 321 +++++++++++++++++++ tinyagentos/tuiui_conduit.py | 402 ++++++++++++++++++++++++ 3 files changed, 726 insertions(+) create mode 100644 changelog.d/tsk-6oaua3-tuiui-conduit.md create mode 100644 tests/test_tuiui_conduit.py create mode 100644 tinyagentos/tuiui_conduit.py diff --git a/changelog.d/tsk-6oaua3-tuiui-conduit.md b/changelog.d/tsk-6oaua3-tuiui-conduit.md new file mode 100644 index 000000000..f850326f9 --- /dev/null +++ b/changelog.d/tsk-6oaua3-tuiui-conduit.md @@ -0,0 +1,3 @@ +### Added + +- Agent terminal conduit (D1): new `tinyagentos.tuiui_conduit` module is a synchronous client for the tuiui apphost Unix socket, speaking newline-delimited externally-tagged JSON per `docs/design/taos-tuiui-spike-findings.md`. Operations: `connect` (default path `$XDG_RUNTIME_DIR/tuiui-$USER/apphost.sock`), `Spawn` (cmd/args/cwd/cols/rows -> AppId + pid), `send_input` (raw bytes as integer array, not base64), `list_apps` (Roster), `kill`, `set_meta`, `shutdown`, `iter_frames`, `frame_lines` (ANSI-free grid -> text), and `rebind_by_meta` for meta-based AppId recovery across daemon restarts. New tests under `tests/test_tuiui_conduit.py` exercise the client against an in-test stub apphost (no real tuiui binary required in CI): 9 tests cover spawn round-trip, input byte encoding on the wire, frame-to-text reconstruction, and meta-based rebind after a simulated apphost restart with the AppId counter reset to 1. \ No newline at end of file diff --git a/tests/test_tuiui_conduit.py b/tests/test_tuiui_conduit.py new file mode 100644 index 000000000..d18a28a23 --- /dev/null +++ b/tests/test_tuiui_conduit.py @@ -0,0 +1,321 @@ +"""Tests for the tuiui apphost protocol client. + +Runs against an in-test stub apphost (a Unix-socket server that speaks the +same newline-delimited externally-tagged JSON). The real tuiui binary is +not required for CI. +""" + +from __future__ import annotations + +import json +import os +import socket +import threading +import time + +import pytest + +from tinyagentos.tuiui_conduit import ( + Frame, + TuiuiConduit, + TuiuiConduitError, + default_socket_path, +) + + +class StubApphost: + """In-test apphost: newline-delimited externally-tagged JSON over AF_UNIX. + + The state model mirrors the real apphost: an ``AppId`` counter starts at + 1 and increments per spawn; ``Roster`` includes the stored meta blob; + ``SetMeta`` updates the meta blob for an existing app. + """ + + def __init__(self, socket_path: str) -> None: + self.socket_path = socket_path + self._listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self._listener.bind(socket_path) + self._listener.listen(1) + self._listener.settimeout(2.0) + self._stop = threading.Event() + self._next_app = 1 + self._apps: dict[int, dict] = {} + self._client: socket.socket | None = None + self._thread = threading.Thread(target=self._serve, daemon=True) + self._thread.start() + + def _serve(self) -> None: + while not self._stop.is_set(): + try: + conn, _ = self._listener.accept() + except socket.timeout: + continue + except OSError: + return + self._client = conn + conn.settimeout(2.0) + buf = b"" + try: + while not self._stop.is_set(): + try: + chunk = conn.recv(65536) + except socket.timeout: + continue + if not chunk: + break + buf += chunk + while b"\n" in buf: + line, buf = buf.split(b"\n", 1) + if not line: + continue + msg = json.loads(line.decode("utf-8")) + self._handle(conn, msg) + except (ConnectionResetError, BrokenPipeError, json.JSONDecodeError, OSError): + return + finally: + try: + conn.close() + except OSError: + pass + + def _send(self, conn: socket.socket, payload: dict) -> None: + line = (json.dumps(payload) + "\n").encode("utf-8") + try: + conn.sendall(line) + except OSError: + pass + + def _frame( + self, + app: int, + rows: int = 3, + cols: int = 6, + text: str = "", + flags: int = 0, + ) -> dict: + chars = list(text.ljust(rows * cols, " ")) + row_cells = [ + [c for c in chars[i * cols:(i + 1) * cols]] + for i in range(rows) + ] + return { + "Frame": { + "grid": { + "cols": cols, + "rows_list": [ + {"cols": [{"ch": ch} for ch in row]} + for row in row_cells + ], + }, + "cursor": [0, 0], + "flags": flags, + "images": [], + "image_data": [], + "clear": False, + "switch_to": None, + "clipboard": None, + } + } + + def _handle(self, conn: socket.socket, msg: dict) -> None: + if "Spawn" in msg: + spawn = msg["Spawn"] + app_id = self._next_app + self._next_app += 1 + self._apps[app_id] = { + "cmd": spawn.get("cmd", ""), + "args": list(spawn.get("args", [])), + "pid": 10000 + app_id, + "cols": int(spawn.get("cols", 80)), + "rows": int(spawn.get("rows", 24)), + "age_secs": 0, + "alive": True, + "meta": None, + } + self._send(conn, {"Spawned": {"app": app_id, "pid": self._apps[app_id]["pid"]}}) + self._send(conn, self._frame(app_id, rows=2, cols=6, text="hi")) + return + if "Input" in msg: + return + if "ListApps" in msg: + roster = [ + { + "app": app_id, + "cmd": info["cmd"], + "args": info["args"], + "pid": info["pid"], + "cols": info["cols"], + "rows": info["rows"], + "age_secs": info["age_secs"], + "alive": info["alive"], + "meta": info["meta"], + } + for app_id, info in self._apps.items() + ] + self._send(conn, {"Roster": roster}) + return + if "SetMeta" in msg: + sm = msg["SetMeta"] + if sm["app"] in self._apps: + self._apps[sm["app"]]["meta"] = sm["meta"] + return + if "Kill" in msg: + if msg["Kill"]["app"] in self._apps: + self._apps[msg["Kill"]["app"]]["alive"] = False + return + if "Shutdown" in msg: + self._stop.set() + try: + conn.shutdown(socket.SHUT_RDWR) + except OSError: + pass + return + + def stop(self) -> None: + self._stop.set() + try: + self._listener.close() + except OSError: + pass + if self._client is not None: + try: + self._client.close() + except OSError: + pass + self._thread.join(timeout=2.0) + try: + os.unlink(self.socket_path) + except OSError: + pass + + +@pytest.fixture +def apphost_sock(tmp_path): + sock_path = str(tmp_path / "apphost.sock") + server = StubApphost(sock_path) + try: + yield server, sock_path + finally: + server.stop() + + +class TestDefaultSocketPath: + def test_default_uses_xdg_runtime_dir(self, monkeypatch): + monkeypatch.setenv("XDG_RUNTIME_DIR", "/run/user/1000") + monkeypatch.setenv("USER", "alice") + assert default_socket_path() == "/run/user/1000/tuiui-alice/apphost.sock" + + def test_default_falls_back_to_tmp_when_xdg_unset(self, monkeypatch): + monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) + monkeypatch.setenv("USER", "bob") + assert default_socket_path() == "/tmp/tuiui-bob/apphost.sock" + + +class TestSpawnRoundTrip: + def test_spawn_returns_app_and_pid(self, apphost_sock): + _, sock_path = apphost_sock + with TuiuiConduit(sock_path, timeout=2.0) as c: + result = c.spawn("sh", ["-c", "echo hi"], cols=80, rows=24) + assert result.app == 1 + assert result.pid == 10001 + + +class TestInputByteEncoding: + def test_send_input_writes_integer_array_on_wire(self, apphost_sock): + """Each byte must serialise as its integer code point, not base64. + + The apphost's ``_handle`` is wrapped so every line it parses off the + socket (the bytes the client actually wrote) is recorded, then we + verify the ``Input`` line carries an integer array. + """ + server, sock_path = apphost_sock + captured_lines: list[bytes] = [] + + original_handle = server._handle + + def _capturing_handle(conn, msg): + captured_lines.append(json.dumps(msg).encode("utf-8")) + original_handle(conn, msg) + + server._handle = _capturing_handle + + with TuiuiConduit(sock_path, timeout=2.0) as c: + c.spawn("sh", [], cols=6, rows=2) + c.send_input(1, b"hello") + time.sleep(0.1) + + server._handle = original_handle + + input_lines = [ln for ln in captured_lines if b'"Input"' in ln] + assert input_lines, f"no Input line captured: {captured_lines!r}" + decoded = json.loads(input_lines[0].decode("utf-8")) + assert decoded == {"Input": {"app": 1, "bytes": [104, 101, 108, 108, 111]}} + assert all(isinstance(b, int) for b in decoded["Input"]["bytes"]) + + +class TestFrameReconstruction: + def test_frame_lines_reconstructs_visible_text(self, apphost_sock): + """A Frame carrying 'hi' in a 2x6 grid must yield ['hi', ''].""" + _, sock_path = apphost_sock + with TuiuiConduit(sock_path, timeout=2.0) as c: + c.spawn("sh", ["-c", "echo hi"], cols=6, rows=2) + frame = next(c.iter_frames()) + lines = TuiuiConduit.frame_lines(frame) + assert lines == ["hi", ""] + + +class TestMetaRebindAcrossRestart: + def test_rebind_finds_app_after_counter_reset(self, tmp_path): + """Spawn under one apphost, simulate daemon restart, rebind by meta.""" + sock_a = str(tmp_path / "first.sock") + server_a = StubApphost(sock_a) + try: + with TuiuiConduit(sock_a, timeout=2.0) as c: + spawned = c.spawn("sh", ["-c", "echo hi"], cols=80, rows=24) + c.set_meta(spawned.app, [{"title": "agent-shell", "app_key": "k1"}]) + time.sleep(0.05) + finally: + server_a.stop() + + sock_b = str(tmp_path / "second.sock") + server_b = StubApphost(sock_b) + try: + with TuiuiConduit(sock_b, timeout=2.0) as c: + server_b._apps[1] = { + "cmd": "sh", + "args": ["-c", "echo hi"], + "pid": 99999, + "cols": 80, + "rows": 24, + "age_secs": 0, + "alive": True, + "meta": [{"title": "agent-shell", "app_key": "k1"}], + } + rebind = c.rebind_by_meta("agent-shell", app_key="k1") + assert rebind is not None + assert rebind.app == 1 + assert rebind.meta[0]["title"] == "agent-shell" + finally: + server_b.stop() + + def test_rebind_returns_none_when_no_match(self, apphost_sock): + _, sock_path = apphost_sock + with TuiuiConduit(sock_path, timeout=2.0) as c: + assert c.rebind_by_meta("does-not-exist") is None + + +class TestProtocolErrors: + def test_not_connected_raises(self, tmp_path): + c = TuiuiConduit(str(tmp_path / "missing.sock"), timeout=2.0) + with pytest.raises(TuiuiConduitError): + c.send_input(1, b"x") + + def test_list_apps_parses_roster(self, apphost_sock): + _, sock_path = apphost_sock + with TuiuiConduit(sock_path, timeout=2.0) as c: + c.spawn("sh", ["-c", "echo a"], cols=80, rows=24) + roster = c.list_apps() + assert len(roster) == 1 + assert roster[0].app == 1 + assert roster[0].cmd == "sh" + assert roster[0].alive is True \ No newline at end of file diff --git a/tinyagentos/tuiui_conduit.py b/tinyagentos/tuiui_conduit.py new file mode 100644 index 000000000..da9e7e3d1 --- /dev/null +++ b/tinyagentos/tuiui_conduit.py @@ -0,0 +1,402 @@ +"""tuiui apphost Unix-socket client. + +The tuiui apphost is the agent-terminal surface of taOS (not a sandboxed app). +This module is the protocol client that talks to it over the apphost Unix +socket using newline-delimited externally-tagged JSON, as verified by the +spike at docs/design/taos-tuiui-spike-findings.md. + +Scope (D1): protocol client only. No container plumbing, no bind-mounts, no UI. + +Session identity: AppId is transient within one apphost lifetime. The +SetMeta blob is the persistent identifier across a daemon restart. Use +:meth:`TuiuiConduit.rebind_by_meta` after a reconnect to recover the new +AppId for a known meta title. +""" + +from __future__ import annotations + +import json +import os +import socket +import threading +from dataclasses import dataclass, field +from typing import Any, Iterator + + +def default_socket_path() -> str: + """Return the default apphost socket path. + + Matches ``$XDG_RUNTIME_DIR/tuiui-$USER/apphost.sock`` per the spike + (per-user, mode 0600 socket, 0700 directory). Falls back to + ``/tmp/tuiui-$USER/apphost.sock`` when ``XDG_RUNTIME_DIR`` is unset. + """ + user = os.environ.get("USER", "unknown") + if xdg := os.environ.get("XDG_RUNTIME_DIR"): + return os.path.join(xdg, f"tuiui-{user}", "apphost.sock") + return os.path.join("/tmp", f"tuiui-{user}", "apphost.sock") + + +class TuiuiConduitError(Exception): + """Raised when the apphost returns an error or the protocol is violated.""" + + +@dataclass +class SpawnedApp: + """Result of a successful :meth:`TuiuiConduit.spawn` call.""" + + app: int + pid: int + + +@dataclass +class RosterEntry: + """One entry from a :meth:`TuiuiConduit.list_apps` (Roster) response.""" + + app: int + cmd: str + args: list[str] + pid: int + cols: int + rows: int + age_secs: int + alive: bool + meta: list | None = None + + +@dataclass +class Frame: + """A single ``Frame`` event from the apphost. + + ``cells`` is the row-major char grid extracted from the raw grid dict. + ``cols`` is the width of each row; ``rows`` is the row count. Per the + spike, cells are ANSI-free; reconstruct lines with + :meth:`TuiuiConduit.frame_lines`. + """ + + cells: list[str] + cols: int + rows: int + cursor: tuple[int, int] | None = None + flags: int = 0 + images: list = field(default_factory=list) + image_data: list = field(default_factory=list) + clear: bool = False + switch_to: int | None = None + clipboard: str | None = None + + +def _socket_connect(path: str, timeout: float) -> socket.socket: + """Open an AF_UNIX socket to ``path`` with the given timeout.""" + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.settimeout(timeout) + sock.connect(path) + return sock + + +class TuiuiConduit: + """Synchronous client for the tuiui apphost Unix socket. + + The wire format is one JSON object per line, externally tagged (the Rust + side uses serde's externally-tagged enums). One client at a time per + apphost instance. + + Use as a context manager or call :meth:`close` explicitly. + """ + + def __init__( + self, + socket_path: str | None = None, + *, + timeout: float = 30.0, + ) -> None: + self.socket_path = socket_path or default_socket_path() + self.timeout = timeout + self._sock: socket.socket | None = None + self._read_buf = b"" + self._req_counter = 0 + self._lock = threading.RLock() + + def __enter__(self) -> "TuiuiConduit": + self.connect() + return self + + def __exit__(self, exc_type, exc, tb) -> None: + self.close() + + def connect(self) -> None: + """Open the socket and prepare for JSON line exchange.""" + if self._sock is not None: + return + self._sock = _socket_connect(self.socket_path, self.timeout) + self._read_buf = b"" + + def close(self) -> None: + """Close the socket if open.""" + if self._sock is not None: + try: + self._sock.close() + finally: + self._sock = None + self._read_buf = b"" + + def _send(self, payload: dict[str, Any]) -> None: + """Encode ``payload`` as one JSON line and write it.""" + if self._sock is None: + raise TuiuiConduitError("not connected") + line = (json.dumps(payload, separators=(",", ":")) + "\n").encode("utf-8") + with self._lock: + self._sock.sendall(line) + + def _recv_event(self) -> dict[str, Any]: + """Read one full newline-delimited JSON object from the socket. + + Frames are pushed by the apphost without a request, so this is the + generic receive path. A request that expects a reply also goes through + here because the apphost interleaves frames freely. + """ + if self._sock is None: + raise TuiuiConduitError("not connected") + while True: + if b"\n" in self._read_buf: + line, self._read_buf = self._read_buf.split(b"\n", 1) + if not line: + continue + try: + return json.loads(line.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise TuiuiConduitError(f"bad frame: {exc!r}") from exc + chunk = self._sock.recv(65536) + if not chunk: + raise TuiuiConduitError("apphost closed connection") + self._read_buf += chunk + + def _next_req_id(self) -> int: + self._req_counter += 1 + return self._req_counter + + def spawn( + self, + cmd: str, + args: list[str] | None = None, + *, + cwd: str | None = None, + cols: int = 80, + rows: int = 24, + req_id: int | None = None, + timeout: float | None = None, + ) -> SpawnedApp: + """Spawn a PTY-backed app. + + ``cols``/``rows`` set the initial PTY size; the apphost pushes Frame + events with the live grid as the app produces output. + """ + payload: dict[str, Any] = { + "Spawn": { + "req_id": req_id if req_id is not None else self._next_req_id(), + "cmd": cmd, + "args": list(args) if args else [], + "cols": cols, + "rows": rows, + } + } + if cwd is not None: + payload["Spawn"]["cwd"] = cwd + with self._lock: + self._send(payload) + evt = self._wait_for_matching(self._match_spawned, timeout=timeout) + return SpawnedApp(app=int(evt["Spawned"]["app"]), pid=int(evt["Spawned"]["pid"])) + + @staticmethod + def _match_spawned(evt: dict[str, Any]) -> bool: + return "Spawned" in evt and "app" in evt["Spawned"] and "pid" in evt["Spawned"] + + def _wait_for_matching( + self, + predicate, + *, + timeout: float | None = None, + ) -> dict[str, Any]: + """Drain incoming events until one matches ``predicate``. + + The apphost may push Frame events between a request and its reply, + so callers cannot assume the next event is the reply. + """ + if timeout is None: + timeout = self.timeout + if self._sock is None: + raise TuiuiConduitError("not connected") + prev_timeout = self._sock.gettimeout() + if timeout is not None: + self._sock.settimeout(timeout) + try: + while True: + evt = self._recv_event() + if predicate(evt): + return evt + finally: + self._sock.settimeout(prev_timeout) + + def send_input(self, app: int, data: bytes) -> None: + """Write raw PTY bytes to ``app``. + + Per the spike, ``bytes`` serializes as an integer array (NOT base64): + ``{"Input": {"app": 1, "bytes": [104, 101, 108, 108, 111]}}`` writes + ``"hello"`` to the PTY. + """ + self._send({"Input": {"app": app, "bytes": list(data)}}) + + def list_apps(self) -> list[RosterEntry]: + """Ask the apphost for a Roster and return all live apps.""" + with self._lock: + self._send({"ListApps": {}}) + evt = self._wait_for_matching(self._match_roster) + apps = evt.get("Roster") or evt.get("apps") or [] + return [self._parse_roster(a) for a in apps] + + @staticmethod + def _match_roster(evt: dict[str, Any]) -> bool: + return "Roster" in evt or "apps" in evt + + @staticmethod + def _parse_roster(raw: dict[str, Any]) -> RosterEntry: + return RosterEntry( + app=int(raw["app"]), + cmd=str(raw.get("cmd", "")), + args=list(raw.get("args", [])), + pid=int(raw.get("pid", 0)), + cols=int(raw.get("cols", 0)), + rows=int(raw.get("rows", 0)), + age_secs=int(raw.get("age_secs", 0)), + alive=bool(raw.get("alive", True)), + meta=raw.get("meta"), + ) + + def kill(self, app: int) -> None: + """Kill a single app by AppId.""" + self._send({"Kill": {"app": app}}) + + def set_meta(self, app: int, meta: list) -> None: + """Store an opaque meta blob for ``app``. + + The meta blob (typically ``[{title, rect, z, minimized, app_key}]``) + is the persistent session identifier across daemon restarts; the + AppId alone resets on restart. + """ + self._send({"SetMeta": {"app": app, "meta": meta}}) + + def shutdown(self) -> None: + """Ask the apphost daemon to shut down.""" + self._send({"Shutdown": {}}) + + def iter_frames(self) -> Iterator[Frame]: + """Yield :class:`Frame` objects as the apphost pushes them. + + Mixed events (Roster replies, Spawned replies) are filtered out; + unknown event shapes are skipped so the iterator can run alongside + request/response traffic on the same socket. + """ + while True: + evt = self._recv_event() + if "Frame" not in evt: + continue + yield self._parse_frame(evt["Frame"]) + + @classmethod + def _parse_frame(cls, raw: dict[str, Any]) -> Frame: + grid = raw.get("grid") or {} + cells, cols, rows = _extract_grid(grid) + cursor = raw.get("cursor") + cur_pair: tuple[int, int] | None = None + if isinstance(cursor, (list, tuple)) and len(cursor) == 2: + cur_pair = (int(cursor[0]), int(cursor[1])) + return Frame( + cells=cells, + cols=cols, + rows=rows, + cursor=cur_pair, + flags=int(raw.get("flags", 0)), + images=list(raw.get("images", [])), + image_data=list(raw.get("image_data", [])), + clear=bool(raw.get("clear", False)), + switch_to=(int(raw["switch_to"]) if raw.get("switch_to") is not None else None), + clipboard=(str(raw["clipboard"]) if raw.get("clipboard") is not None else None), + ) + + @staticmethod + def frame_lines(frame: Frame) -> list[str]: + """Reconstruct visible text lines from a :class:`Frame`. + + Per the spike, every cell is ANSI-free, so the rows are read + row-major and each row's ``cols`` chars are joined. Trailing spaces + are stripped per line so empty rows show as ``""``. + """ + out: list[str] = [] + width = frame.cols + for r in range(frame.rows): + start = r * width + row_chars = frame.cells[start:start + width] + out.append("".join(row_chars).rstrip()) + return out + + def rebind_by_meta( + self, + title: str, + *, + app_key: str | None = None, + ) -> RosterEntry | None: + """Find a live app whose meta matches ``title`` (and ``app_key``). + + Use this after a reconnect, when the AppId counter has reset on + daemon restart: the meta blob is the persistent identifier, and the + new AppId for the same window lives at ``entry.app``. + """ + for entry in self.list_apps(): + meta = entry.meta or [] + for blob in meta: + if not isinstance(blob, dict): + continue + if blob.get("title") != title: + continue + if app_key is not None and blob.get("app_key") != app_key: + continue + return entry + return None + + +def _extract_grid(grid: dict[str, Any]) -> tuple[list[str], int, int]: + """Flatten a grid dict into a row-major list of ``ch`` strings. + + Returns ``(cells, cols, rows)``. Accepts both + ``{"rows_list": [{"cols": [...]}]}`` and ``{"cells": [...]}`` shapes. + When the grid only carries a flat ``cells`` list, ``cols`` is taken from + ``grid["cols"]`` (default 80) so the row-major join in + :meth:`Frame.frame_lines` stays deterministic. + """ + if not grid: + return [], 0, 0 + if isinstance(grid.get("cells"), list): + flat = grid["cells"] + cells = [str(c.get("ch", "") if isinstance(c, dict) else c) for c in flat] + cols = int(grid.get("cols", 80)) + rows = int(grid.get("rows", max(1, len(cells) // max(1, cols)))) + return cells, cols, rows + rows_raw = grid.get("rows_list") or grid.get("rows") + if isinstance(rows_raw, list): + cells: list[str] = [] + cols = 0 + for row in rows_raw: + row_cells: list[str] = [] + if isinstance(row, dict): + cols_raw = row.get("cols") or row.get("cells") or [] + for c in cols_raw: + row_cells.append(str(c.get("ch", "") if isinstance(c, dict) else c)) + if cols == 0: + cols = int(row.get("cols_len", len(row_cells))) + elif isinstance(row, list): + for c in row: + row_cells.append(str(c.get("ch", "") if isinstance(c, dict) else c)) + if cols == 0: + cols = len(row_cells) + cells.extend(row_cells) + return cells, cols, len(rows_raw) + return [], 0, 0 \ No newline at end of file From d8d574c23958dbb4f933012d1cd7654c01b4b2c9 Mon Sep 17 00:00:00 2001 From: jaylfc Date: Fri, 4 Sep 2026 12:54:55 +0000 Subject: [PATCH 2/3] fix(agent-terminal): demux the conduit socket and verify its owner Both review passes on #2691 landed on the same two defects in the D1 conduit, plus three smaller ones in the grid parser and connect(). Reads were unsynchronised across threads. iter_frames() called _recv_event() with no lock at all while another thread could be inside spawn()/list_apps() doing the same, so both were blocked in recv() on one socket and whichever the kernel woke first took the bytes. The loser then lost its event outright: iter_frames() `continue`d past any non-Frame event, discarding a Spawned/Roster reply a request was still waiting on, and _wait_for_matching() discarded any Frame that arrived before its reply. Either way the other consumer blocked until timeout. A single background reader thread now owns the socket and _read_buf and sorts every event into a frame backlog or a reply backlog. Nothing is discarded on the read path: a waiter takes only the reply it matches and leaves the rest, iter_frames() takes only frames. A consumer slower than frame_backlog loses the oldest frames, counted in dropped_frames rather than dropped silently. The socket path was also trusted blindly. With XDG_RUNTIME_DIR unset the default fell back to /tmp/tuiui-$USER/apphost.sock, which any local user can predict and pre-create, and nothing anywhere stat()ed the socket before connecting to it (CWE-377): the attacker's listener would receive every PTY keystroke this client sends and could forge replies back. connect() now refuses a socket, or a directory holding one, that is not owned by the calling euid or that grants group/other access, and refuses a symlink standing in for the socket. The fallback path is keyed on the numeric uid rather than $USER. connect() also wraps connection failures in TuiuiConduitError, which is what its documented contract promised. Grid parsing no longer guesses or drops cells: a flat cells list with no cols is one row instead of an assumed 80 wide (an assumed width silently re-flows every other width into nonsense), a partial final row is kept by ceiling division instead of dropped, and cells is normalised to exactly rows * cols with a new Frame.truncated flag so a short row is distinguishable from a blank one instead of being rstripped away. Tests: the race is reproduced with a real socketpair and two threads parked on it -- one in iter_frames(), one in spawn() -- with the apphost writing Frame, Spawned, Frame. On the pre-fix code spawn() fails with TimeoutError('timed out') because the iterator ate and discarded its reply. Also covered: a Frame preceding a reply, a Roster buffered before its waiter asks, ownership/mode/symlink refusal, connection-failure wrapping, and the grid geometry cases. The two tests review flagged as weak are tightened: the fixed sleep is replaced by an event the stub sets when it parses the Input line, and the restart test now burns AppId 1 so the pre- and post-restart ids provably differ. Docs-Reviewed: changelog.d/tsk-6oaua3-tuiui-conduit.md updated with the demux, ownership and grid-geometry behaviour; the module is still unreferenced outside its own tests, so no other doc describes it yet. --- changelog.d/tsk-6oaua3-tuiui-conduit.md | 5 +- tests/test_tuiui_conduit.py | 294 +++++++++++++++++- tinyagentos/tuiui_conduit.py | 389 +++++++++++++++++++----- 3 files changed, 599 insertions(+), 89 deletions(-) diff --git a/changelog.d/tsk-6oaua3-tuiui-conduit.md b/changelog.d/tsk-6oaua3-tuiui-conduit.md index f850326f9..ff625c893 100644 --- a/changelog.d/tsk-6oaua3-tuiui-conduit.md +++ b/changelog.d/tsk-6oaua3-tuiui-conduit.md @@ -1,3 +1,6 @@ ### Added -- Agent terminal conduit (D1): new `tinyagentos.tuiui_conduit` module is a synchronous client for the tuiui apphost Unix socket, speaking newline-delimited externally-tagged JSON per `docs/design/taos-tuiui-spike-findings.md`. Operations: `connect` (default path `$XDG_RUNTIME_DIR/tuiui-$USER/apphost.sock`), `Spawn` (cmd/args/cwd/cols/rows -> AppId + pid), `send_input` (raw bytes as integer array, not base64), `list_apps` (Roster), `kill`, `set_meta`, `shutdown`, `iter_frames`, `frame_lines` (ANSI-free grid -> text), and `rebind_by_meta` for meta-based AppId recovery across daemon restarts. New tests under `tests/test_tuiui_conduit.py` exercise the client against an in-test stub apphost (no real tuiui binary required in CI): 9 tests cover spawn round-trip, input byte encoding on the wire, frame-to-text reconstruction, and meta-based rebind after a simulated apphost restart with the AppId counter reset to 1. \ No newline at end of file +- Agent terminal conduit (D1): new `tinyagentos.tuiui_conduit` module is a synchronous client for the tuiui apphost Unix socket, speaking newline-delimited externally-tagged JSON per `docs/design/taos-tuiui-spike-findings.md`. Operations: `connect` (default path `$XDG_RUNTIME_DIR/tuiui-$USER/apphost.sock`), `Spawn` (cmd/args/cwd/cols/rows -> AppId + pid), `send_input` (raw bytes as integer array, not base64), `list_apps` (Roster), `kill`, `set_meta`, `shutdown`, `iter_frames`, `frame_lines` (ANSI-free grid -> text), and `rebind_by_meta` for meta-based AppId recovery across daemon restarts. New tests under `tests/test_tuiui_conduit.py` exercise the client against an in-test stub apphost (no real tuiui binary required in CI): 21 tests cover spawn round-trip, input byte encoding on the wire, frame-to-text reconstruction, meta-based rebind after a simulated apphost restart with the AppId counter reset, concurrent frame/reply demultiplexing, socket ownership refusal, and grid geometry handling. +- A single background reader thread now owns the conduit socket and demultiplexes incoming events into a frame backlog and a reply backlog. Previously `iter_frames()` and the request/response calls both called `recv()` unsynchronised on the same socket, so a frame consumer could swallow (and silently discard) a `Spawned`/`Roster` reply another thread was waiting on, and a request waiter could discard a `Frame` that arrived before its reply — either way the loser blocked until timeout. Frames dropped because a consumer fell behind `frame_backlog` (default 512) are counted in `TuiuiConduit.dropped_frames` rather than lost silently. +- `connect()` verifies the apphost socket before speaking to it: the socket and the directory holding it must be owned by the calling euid and must not grant group/other access, and the socket must not be a symlink. Without `XDG_RUNTIME_DIR` the default path is now keyed on the numeric uid instead of `$USER`, which any local user could predict and pre-create a listener for (CWE-377). `connect()` also wraps connection failures in `TuiuiConduitError` instead of leaking raw `OSError`. +- Grid parsing no longer loses or invents cells: a flat `cells` list with no `cols` is read as one row instead of being assumed 80 wide, a partial final row is kept via ceiling division rather than dropped, and `Frame.cells` is normalised to exactly `rows * cols` with the new `Frame.truncated` flag reporting a grid that did not match its declared geometry — so a short row is distinguishable from a blank one instead of being quietly stripped by `frame_lines()`. diff --git a/tests/test_tuiui_conduit.py b/tests/test_tuiui_conduit.py index d18a28a23..34981aca6 100644 --- a/tests/test_tuiui_conduit.py +++ b/tests/test_tuiui_conduit.py @@ -17,12 +17,54 @@ from tinyagentos.tuiui_conduit import ( Frame, + SpawnedApp, TuiuiConduit, TuiuiConduitError, + _extract_grid, default_socket_path, ) +def _line(payload: dict) -> bytes: + """Encode one wire event the way the apphost writes it.""" + return (json.dumps(payload) + "\n").encode("utf-8") + + +def _read_line(sock: socket.socket, timeout: float = 5.0) -> bytes: + """Read one newline-delimited line off a raw peer socket.""" + sock.settimeout(timeout) + buf = b"" + while b"\n" not in buf: + chunk = sock.recv(65536) + if not chunk: + raise AssertionError("peer closed before a full line arrived") + buf += chunk + return buf.split(b"\n", 1)[0] + + +# Time allowed for a thread to settle into its blocking read before the next +# event goes out, so both consumers really are parked on the socket at once. +# It shapes which reader the kernel wakes, not whether the assertions hold. +_SETTLE_SECS = 0.25 + + +def _flat_frame(text: str, *, cols: int, rows: int) -> dict: + """A Frame event carrying ``text`` in a flat ``cells`` grid.""" + chars = list(text.ljust(cols * rows, " ")) + return { + "Frame": { + "grid": {"cols": cols, "rows": rows, "cells": [{"ch": ch} for ch in chars]}, + "cursor": [0, 0], + "flags": 0, + "images": [], + "image_data": [], + "clear": False, + "switch_to": None, + "clipboard": None, + } + } + + class StubApphost: """In-test apphost: newline-delimited externally-tagged JSON over AF_UNIX. @@ -35,6 +77,10 @@ def __init__(self, socket_path: str) -> None: self.socket_path = socket_path self._listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self._listener.bind(socket_path) + # The real apphost publishes a mode-0600 socket inside a mode-0700 + # per-user directory; bind() alone leaves it at 0777 & ~umask, which + # the client's ownership guard rightly refuses. + os.chmod(socket_path, 0o600) self._listener.listen(1) self._listener.settimeout(2.0) self._stop = threading.Event() @@ -205,10 +251,18 @@ def test_default_uses_xdg_runtime_dir(self, monkeypatch): monkeypatch.setenv("USER", "alice") assert default_socket_path() == "/run/user/1000/tuiui-alice/apphost.sock" - def test_default_falls_back_to_tmp_when_xdg_unset(self, monkeypatch): + def test_fallback_is_uid_scoped_not_user_scoped(self, monkeypatch): + """$USER is caller-controlled env; the numeric uid is not. + + A $USER-keyed fallback path is trivially predictable by any local + user, who can then pre-create it and receive this client's PTY + input (CWE-377). + """ monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) monkeypatch.setenv("USER", "bob") - assert default_socket_path() == "/tmp/tuiui-bob/apphost.sock" + path = default_socket_path() + assert f"tuiui-{os.geteuid()}" in path + assert "tuiui-bob" not in path class TestSpawnRoundTrip: @@ -230,21 +284,25 @@ def test_send_input_writes_integer_array_on_wire(self, apphost_sock): """ server, sock_path = apphost_sock captured_lines: list[bytes] = [] + input_seen = threading.Event() original_handle = server._handle def _capturing_handle(conn, msg): captured_lines.append(json.dumps(msg).encode("utf-8")) + if "Input" in msg: + input_seen.set() original_handle(conn, msg) server._handle = _capturing_handle - with TuiuiConduit(sock_path, timeout=2.0) as c: - c.spawn("sh", [], cols=6, rows=2) - c.send_input(1, b"hello") - time.sleep(0.1) - - server._handle = original_handle + try: + with TuiuiConduit(sock_path, timeout=2.0) as c: + c.spawn("sh", [], cols=6, rows=2) + c.send_input(1, b"hello") + assert input_seen.wait(5.0), "apphost never parsed the Input line" + finally: + server._handle = original_handle input_lines = [ln for ln in captured_lines if b'"Input"' in ln] assert input_lines, f"no Input line captured: {captured_lines!r}" @@ -271,9 +329,17 @@ def test_rebind_finds_app_after_counter_reset(self, tmp_path): server_a = StubApphost(sock_a) try: with TuiuiConduit(sock_a, timeout=2.0) as c: + # Burn AppId 1 first so the pre-restart id cannot coincide + # with the id the restarted daemon hands out. + c.spawn("sh", ["-c", "true"], cols=80, rows=24) spawned = c.spawn("sh", ["-c", "echo hi"], cols=80, rows=24) + assert spawned.app == 2 c.set_meta(spawned.app, [{"title": "agent-shell", "app_key": "k1"}]) - time.sleep(0.05) + # A round trip on the same connection is an ordered barrier: + # the Roster reply cannot come back before SetMeta was applied. + assert any( + e.app == spawned.app and e.meta for e in c.list_apps() + ), "SetMeta was not applied before the restart" finally: server_a.stop() @@ -281,6 +347,8 @@ def test_rebind_finds_app_after_counter_reset(self, tmp_path): server_b = StubApphost(sock_b) try: with TuiuiConduit(sock_b, timeout=2.0) as c: + # The restarted daemon reset its counter: the same window is + # back under AppId 1, and only the meta blob still matches. server_b._apps[1] = { "cmd": "sh", "args": ["-c", "echo hi"], @@ -294,6 +362,7 @@ def test_rebind_finds_app_after_counter_reset(self, tmp_path): rebind = c.rebind_by_meta("agent-shell", app_key="k1") assert rebind is not None assert rebind.app == 1 + assert rebind.app != spawned.app, "the AppId did not actually change" assert rebind.meta[0]["title"] == "agent-shell" finally: server_b.stop() @@ -318,4 +387,209 @@ def test_list_apps_parses_roster(self, apphost_sock): assert len(roster) == 1 assert roster[0].app == 1 assert roster[0].cmd == "sh" - assert roster[0].alive is True \ No newline at end of file + assert roster[0].alive is True + +@pytest.fixture +def paired_conduit(tmp_path, monkeypatch): + """A conduit wired to one end of a real ``socketpair``. + + The other end is handed to the test so it can write exactly the event + sequence a race needs, byte for byte, with no stub scheduling in the + way. A real mode-0600 socket file is published at the conduit's path so + ``connect()``'s ownership guard sees production-shaped permissions. + """ + sock_path = str(tmp_path / "paired.sock") + listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + listener.bind(sock_path) + os.chmod(sock_path, 0o600) + client_end, peer_end = socket.socketpair() + + def _fake_connect(path: str, timeout: float) -> socket.socket: + client_end.settimeout(timeout) + return client_end + + monkeypatch.setattr("tinyagentos.tuiui_conduit._socket_connect", _fake_connect) + conduit = TuiuiConduit(sock_path, timeout=2.0) + try: + yield conduit, peer_end + finally: + conduit.close() + for sock in (peer_end, client_end, listener): + try: + sock.close() + except OSError: + pass + + +class TestConcurrentFramesAndRequests: + """The socket has two consumers; neither may eat the other's events.""" + + def test_spawn_reply_survives_a_concurrent_frame_consumer(self, paired_conduit): + """A frame consumer and a request waiter must not eat each other's events. + + Two threads are parked on the same socket -- one inside + ``iter_frames()``, one inside ``spawn()`` -- when the apphost writes + Frame, Spawned, Frame. Whichever blocked reader the kernel picks + first, the reply belongs to ``spawn()`` and both frames belong to + the iterator; no consumer may discard the other's event. + """ + conduit, peer = paired_conduit + conduit.connect() + + spawned: list[SpawnedApp] = [] + spawn_error: list[BaseException] = [] + consumed: list[str] = [] + consumer_error: list[BaseException] = [] + + def request() -> None: + try: + spawned.append(conduit.spawn("sh", [], cols=6, rows=2)) + except BaseException as exc: # noqa: BLE001 - reported to the test + spawn_error.append(exc) + + requester = threading.Thread(target=request, daemon=True) + requester.start() + # Reading the request off the peer is the deterministic proof that + # spawn() has written and is now sitting in the read path. + _read_line(peer) + time.sleep(_SETTLE_SECS) + + def consume() -> None: + try: + frames = conduit.iter_frames() + for _ in range(2): + consumed.append(TuiuiConduit.frame_lines(next(frames))[0]) + except BaseException as exc: # noqa: BLE001 - reported to the test + consumer_error.append(exc) + + consumer = threading.Thread(target=consume, daemon=True) + consumer.start() + time.sleep(_SETTLE_SECS) + + peer.sendall(_line(_flat_frame("one", cols=6, rows=2))) + time.sleep(_SETTLE_SECS) + peer.sendall(_line({"Spawned": {"app": 1, "pid": 4242}})) + time.sleep(_SETTLE_SECS) + peer.sendall(_line(_flat_frame("two", cols=6, rows=2))) + + requester.join(timeout=10.0) + consumer.join(timeout=10.0) + assert not spawn_error, f"spawn() lost its reply: {spawn_error!r}" + assert not consumer_error, f"frame consumer lost a frame: {consumer_error!r}" + assert [(s.app, s.pid) for s in spawned] == [(1, 4242)] + assert consumed == ["one", "two"] + + def test_frame_arriving_before_a_reply_is_not_discarded(self, paired_conduit): + """A Frame that precedes the reply must still reach ``iter_frames()``.""" + conduit, peer = paired_conduit + conduit.connect() + + def reply() -> None: + _read_line(peer) # the Spawn request + peer.sendall(_line(_flat_frame("early", cols=6, rows=2))) + peer.sendall(_line({"Spawned": {"app": 7, "pid": 99}})) + + threading.Thread(target=reply, daemon=True).start() + + spawned = conduit.spawn("sh", [], cols=6, rows=2) + assert spawned.app == 7 + frame = next(conduit.iter_frames()) + assert TuiuiConduit.frame_lines(frame)[0] == "early" + + def test_unmatched_reply_is_kept_for_its_own_waiter(self, paired_conduit): + """A Roster that arrives while Spawned is awaited must not be lost.""" + conduit, peer = paired_conduit + conduit.connect() + + def reply() -> None: + _read_line(peer) # the Spawn request + peer.sendall(_line({"Roster": [{"app": 3, "cmd": "sh", "pid": 1}]})) + peer.sendall(_line({"Spawned": {"app": 3, "pid": 1}})) + + threading.Thread(target=reply, daemon=True).start() + + assert conduit.spawn("sh", []).app == 3 + # The Roster was already buffered before anyone asked for it; it must + # still be there for list_apps() rather than dropped by spawn(). + roster = conduit.list_apps() + assert [e.app for e in roster] == [3] + + +class TestSocketOwnershipGuard: + """The client must not hand PTY input to someone else's listener.""" + + def test_connect_refuses_a_socket_owned_by_another_uid(self, apphost_sock, monkeypatch): + _, sock_path = apphost_sock + real_uid = os.stat(sock_path).st_uid + monkeypatch.setattr(os, "geteuid", lambda: real_uid + 1) + with pytest.raises(TuiuiConduitError, match="owned by uid"): + TuiuiConduit(sock_path, timeout=2.0).connect() + + def test_connect_refuses_a_group_or_world_accessible_socket(self, apphost_sock): + _, sock_path = apphost_sock + os.chmod(sock_path, 0o666) + with pytest.raises(TuiuiConduitError, match="group/other access"): + TuiuiConduit(sock_path, timeout=2.0).connect() + + def test_connect_refuses_a_world_writable_parent_directory(self, apphost_sock, tmp_path): + _, sock_path = apphost_sock + os.chmod(tmp_path, 0o777) + try: + with pytest.raises(TuiuiConduitError, match="group/other access"): + TuiuiConduit(sock_path, timeout=2.0).connect() + finally: + os.chmod(tmp_path, 0o700) + + def test_connect_wraps_connection_failure(self, tmp_path): + """A refused connection is part of the TuiuiConduitError contract.""" + dead_path = str(tmp_path / "dead.sock") + bound = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + bound.bind(dead_path) # bound but never listening -> ECONNREFUSED + os.chmod(dead_path, 0o600) + try: + with pytest.raises(TuiuiConduitError, match="cannot connect"): + TuiuiConduit(dead_path, timeout=2.0).connect() + finally: + bound.close() + + def test_connect_reports_a_missing_socket(self, tmp_path): + with pytest.raises(TuiuiConduitError, match="cannot stat"): + TuiuiConduit(str(tmp_path / "missing.sock"), timeout=2.0).connect() + + +class TestGridGeometry: + """The grid parser must never silently lose or invent cells.""" + + def test_partial_final_row_is_kept(self): + cells, cols, rows, truncated = _extract_grid({"cells": list("abcdefg"), "cols": 6}) + assert (cols, rows) == (6, 2) + assert truncated is True + frame = Frame(cells=cells, cols=cols, rows=rows, truncated=truncated) + assert TuiuiConduit.frame_lines(frame) == ["abcdef", "g"] + + def test_flat_grid_without_cols_does_not_assume_eighty(self): + cells, cols, rows, truncated = _extract_grid({"cells": list("x" * 132), "rows": 1}) + assert (cols, rows) == (132, 1) + assert truncated is False + assert len(cells) == 132 + + def test_flat_grid_without_any_geometry_is_one_row(self): + cells, cols, rows, truncated = _extract_grid({"cells": list("abc")}) + assert (cols, rows) == (3, 1) + assert truncated is False + assert cells == ["a", "b", "c"] + + def test_short_row_is_padded_and_flagged(self): + grid = { + "cols": 4, + "rows_list": [ + {"cols": [{"ch": ch} for ch in "ab"]}, + {"cols": [{"ch": ch} for ch in "cdef"]}, + ], + } + cells, cols, rows, truncated = _extract_grid(grid) + assert (cols, rows) == (4, 2) + assert truncated is True, "a short row must be reported, not rstripped away" + assert len(cells) == 8 + frame = Frame(cells=cells, cols=cols, rows=rows, truncated=truncated) + assert TuiuiConduit.frame_lines(frame) == ["ab", "cdef"] diff --git a/tinyagentos/tuiui_conduit.py b/tinyagentos/tuiui_conduit.py index da9e7e3d1..ed176e24e 100644 --- a/tinyagentos/tuiui_conduit.py +++ b/tinyagentos/tuiui_conduit.py @@ -11,6 +11,11 @@ SetMeta blob is the persistent identifier across a daemon restart. Use :meth:`TuiuiConduit.rebind_by_meta` after a reconnect to recover the new AppId for a known meta title. + +Threading: one background reader thread owns the socket and demultiplexes +every incoming event into a frame backlog and a reply backlog, so a thread +iterating frames and a thread awaiting a reply never compete for bytes and +never discard each other's events. """ from __future__ import annotations @@ -18,22 +23,41 @@ import json import os import socket +import stat +import tempfile import threading +import time +from collections import deque from dataclasses import dataclass, field from typing import Any, Iterator +# How long the reader thread blocks in one recv() before looping. Only a +# liveness knob: it bounds how quickly close() joins the reader. +_READ_POLL_SECS = 0.5 + +# Replies buffered for waiters that have not asked for them yet. Replies are +# small and one-per-request; this is a leak guard, not flow control. +_REPLY_BACKLOG = 256 + def default_socket_path() -> str: """Return the default apphost socket path. Matches ``$XDG_RUNTIME_DIR/tuiui-$USER/apphost.sock`` per the spike - (per-user, mode 0600 socket, 0700 directory). Falls back to - ``/tmp/tuiui-$USER/apphost.sock`` when ``XDG_RUNTIME_DIR`` is unset. + (per-user, mode 0600 socket, 0700 directory). + + With no ``XDG_RUNTIME_DIR`` there is no per-user runtime directory to + use, so the fallback is keyed on the numeric uid rather than ``$USER``: + ``$USER`` is caller-controlled environment that any local user can + predict and pre-create a listener for (CWE-377), while the uid is not + forgeable from the environment. Either way + :meth:`TuiuiConduit.connect` verifies the socket's owner and mode + before it speaks to whatever is listening there. """ - user = os.environ.get("USER", "unknown") if xdg := os.environ.get("XDG_RUNTIME_DIR"): + user = os.environ.get("USER", "unknown") return os.path.join(xdg, f"tuiui-{user}", "apphost.sock") - return os.path.join("/tmp", f"tuiui-{user}", "apphost.sock") + return os.path.join(tempfile.gettempdir(), f"tuiui-{os.geteuid()}", "apphost.sock") class TuiuiConduitError(Exception): @@ -67,10 +91,15 @@ class RosterEntry: class Frame: """A single ``Frame`` event from the apphost. - ``cells`` is the row-major char grid extracted from the raw grid dict. - ``cols`` is the width of each row; ``rows`` is the row count. Per the - spike, cells are ANSI-free; reconstruct lines with - :meth:`TuiuiConduit.frame_lines`. + ``cells`` is the row-major char grid extracted from the raw grid dict, + always exactly ``rows * cols`` long. ``cols`` is the width of each row; + ``rows`` is the row count. Per the spike, cells are ANSI-free; + reconstruct lines with :meth:`TuiuiConduit.frame_lines`. + + ``truncated`` is True when the wire grid did not match its own declared + geometry (short rows padded with spaces, or surplus cells dropped), so + a caller can tell missing data from genuinely blank cells instead of + having :meth:`TuiuiConduit.frame_lines` quietly strip the gap away. """ cells: list[str] @@ -83,6 +112,7 @@ class Frame: clear: bool = False switch_to: int | None = None clipboard: str | None = None + truncated: bool = False def _socket_connect(path: str, timeout: float) -> socket.socket: @@ -93,6 +123,43 @@ def _socket_connect(path: str, timeout: float) -> socket.socket: return sock +def _verify_socket_owner(path: str) -> None: + """Refuse an apphost socket another local user could have planted. + + The apphost publishes a mode-0600 socket inside a mode-0700 per-user + directory. Anything looser means a second local user can substitute a + listener of their own and then read every keystroke this client sends + to the PTY, or forge replies back to it. Both the socket and the + directory holding it are checked, because a socket nobody else can open + is still replaceable if its directory is writable by others. + """ + euid = os.geteuid() + directory = os.path.dirname(path) or "." + for target, label, is_expected_type in ( + (directory, "directory", stat.S_ISDIR), + (path, "socket", stat.S_ISSOCK), + ): + try: + # lstat, not stat: a symlink standing in for the socket is + # itself the substitution this check exists to catch. + st = os.lstat(target) if label == "socket" else os.stat(target) + except OSError as exc: + raise TuiuiConduitError( + f"cannot stat apphost {label} {target}: {exc}" + ) from exc + if not is_expected_type(st.st_mode): + raise TuiuiConduitError(f"apphost {label} {target} is not a {label}") + if st.st_uid != euid: + raise TuiuiConduitError( + f"refusing apphost {label} {target}: owned by uid {st.st_uid}, not {euid}" + ) + if st.st_mode & (stat.S_IRWXG | stat.S_IRWXO): + raise TuiuiConduitError( + f"refusing apphost {label} {target}: mode " + f"{stat.S_IMODE(st.st_mode):04o} grants group/other access" + ) + + class TuiuiConduit: """Synchronous client for the tuiui apphost Unix socket. @@ -100,6 +167,11 @@ class TuiuiConduit: side uses serde's externally-tagged enums). One client at a time per apphost instance. + A background reader thread is the only caller of ``recv()``; it sorts + every event into a frame backlog and a reply backlog. That is what lets + :meth:`iter_frames` run concurrently with :meth:`spawn` / :meth:`list_apps` + without either side consuming the other's events. + Use as a context manager or call :meth:`close` explicitly. """ @@ -108,13 +180,23 @@ def __init__( socket_path: str | None = None, *, timeout: float = 30.0, + frame_backlog: int = 512, ) -> None: self.socket_path = socket_path or default_socket_path() self.timeout = timeout + #: Frames dropped because no consumer kept up with ``frame_backlog``. + self.dropped_frames = 0 self._sock: socket.socket | None = None self._read_buf = b"" self._req_counter = 0 self._lock = threading.RLock() + self._cv = threading.Condition() + self._frames: deque[dict[str, Any]] = deque(maxlen=max(1, frame_backlog)) + self._replies: deque[dict[str, Any]] = deque(maxlen=_REPLY_BACKLOG) + self._reader: threading.Thread | None = None + self._reader_error: BaseException | None = None + self._closed = False + self._stopping = threading.Event() def __enter__(self) -> "TuiuiConduit": self.connect() @@ -124,38 +206,113 @@ def __exit__(self, exc_type, exc, tb) -> None: self.close() def connect(self) -> None: - """Open the socket and prepare for JSON line exchange.""" + """Open the socket and start the demultiplexing reader. + + Raises :class:`TuiuiConduitError` when the socket is not ours to + talk to, or when the connection itself fails: connection failures + are the most common error a caller sees, so they belong to the same + exception contract as protocol errors rather than leaking raw + ``OSError``. + """ if self._sock is not None: return - self._sock = _socket_connect(self.socket_path, self.timeout) + _verify_socket_owner(self.socket_path) + try: + sock = _socket_connect(self.socket_path, self.timeout) + except OSError as exc: + raise TuiuiConduitError( + f"cannot connect to apphost at {self.socket_path}: {exc}" + ) from exc self._read_buf = b"" + self._stopping.clear() + with self._cv: + self._frames.clear() + self._replies.clear() + self._reader_error = None + self._closed = False + self._sock = sock + sock.settimeout(_READ_POLL_SECS) + self._reader = threading.Thread( + target=self._read_loop, + args=(sock,), + name="tuiui-conduit-reader", + daemon=True, + ) + self._reader.start() def close(self) -> None: - """Close the socket if open.""" - if self._sock is not None: + """Close the socket and stop the reader thread.""" + self._stopping.set() + sock, self._sock = self._sock, None + if sock is not None: try: - self._sock.close() - finally: - self._sock = None - self._read_buf = b"" + sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass + try: + sock.close() + except OSError: + pass + reader, self._reader = self._reader, None + if reader is not None and reader is not threading.current_thread(): + reader.join(timeout=_READ_POLL_SECS * 4) + self._read_buf = b"" + with self._cv: + self._closed = True + self._cv.notify_all() def _send(self, payload: dict[str, Any]) -> None: """Encode ``payload`` as one JSON line and write it.""" - if self._sock is None: + sock = self._sock + if sock is None: raise TuiuiConduitError("not connected") line = (json.dumps(payload, separators=(",", ":")) + "\n").encode("utf-8") with self._lock: - self._sock.sendall(line) + try: + sock.sendall(line) + except OSError as exc: + raise TuiuiConduitError(f"apphost write failed: {exc}") from exc - def _recv_event(self) -> dict[str, Any]: + def _read_loop(self, sock: socket.socket) -> None: + """Own the socket and sort every event into its consumer's backlog. + + This is the only place ``recv()`` and ``_read_buf`` are touched, so + no two threads can ever split a JSON line between them or claim + bytes meant for the other. Frame events go to the frame backlog, + everything else to the reply backlog; nothing is discarded here. + """ + try: + while not self._stopping.is_set(): + try: + evt = self._recv_event(sock) + except TimeoutError: + continue + with self._cv: + if "Frame" in evt: + if len(self._frames) == self._frames.maxlen: + self.dropped_frames += 1 + self._frames.append(evt) + else: + self._replies.append(evt) + self._cv.notify_all() + except BaseException as exc: # noqa: BLE001 - handed to every waiter + with self._cv: + if not self._stopping.is_set(): + self._reader_error = exc + self._cv.notify_all() + finally: + with self._cv: + self._closed = True + self._cv.notify_all() + + def _recv_event(self, sock: socket.socket) -> dict[str, Any]: """Read one full newline-delimited JSON object from the socket. - Frames are pushed by the apphost without a request, so this is the - generic receive path. A request that expects a reply also goes through - here because the apphost interleaves frames freely. + Called only from the reader thread. Frames are pushed by the apphost + without a request, so this is the generic receive path: a request + that expects a reply also arrives here because the apphost + interleaves frames freely. """ - if self._sock is None: - raise TuiuiConduitError("not connected") while True: if b"\n" in self._read_buf: line, self._read_buf = self._read_buf.split(b"\n", 1) @@ -165,14 +322,24 @@ def _recv_event(self) -> dict[str, Any]: return json.loads(line.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise TuiuiConduitError(f"bad frame: {exc!r}") from exc - chunk = self._sock.recv(65536) + chunk = sock.recv(65536) if not chunk: raise TuiuiConduitError("apphost closed connection") self._read_buf += chunk + def _fail_if_reader_stopped(self) -> None: + """Raise the reader's failure (or a clean close) for a waiter. Caller holds ``_cv``.""" + if self._reader_error is not None: + raise TuiuiConduitError( + f"apphost read failed: {self._reader_error!r}" + ) from self._reader_error + if self._closed: + raise TuiuiConduitError("apphost closed connection") + def _next_req_id(self) -> int: - self._req_counter += 1 - return self._req_counter + with self._lock: + self._req_counter += 1 + return self._req_counter def spawn( self, @@ -216,25 +383,26 @@ def _wait_for_matching( *, timeout: float | None = None, ) -> dict[str, Any]: - """Drain incoming events until one matches ``predicate``. + """Return the first buffered reply matching ``predicate``. - The apphost may push Frame events between a request and its reply, - so callers cannot assume the next event is the reply. + Replies that do not match stay in the backlog for the waiter that + does want them, and Frame events never reach this queue at all, so + waiting for a reply can no longer destroy either. """ if timeout is None: timeout = self.timeout - if self._sock is None: - raise TuiuiConduitError("not connected") - prev_timeout = self._sock.gettimeout() - if timeout is not None: - self._sock.settimeout(timeout) - try: + deadline = None if timeout is None else time.monotonic() + timeout + with self._cv: while True: - evt = self._recv_event() - if predicate(evt): - return evt - finally: - self._sock.settimeout(prev_timeout) + for index, evt in enumerate(self._replies): + if predicate(evt): + del self._replies[index] + return evt + self._fail_if_reader_stopped() + remaining = None if deadline is None else deadline - time.monotonic() + if remaining is not None and remaining <= 0: + raise TuiuiConduitError("timed out waiting for an apphost reply") + self._cv.wait(remaining) def send_input(self, app: int, data: bytes) -> None: """Write raw PTY bytes to ``app``. @@ -245,11 +413,11 @@ def send_input(self, app: int, data: bytes) -> None: """ self._send({"Input": {"app": app, "bytes": list(data)}}) - def list_apps(self) -> list[RosterEntry]: + def list_apps(self, *, timeout: float | None = None) -> list[RosterEntry]: """Ask the apphost for a Roster and return all live apps.""" with self._lock: self._send({"ListApps": {}}) - evt = self._wait_for_matching(self._match_roster) + evt = self._wait_for_matching(self._match_roster, timeout=timeout) apps = evt.get("Roster") or evt.get("apps") or [] return [self._parse_roster(a) for a in apps] @@ -288,23 +456,41 @@ def shutdown(self) -> None: """Ask the apphost daemon to shut down.""" self._send({"Shutdown": {}}) - def iter_frames(self) -> Iterator[Frame]: + def iter_frames(self, *, timeout: float | None = None) -> Iterator[Frame]: """Yield :class:`Frame` objects as the apphost pushes them. - Mixed events (Roster replies, Spawned replies) are filtered out; - unknown event shapes are skipped so the iterator can run alongside - request/response traffic on the same socket. + Frames come from the backlog the reader thread fills, so this really + does run alongside request/response traffic: it can neither consume + a reply another thread is waiting on nor lose a frame that arrived + while a request was in flight. + + Waits up to ``timeout`` seconds (default: the conduit timeout) for + each frame and raises :class:`TuiuiConduitError` if none arrives or + the apphost hangs up. A consumer slower than ``frame_backlog`` + loses the oldest frames; :attr:`dropped_frames` counts them. """ while True: - evt = self._recv_event() - if "Frame" not in evt: - continue - yield self._parse_frame(evt["Frame"]) + yield self._parse_frame(self._next_frame(timeout)["Frame"]) + + def _next_frame(self, timeout: float | None) -> dict[str, Any]: + """Pop the oldest buffered Frame event, waiting up to ``timeout``.""" + if timeout is None: + timeout = self.timeout + deadline = None if timeout is None else time.monotonic() + timeout + with self._cv: + while True: + if self._frames: + return self._frames.popleft() + self._fail_if_reader_stopped() + remaining = None if deadline is None else deadline - time.monotonic() + if remaining is not None and remaining <= 0: + raise TuiuiConduitError("timed out waiting for an apphost frame") + self._cv.wait(remaining) @classmethod def _parse_frame(cls, raw: dict[str, Any]) -> Frame: grid = raw.get("grid") or {} - cells, cols, rows = _extract_grid(grid) + cells, cols, rows, truncated = _extract_grid(grid) cursor = raw.get("cursor") cur_pair: tuple[int, int] | None = None if isinstance(cursor, (list, tuple)) and len(cursor) == 2: @@ -320,6 +506,7 @@ def _parse_frame(cls, raw: dict[str, Any]) -> Frame: clear=bool(raw.get("clear", False)), switch_to=(int(raw["switch_to"]) if raw.get("switch_to") is not None else None), clipboard=(str(raw["clipboard"]) if raw.get("clipboard") is not None else None), + truncated=truncated, ) @staticmethod @@ -328,7 +515,10 @@ def frame_lines(frame: Frame) -> list[str]: Per the spike, every cell is ANSI-free, so the rows are read row-major and each row's ``cols`` chars are joined. Trailing spaces - are stripped per line so empty rows show as ``""``. + are stripped per line so empty rows show as ``""``. ``cells`` is + already normalised to ``rows * cols``, so a short row cannot be + mistaken here for a blank one -- check ``frame.truncated`` to tell + the two apart. """ out: list[str] = [] width = frame.cols @@ -363,40 +553,83 @@ def rebind_by_meta( return None -def _extract_grid(grid: dict[str, Any]) -> tuple[list[str], int, int]: +def _cell_char(cell: Any) -> str: + """Render one wire cell as its character.""" + return str(cell.get("ch", "") if isinstance(cell, dict) else cell) + + +def _ceil_div(numerator: int, denominator: int) -> int: + return -(-numerator // denominator) if denominator > 0 else 0 + + +def _fit(cells: list[str], cols: int, rows: int) -> tuple[list[str], int, int, bool]: + """Normalise ``cells`` to exactly ``rows * cols`` entries. + + Returns the ``truncated`` flag so a caller can tell a grid that did not + match its declared geometry from one that is simply blank. + """ + want = cols * rows + if len(cells) < want: + return cells + [" "] * (want - len(cells)), cols, rows, True + if len(cells) > want: + return cells[:want], cols, rows, True + return cells, cols, rows, False + + +def _extract_grid(grid: dict[str, Any]) -> tuple[list[str], int, int, bool]: """Flatten a grid dict into a row-major list of ``ch`` strings. - Returns ``(cells, cols, rows)``. Accepts both + Returns ``(cells, cols, rows, truncated)``. Accepts both ``{"rows_list": [{"cols": [...]}]}`` and ``{"cells": [...]}`` shapes. - When the grid only carries a flat ``cells`` list, ``cols`` is taken from - ``grid["cols"]`` (default 80) so the row-major join in - :meth:`Frame.frame_lines` stays deterministic. + ``cells`` is always exactly ``rows * cols`` long, and ``truncated`` + reports whether the wire grid had to be padded or clipped to get there. + + Geometry is never guessed: a flat ``cells`` list with no ``cols`` is + read as a single row rather than assumed to be 80 wide, because an + assumed width silently garbles every other width (a 132-column grid + would be re-flowed into nonsense with no signal at all). """ if not grid: - return [], 0, 0 + return [], 0, 0, False if isinstance(grid.get("cells"), list): - flat = grid["cells"] - cells = [str(c.get("ch", "") if isinstance(c, dict) else c) for c in flat] - cols = int(grid.get("cols", 80)) - rows = int(grid.get("rows", max(1, len(cells) // max(1, cols)))) - return cells, cols, rows + cells = [_cell_char(c) for c in grid["cells"]] + declared_cols = grid.get("cols") + declared_rows = grid.get("rows") + if declared_cols is not None: + cols = int(declared_cols) + elif declared_rows is not None and int(declared_rows) > 0: + cols = _ceil_div(len(cells), int(declared_rows)) + else: + cols = len(cells) + if cols <= 0: + return [], 0, 0, False + rows = int(declared_rows) if declared_rows is not None else _ceil_div(len(cells), cols) + return _fit(cells, cols, max(0, rows)) rows_raw = grid.get("rows_list") or grid.get("rows") if isinstance(rows_raw, list): - cells: list[str] = [] - cols = 0 + per_row: list[list[str]] = [] for row in rows_raw: - row_cells: list[str] = [] if isinstance(row, dict): - cols_raw = row.get("cols") or row.get("cells") or [] - for c in cols_raw: - row_cells.append(str(c.get("ch", "") if isinstance(c, dict) else c)) - if cols == 0: - cols = int(row.get("cols_len", len(row_cells))) + raw_cells = row.get("cols") or row.get("cells") or [] elif isinstance(row, list): - for c in row: - row_cells.append(str(c.get("ch", "") if isinstance(c, dict) else c)) - if cols == 0: - cols = len(row_cells) + raw_cells = row + else: + raw_cells = [] + per_row.append([_cell_char(c) for c in raw_cells]) + declared_cols = grid.get("cols") + if declared_cols is not None: + cols = int(declared_cols) + else: + cols = max((len(r) for r in per_row), default=0) + cells: list[str] = [] + truncated = False + for row_cells in per_row: + if len(row_cells) != cols: + truncated = True + if len(row_cells) < cols: + row_cells = row_cells + [" "] * (cols - len(row_cells)) + elif len(row_cells) > cols: + row_cells = row_cells[:cols] cells.extend(row_cells) - return cells, cols, len(rows_raw) - return [], 0, 0 \ No newline at end of file + return cells, cols, len(per_row), truncated + return [], 0, 0, False From c8a62fe978b1d8cce88c26dce9e63b681b96a401 Mon Sep 17 00:00:00 2001 From: jaylfc Date: Fri, 4 Sep 2026 13:22:47 +0000 Subject: [PATCH 3/3] fix(agent-terminal): bind the conduit's ownership check to the connection Second review pass on the demux commit. All five findings hold except one claim about blast radius, noted below. The ownership check validated a pathname and then handed the same pathname to connect(), so a local user with a writable non-sticky ancestor could swap the directory between the two lookups (CWE-367) and still get their listener talking to us. SO_PEERCRED closes it: the kernel reports the uid of the process actually holding the other end of this established connection, and there is nothing left to swap. The pathname checks stay -- they are what stops us opening the wrong thing in the first place -- and the peer check is what makes the connection itself trustworthy. Platforms without SO_PEERCRED keep the pathname checks alone. connect() and close() now run under a dedicated _conn_lock. Two threads racing through __enter__ could both pass the `if self._sock is not None` guard, both connect and both start a reader, leaking one socket and one thread. _send() reads self._sock under the same lock and holds it across sendall, so a concurrent close() can no longer pull the fd out mid-write and make a shutdown look like an apphost write failure. Every int() over wire data goes through _as_int(), which reports a protocol violation instead of a bare ValueError -- grid geometry, cursor, flags, switch_to, Spawned and Roster fields alike. An explicit JSON null still means "not sent", as it already did for switch_to. The review's claim that this took down the whole client is not accurate: _parse_frame() runs in the consumer thread, not the reader thread, so a malformed frame only ever reached the caller who asked for that frame. A test now pins that: after a bad frame raises for the frame consumer, the reply queued behind it is still there for its waiter. The reply backlog dropped its oldest entry silently once full, which is exactly the reply a waiter may be blocked on; dropped_replies now counts it the way dropped_frames already did, and both counters are covered. The reader also stopped borrowing the socket timeout to bound how long it parks: it polls with select() instead, so shortening the read poll no longer shortens how long a write is allowed to take. Docs-Reviewed: changelog.d/tsk-6oaua3-tuiui-conduit.md extended with the peer-credential check and the drop counters; the module still has no callers outside its own tests, so no other doc describes it. --- changelog.d/tsk-6oaua3-tuiui-conduit.md | 3 +- tests/test_tuiui_conduit.py | 141 ++++++++++++++ tinyagentos/tuiui_conduit.py | 232 +++++++++++++++++------- 3 files changed, 306 insertions(+), 70 deletions(-) diff --git a/changelog.d/tsk-6oaua3-tuiui-conduit.md b/changelog.d/tsk-6oaua3-tuiui-conduit.md index ff625c893..53a0e8396 100644 --- a/changelog.d/tsk-6oaua3-tuiui-conduit.md +++ b/changelog.d/tsk-6oaua3-tuiui-conduit.md @@ -2,5 +2,6 @@ - Agent terminal conduit (D1): new `tinyagentos.tuiui_conduit` module is a synchronous client for the tuiui apphost Unix socket, speaking newline-delimited externally-tagged JSON per `docs/design/taos-tuiui-spike-findings.md`. Operations: `connect` (default path `$XDG_RUNTIME_DIR/tuiui-$USER/apphost.sock`), `Spawn` (cmd/args/cwd/cols/rows -> AppId + pid), `send_input` (raw bytes as integer array, not base64), `list_apps` (Roster), `kill`, `set_meta`, `shutdown`, `iter_frames`, `frame_lines` (ANSI-free grid -> text), and `rebind_by_meta` for meta-based AppId recovery across daemon restarts. New tests under `tests/test_tuiui_conduit.py` exercise the client against an in-test stub apphost (no real tuiui binary required in CI): 21 tests cover spawn round-trip, input byte encoding on the wire, frame-to-text reconstruction, meta-based rebind after a simulated apphost restart with the AppId counter reset, concurrent frame/reply demultiplexing, socket ownership refusal, and grid geometry handling. - A single background reader thread now owns the conduit socket and demultiplexes incoming events into a frame backlog and a reply backlog. Previously `iter_frames()` and the request/response calls both called `recv()` unsynchronised on the same socket, so a frame consumer could swallow (and silently discard) a `Spawned`/`Roster` reply another thread was waiting on, and a request waiter could discard a `Frame` that arrived before its reply — either way the loser blocked until timeout. Frames dropped because a consumer fell behind `frame_backlog` (default 512) are counted in `TuiuiConduit.dropped_frames` rather than lost silently. -- `connect()` verifies the apphost socket before speaking to it: the socket and the directory holding it must be owned by the calling euid and must not grant group/other access, and the socket must not be a symlink. Without `XDG_RUNTIME_DIR` the default path is now keyed on the numeric uid instead of `$USER`, which any local user could predict and pre-create a listener for (CWE-377). `connect()` also wraps connection failures in `TuiuiConduitError` instead of leaking raw `OSError`. +- `connect()` verifies the apphost socket before speaking to it: the socket and the directory holding it must be owned by the calling euid and must not grant group/other access, and the socket must not be a symlink. Because a pathname can be swapped between the check and the connect (CWE-367), the connected peer's uid is then read from `SO_PEERCRED` and must also match; `connect()`/`close()` run under a dedicated lock so racing callers cannot leak a second socket and reader thread. Without `XDG_RUNTIME_DIR` the default path is now keyed on the numeric uid instead of `$USER`, which any local user could predict and pre-create a listener for (CWE-377). `connect()` also wraps connection failures in `TuiuiConduitError` instead of leaking raw `OSError`. - Grid parsing no longer loses or invents cells: a flat `cells` list with no `cols` is read as one row instead of being assumed 80 wide, a partial final row is kept via ceiling division rather than dropped, and `Frame.cells` is normalised to exactly `rows * cols` with the new `Frame.truncated` flag reporting a grid that did not match its declared geometry — so a short row is distinguishable from a blank one instead of being quietly stripped by `frame_lines()`. +- Values arriving on the wire are coerced through a checked converter, so a malformed grid geometry, cursor, `flags`, `switch_to`, `Spawned` or `Roster` field raises `TuiuiConduitError` rather than a bare `ValueError`. Replies dropped because the reply backlog filled are counted in `TuiuiConduit.dropped_replies`, mirroring `dropped_frames`. diff --git a/tests/test_tuiui_conduit.py b/tests/test_tuiui_conduit.py index 34981aca6..9c996292d 100644 --- a/tests/test_tuiui_conduit.py +++ b/tests/test_tuiui_conduit.py @@ -12,15 +12,18 @@ import socket import threading import time +from collections import deque import pytest +from tinyagentos import tuiui_conduit from tinyagentos.tuiui_conduit import ( Frame, SpawnedApp, TuiuiConduit, TuiuiConduitError, _extract_grid, + _verify_connected_peer, default_socket_path, ) @@ -593,3 +596,141 @@ def test_short_row_is_padded_and_flagged(self): assert len(cells) == 8 frame = Frame(cells=cells, cols=cols, rows=rows, truncated=truncated) assert TuiuiConduit.frame_lines(frame) == ["ab", "cdef"] + + +class TestPeerCredentialGuard: + """A verified pathname can be swapped; the connected peer cannot.""" + + def test_peer_running_as_another_uid_is_refused(self, monkeypatch): + ours, _theirs = socket.socketpair() + try: + monkeypatch.setattr(os, "geteuid", lambda: os.getuid() + 1) + with pytest.raises(TuiuiConduitError, match="peer runs as uid"): + _verify_connected_peer(ours, "/run/user/x/apphost.sock") + finally: + ours.close() + _theirs.close() + + def test_peer_running_as_us_is_accepted(self): + ours, theirs = socket.socketpair() + try: + _verify_connected_peer(ours, "/run/user/x/apphost.sock") + finally: + ours.close() + theirs.close() + + def test_connect_verifies_the_peer_it_actually_reached(self, apphost_sock): + """The guard runs on the live connection, not just on the path.""" + _, sock_path = apphost_sock + calls: list[str] = [] + monkeypatched = pytest.MonkeyPatch() + monkeypatched.setattr( + "tinyagentos.tuiui_conduit._verify_connected_peer", + lambda sock, path: calls.append(path), + ) + try: + with TuiuiConduit(sock_path, timeout=2.0): + pass + finally: + monkeypatched.undo() + assert calls == [sock_path] + + +class TestMalformedWireData: + """Garbage on the wire is a protocol error, not an arbitrary ValueError.""" + + @pytest.mark.parametrize( + "grid", + [ + {"cells": list("ab"), "rows": "two"}, + {"cells": list("ab"), "cols": "six"}, + {"cols": "wide", "rows_list": [{"cols": [{"ch": "a"}]}]}, + ], + ) + def test_non_integer_geometry_raises_conduit_error(self, grid): + with pytest.raises(TuiuiConduitError, match="not an integer"): + _extract_grid(grid) + + def test_null_geometry_is_treated_as_absent(self): + """An explicit JSON null means "not sent", as it does elsewhere.""" + cells, cols, rows, truncated = _extract_grid({"cells": list("ab"), "cols": None}) + assert (cols, rows) == (2, 1) + assert truncated is False + + def test_non_integer_roster_field_raises_conduit_error(self): + with pytest.raises(TuiuiConduitError, match="not an integer"): + TuiuiConduit._parse_roster({"app": "one"}) + + def test_a_malformed_frame_does_not_kill_the_reader(self, paired_conduit): + """One bad frame must not take down every other in-flight consumer.""" + conduit, peer = paired_conduit + conduit.connect() + bad = _flat_frame("x", cols=1, rows=1) + bad["Frame"]["grid"]["cols"] = "wide" + peer.sendall(_line(bad)) + peer.sendall(_line({"Spawned": {"app": 5, "pid": 55}})) + + frames = conduit.iter_frames() + with pytest.raises(TuiuiConduitError, match="not an integer"): + next(frames) + # The reader thread is untouched: the reply behind the bad frame is + # still there for its waiter. + assert conduit._wait_for_matching(TuiuiConduit._match_spawned)["Spawned"]["app"] == 5 + + +class TestBacklogAccounting: + """A dropped event must be counted, never silently vanish.""" + + def test_dropped_replies_are_counted(self, paired_conduit): + conduit, peer = paired_conduit + conduit._replies = deque(maxlen=2) + conduit.connect() + conduit._replies = deque(maxlen=2) + for app in range(4): + peer.sendall(_line({"Roster": [{"app": app, "cmd": "sh", "pid": 1}]})) + deadline = time.monotonic() + 5.0 + while conduit.dropped_replies < 2 and time.monotonic() < deadline: + time.sleep(0.05) + assert conduit.dropped_replies == 2 + + def test_dropped_frames_are_counted(self, paired_conduit): + conduit, peer = paired_conduit + conduit.connect() + conduit._frames = deque(maxlen=2) + for text in ("a", "b", "c", "d"): + peer.sendall(_line(_flat_frame(text, cols=1, rows=1))) + deadline = time.monotonic() + 5.0 + while conduit.dropped_frames < 2 and time.monotonic() < deadline: + time.sleep(0.05) + assert conduit.dropped_frames == 2 + + +class TestConnectIsAtomic: + """Racing __enter__ calls must not leak a socket or a reader thread.""" + + def test_concurrent_connect_opens_exactly_one_socket(self, apphost_sock): + _, sock_path = apphost_sock + opened: list[socket.socket] = [] + real_connect = tuiui_conduit._socket_connect + + def counting_connect(path: str, timeout: float) -> socket.socket: + sock = real_connect(path, timeout) + opened.append(sock) + # Widen the window between the guard and the assignment. + time.sleep(0.1) + return sock + + monkeypatched = pytest.MonkeyPatch() + monkeypatched.setattr(tuiui_conduit, "_socket_connect", counting_connect) + conduit = TuiuiConduit(sock_path, timeout=2.0) + try: + threads = [threading.Thread(target=conduit.connect) for _ in range(4)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=10.0) + assert len(opened) == 1, f"connect() opened {len(opened)} sockets" + assert threading.active_count() >= 1 + finally: + monkeypatched.undo() + conduit.close() diff --git a/tinyagentos/tuiui_conduit.py b/tinyagentos/tuiui_conduit.py index ed176e24e..aa8c5ef59 100644 --- a/tinyagentos/tuiui_conduit.py +++ b/tinyagentos/tuiui_conduit.py @@ -22,8 +22,10 @@ import json import os +import select import socket import stat +import struct import tempfile import threading import time @@ -39,6 +41,9 @@ # small and one-per-request; this is a leak guard, not flow control. _REPLY_BACKLOG = 256 +# struct ucred: pid, uid, gid as native ints. +_UCRED = "3i" + def default_socket_path() -> str: """Return the default apphost socket path. @@ -64,6 +69,22 @@ class TuiuiConduitError(Exception): """Raised when the apphost returns an error or the protocol is violated.""" +def _as_int(value: Any, field: str) -> int: + """Coerce one wire value to int, or report it as a protocol violation. + + Everything arriving on the socket is untrusted input. A bare + ``int("two")`` would surface as a ValueError from whichever call the + caller happened to make, which is neither the documented contract nor + something a caller can reasonably catch. + """ + try: + return int(value) + except (TypeError, ValueError) as exc: + raise TuiuiConduitError( + f"bad frame: {field}={value!r} is not an integer" + ) from exc + + @dataclass class SpawnedApp: """Result of a successful :meth:`TuiuiConduit.spawn` call.""" @@ -160,6 +181,33 @@ def _verify_socket_owner(path: str) -> None: ) +def _verify_connected_peer(sock: socket.socket, path: str) -> None: + """Confirm the process holding the other end really is this user. + + :func:`_verify_socket_owner` validates a *pathname*, and a pathname can + be swapped between the check and the connect (CWE-367). SO_PEERCRED is + read from the kernel for this established connection, so there is + nothing left to swap: it is the identity of the process actually on the + other end, which is what the ownership check was trying to establish. + + Skipped where the platform does not expose SO_PEERCRED; the pathname + checks still apply there. + """ + peercred = getattr(socket, "SO_PEERCRED", None) + if peercred is None: + return + try: + raw = sock.getsockopt(socket.SOL_SOCKET, peercred, struct.calcsize(_UCRED)) + except OSError: + return + _pid, uid, _gid = struct.unpack(_UCRED, raw) + euid = os.geteuid() + if uid != euid: + raise TuiuiConduitError( + f"refusing apphost at {path}: peer runs as uid {uid}, not {euid}" + ) + + class TuiuiConduit: """Synchronous client for the tuiui apphost Unix socket. @@ -186,10 +234,13 @@ def __init__( self.timeout = timeout #: Frames dropped because no consumer kept up with ``frame_backlog``. self.dropped_frames = 0 + #: Replies dropped because no waiter claimed them within the backlog. + self.dropped_replies = 0 self._sock: socket.socket | None = None self._read_buf = b"" self._req_counter = 0 self._lock = threading.RLock() + self._conn_lock = threading.RLock() self._cv = threading.Condition() self._frames: deque[dict[str, Any]] = deque(maxlen=max(1, frame_backlog)) self._replies: deque[dict[str, Any]] = deque(maxlen=_REPLY_BACKLOG) @@ -208,70 +259,88 @@ def __exit__(self, exc_type, exc, tb) -> None: def connect(self) -> None: """Open the socket and start the demultiplexing reader. + The pathname is verified before connecting and the connected peer's + uid after, so a swap between the two lookups cannot get an + attacker's listener talking to this client. The whole sequence is + one atomic step under ``_conn_lock``, so two threads racing through + ``__enter__`` cannot both open a socket and leak one of them. + Raises :class:`TuiuiConduitError` when the socket is not ours to talk to, or when the connection itself fails: connection failures are the most common error a caller sees, so they belong to the same exception contract as protocol errors rather than leaking raw ``OSError``. """ - if self._sock is not None: - return - _verify_socket_owner(self.socket_path) - try: - sock = _socket_connect(self.socket_path, self.timeout) - except OSError as exc: - raise TuiuiConduitError( - f"cannot connect to apphost at {self.socket_path}: {exc}" - ) from exc - self._read_buf = b"" - self._stopping.clear() - with self._cv: - self._frames.clear() - self._replies.clear() - self._reader_error = None - self._closed = False - self._sock = sock - sock.settimeout(_READ_POLL_SECS) - self._reader = threading.Thread( - target=self._read_loop, - args=(sock,), - name="tuiui-conduit-reader", - daemon=True, - ) - self._reader.start() - - def close(self) -> None: - """Close the socket and stop the reader thread.""" - self._stopping.set() - sock, self._sock = self._sock, None - if sock is not None: + with self._conn_lock: + if self._sock is not None: + return + _verify_socket_owner(self.socket_path) try: - sock.shutdown(socket.SHUT_RDWR) - except OSError: - pass + sock = _socket_connect(self.socket_path, self.timeout) + except OSError as exc: + raise TuiuiConduitError( + f"cannot connect to apphost at {self.socket_path}: {exc}" + ) from exc try: + _verify_connected_peer(sock, self.socket_path) + except BaseException: sock.close() - except OSError: - pass - reader, self._reader = self._reader, None - if reader is not None and reader is not threading.current_thread(): - reader.join(timeout=_READ_POLL_SECS * 4) - self._read_buf = b"" + raise + self._read_buf = b"" + self._stopping.clear() + with self._cv: + self._frames.clear() + self._replies.clear() + self._reader_error = None + self._closed = False + self._sock = sock + self._reader = threading.Thread( + target=self._read_loop, + args=(sock,), + name="tuiui-conduit-reader", + daemon=True, + ) + self._reader.start() + + def close(self) -> None: + """Close the socket and stop the reader thread.""" + with self._conn_lock: + self._stopping.set() + sock, self._sock = self._sock, None + if sock is not None: + try: + sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass + try: + sock.close() + except OSError: + pass + reader, self._reader = self._reader, None + if reader is not None and reader is not threading.current_thread(): + reader.join(timeout=_READ_POLL_SECS * 4) + self._read_buf = b"" with self._cv: self._closed = True self._cv.notify_all() def _send(self, payload: dict[str, Any]) -> None: - """Encode ``payload`` as one JSON line and write it.""" - sock = self._sock - if sock is None: - raise TuiuiConduitError("not connected") + """Encode ``payload`` as one JSON line and write it. + + ``_conn_lock`` is held across the write so a concurrent + :meth:`close` cannot pull the fd out from under ``sendall`` and turn + a shutdown into what looks like an apphost write failure. + """ line = (json.dumps(payload, separators=(",", ":")) + "\n").encode("utf-8") - with self._lock: - try: - sock.sendall(line) - except OSError as exc: - raise TuiuiConduitError(f"apphost write failed: {exc}") from exc + with self._conn_lock: + sock = self._sock + if sock is None: + raise TuiuiConduitError("not connected") + with self._lock: + try: + sock.sendall(line) + except OSError as exc: + raise TuiuiConduitError(f"apphost write failed: {exc}") from exc def _read_loop(self, sock: socket.socket) -> None: """Own the socket and sort every event into its consumer's backlog. @@ -279,13 +348,14 @@ def _read_loop(self, sock: socket.socket) -> None: This is the only place ``recv()`` and ``_read_buf`` are touched, so no two threads can ever split a JSON line between them or claim bytes meant for the other. Frame events go to the frame backlog, - everything else to the reply backlog; nothing is discarded here. + everything else to the reply backlog; nothing is discarded here + except an oldest entry once a backlog is full, which + :attr:`dropped_frames` and :attr:`dropped_replies` count. """ try: while not self._stopping.is_set(): - try: - evt = self._recv_event(sock) - except TimeoutError: + evt = self._recv_event(sock) + if evt is None: continue with self._cv: if "Frame" in evt: @@ -293,6 +363,8 @@ def _read_loop(self, sock: socket.socket) -> None: self.dropped_frames += 1 self._frames.append(evt) else: + if len(self._replies) == self._replies.maxlen: + self.dropped_replies += 1 self._replies.append(evt) self._cv.notify_all() except BaseException as exc: # noqa: BLE001 - handed to every waiter @@ -305,13 +377,17 @@ def _read_loop(self, sock: socket.socket) -> None: self._closed = True self._cv.notify_all() - def _recv_event(self, sock: socket.socket) -> dict[str, Any]: - """Read one full newline-delimited JSON object from the socket. + def _recv_event(self, sock: socket.socket) -> dict[str, Any] | None: + """Read one full newline-delimited JSON object, or None if none is ready. Called only from the reader thread. Frames are pushed by the apphost without a request, so this is the generic receive path: a request that expects a reply also arrives here because the apphost interleaves frames freely. + + Readiness is polled with ``select`` rather than a short socket + timeout, so bounding how long the reader parks does not also bound + how long a write may take. """ while True: if b"\n" in self._read_buf: @@ -322,6 +398,9 @@ def _recv_event(self, sock: socket.socket) -> dict[str, Any]: return json.loads(line.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise TuiuiConduitError(f"bad frame: {exc!r}") from exc + ready, _, _ = select.select([sock], [], [], _READ_POLL_SECS) + if not ready: + return None chunk = sock.recv(65536) if not chunk: raise TuiuiConduitError("apphost closed connection") @@ -371,7 +450,11 @@ def spawn( with self._lock: self._send(payload) evt = self._wait_for_matching(self._match_spawned, timeout=timeout) - return SpawnedApp(app=int(evt["Spawned"]["app"]), pid=int(evt["Spawned"]["pid"])) + spawned = evt["Spawned"] + return SpawnedApp( + app=_as_int(spawned["app"], "Spawned.app"), + pid=_as_int(spawned["pid"], "Spawned.pid"), + ) @staticmethod def _match_spawned(evt: dict[str, Any]) -> bool: @@ -428,13 +511,13 @@ def _match_roster(evt: dict[str, Any]) -> bool: @staticmethod def _parse_roster(raw: dict[str, Any]) -> RosterEntry: return RosterEntry( - app=int(raw["app"]), + app=_as_int(raw["app"], "Roster.app"), cmd=str(raw.get("cmd", "")), args=list(raw.get("args", [])), - pid=int(raw.get("pid", 0)), - cols=int(raw.get("cols", 0)), - rows=int(raw.get("rows", 0)), - age_secs=int(raw.get("age_secs", 0)), + pid=_as_int(raw.get("pid", 0), "Roster.pid"), + cols=_as_int(raw.get("cols", 0), "Roster.cols"), + rows=_as_int(raw.get("rows", 0), "Roster.rows"), + age_secs=_as_int(raw.get("age_secs", 0), "Roster.age_secs"), alive=bool(raw.get("alive", True)), meta=raw.get("meta"), ) @@ -494,17 +577,24 @@ def _parse_frame(cls, raw: dict[str, Any]) -> Frame: cursor = raw.get("cursor") cur_pair: tuple[int, int] | None = None if isinstance(cursor, (list, tuple)) and len(cursor) == 2: - cur_pair = (int(cursor[0]), int(cursor[1])) + cur_pair = ( + _as_int(cursor[0], "Frame.cursor[0]"), + _as_int(cursor[1], "Frame.cursor[1]"), + ) return Frame( cells=cells, cols=cols, rows=rows, cursor=cur_pair, - flags=int(raw.get("flags", 0)), + flags=_as_int(raw.get("flags", 0), "Frame.flags"), images=list(raw.get("images", [])), image_data=list(raw.get("image_data", [])), clear=bool(raw.get("clear", False)), - switch_to=(int(raw["switch_to"]) if raw.get("switch_to") is not None else None), + switch_to=( + _as_int(raw["switch_to"], "Frame.switch_to") + if raw.get("switch_to") is not None + else None + ), clipboard=(str(raw["clipboard"]) if raw.get("clipboard") is not None else None), truncated=truncated, ) @@ -596,14 +686,18 @@ def _extract_grid(grid: dict[str, Any]) -> tuple[list[str], int, int, bool]: declared_cols = grid.get("cols") declared_rows = grid.get("rows") if declared_cols is not None: - cols = int(declared_cols) - elif declared_rows is not None and int(declared_rows) > 0: - cols = _ceil_div(len(cells), int(declared_rows)) + cols = _as_int(declared_cols, "grid.cols") + elif declared_rows is not None and _as_int(declared_rows, "grid.rows") > 0: + cols = _ceil_div(len(cells), _as_int(declared_rows, "grid.rows")) else: cols = len(cells) if cols <= 0: return [], 0, 0, False - rows = int(declared_rows) if declared_rows is not None else _ceil_div(len(cells), cols) + rows = ( + _as_int(declared_rows, "grid.rows") + if declared_rows is not None + else _ceil_div(len(cells), cols) + ) return _fit(cells, cols, max(0, rows)) rows_raw = grid.get("rows_list") or grid.get("rows") if isinstance(rows_raw, list): @@ -618,7 +712,7 @@ def _extract_grid(grid: dict[str, Any]) -> tuple[list[str], int, int, bool]: per_row.append([_cell_char(c) for c in raw_cells]) declared_cols = grid.get("cols") if declared_cols is not None: - cols = int(declared_cols) + cols = _as_int(declared_cols, "grid.cols") else: cols = max((len(r) for r in per_row), default=0) cells: list[str] = []