diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index d186ce559b..ec4094cf73 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -12,8 +12,6 @@ Usage: from contextlib import asynccontextmanager, contextmanager import asyncio -import atexit -import concurrent.futures import functools from collections import deque from dataclasses import dataclass @@ -40,6 +38,7 @@ import urllib.parse from hermes_cli._subprocess_compat import windows_detach_flags from hermes_cli.install_identity import get_install_id as _shared_get_install_id +from hermes_cli.pty_session import run_reaper import urllib.request from pathlib import Path from typing import Any, Dict, List, Optional, Tuple @@ -94,7 +93,7 @@ except ImportError: from tools.lazy_deps import ensure as _lazy_ensure _lazy_ensure("tool.dashboard", prompt=False) from fastapi import ( - FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect, + FastAPI, HTTPException, Request, ) from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse @@ -7001,715 +7000,34 @@ def _aux_task_summary(aux_rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: return result -# --------------------------------------------------------------------------- -# /api/pty — PTY-over-WebSocket bridge for the dashboard "Chat" tab. -# -# The endpoint spawns the same ``hermes --tui`` binary the CLI uses, behind -# a POSIX pseudo-terminal, and forwards bytes + resize escapes across a -# WebSocket. The browser renders the ANSI through xterm.js (see -# web/src/pages/ChatPage.tsx). -# -# Auth: ``?token=`` query param (browsers can't set -# Authorization on the WS upgrade). Same ephemeral ``_SESSION_TOKEN`` as -# REST. Localhost-only — we defensively reject non-loopback clients even -# though uvicorn binds to 127.0.0.1. -# --------------------------------------------------------------------------- - -# PTY bridge: POSIX uses pty_bridge (fcntl/termios/ptyprocess); native Windows -# uses win_pty_bridge (pywinpty/ConPTY, already a declared dependency). Both -# expose the same public surface — spawn/read/write/resize/close/is_available — -# so the /api/pty WebSocket handler needs no platform guards. -if sys.platform.startswith("win"): - try: - from hermes_cli.win_pty_bridge import WinPtyBridge as PtyBridge, PtyUnavailableError - _PTY_BRIDGE_AVAILABLE = True - except ImportError: # pragma: no cover - pywinpty missing - PtyBridge = None # type: ignore[assignment] - _PTY_BRIDGE_AVAILABLE = False - - class PtyUnavailableError(RuntimeError): # type: ignore[no-redef] - """Stub when win_pty_bridge cannot be imported.""" - pass -else: - try: - from hermes_cli.pty_bridge import PtyBridge, PtyUnavailableError - _PTY_BRIDGE_AVAILABLE = True - except ImportError: # pragma: no cover - dev env without ptyprocess - PtyBridge = None # type: ignore[assignment] - _PTY_BRIDGE_AVAILABLE = False - - class PtyUnavailableError(RuntimeError): # type: ignore[no-redef] - """Stub on platforms where pty_bridge can't be imported.""" - pass - -_RESIZE_RE = re.compile(rb"\x1b\[RESIZE:(\d+);(\d+)\]") -_PTY_READ_CHUNK_TIMEOUT = 0.2 -# Back-off delay between idle PTY reads so a quiet terminal does not spin -# the event loop. A positive sleep lets other coroutines run and keeps -# dashboard idle CPU low (#42627). -_PTY_IDLE_BACKOFF = 0.05 - -# Keep-alive PTY sessions: a terminal connecting with ``?attach=`` is -# bound to a process that survives disconnect/refresh and is reattachable. -from hermes_cli.pty_session import PtySessionRegistry, run_reaper # noqa: E402 - -PTY_REGISTRY = PtySessionRegistry( - ttl=30 * 60, - max_sessions=16, - buffer_cap=1 * 1024 * 1024, - read_timeout=_PTY_READ_CHUNK_TIMEOUT, +from hermes_cli.web_server_chat import ( # noqa: E402,F401 — re-exported; routers/tests reach these via web_server. + PTY_REGISTRY, + PtyBridge, + PtyUnavailableError, + _GATEWAY_WS_PROTOCOL, + _GATEWAY_WS_TICKET_PROTOCOL_PREFIX, + _LOOPBACK_HOSTS, + _PTY_BRIDGE_AVAILABLE, + _RESIZE_RE, + _WILDCARD_HOSTS, + _active_session_file_for_channel, + _build_gateway_ws_url, + _build_sidecar_url, + _get_console_executor, + _legacy_pump, + _resolve_chat_argv, + _resolve_chat_argv_async, + _resolve_client_ws_host, + _ws_auth_ok, + _ws_auth_reason, + _ws_client_is_allowed, + _ws_client_reason, + _ws_host_origin_is_allowed, + _ws_host_origin_reason, + _ws_request_is_allowed, ) -async def _legacy_pump(ws: "WebSocket", bridge) -> None: - """Original 1:1 socket<->PTY pump: stream until disconnect, then close the - bridge. Used when no ``?attach=`` token is supplied (keep-alive opt-in). - - Behavior is identical to the pre-keep-alive ``pty_ws`` body, including the - #54028 half-open-socket protection (reader EOF → close the WS so the - writer's ``ws.receive()`` unparks) and the #53227 ``to_thread`` offloads - for the blocking ``bridge.close()``. - """ - loop = asyncio.get_running_loop() - - # --- reader task: PTY master → WebSocket ---------------------------- - async def pump_pty_to_ws() -> None: - try: - while True: - chunk = await loop.run_in_executor( - None, bridge.read, _PTY_READ_CHUNK_TIMEOUT - ) - if chunk is None: # EOF - return - if not chunk: # no data this tick; yield control and retry - await asyncio.sleep(_PTY_IDLE_BACKOFF) - continue - try: - await ws.send_bytes(chunk) - except Exception: - return - finally: - # The child has exited (EOF) or the send side broke. Close the - # WebSocket so the writer loop's ``ws.receive()`` returns instead - # of blocking forever — otherwise, when the browser's socket is - # half-open (no FIN delivered, common on macOS/launchd) the - # handler never reaches its ``finally`` and the PTY's fds leak. - # With dashboard auto-reconnect (#52962) every dropped socket then - # stacks a fresh PTY on top of the orphaned one, exhausting fds. - # - # Reap the bridge here too (close() is idempotent): on child EOF the - # writer loop's ``finally`` is the usual closer, but if the handler - # task is cancelled the instant we close the WS, that ``finally`` - # can be skipped, leaking the PTY. Closing from the EOF path makes - # the reap independent of that cancellation race (#54028). - try: - await asyncio.to_thread(bridge.close) - except Exception: - pass - try: - await ws.close() - except Exception: - pass - - reader_task = asyncio.create_task(pump_pty_to_ws()) - - # --- writer loop: WebSocket → PTY master ---------------------------- - try: - while True: - try: - msg = await ws.receive() - except RuntimeError: - # Raised when ws.receive() is called after the socket is - # already disconnected (e.g. closed by the reader task above). - break - if msg.get("type") == "websocket.disconnect": - break - raw = msg.get("bytes") - if raw is None: - text = msg.get("text") - raw = text.encode("utf-8") if isinstance(text, str) else b"" - if not raw: - continue - # Resize escape is consumed locally, never written to the PTY. - match = _RESIZE_RE.match(raw) - if match and match.end() == len(raw): - bridge.resize(cols=int(match.group(1)), rows=int(match.group(2))) - continue - bridge.write(raw) - except WebSocketDisconnect: - pass - finally: - reader_task.cancel() - try: - await reader_task - except (asyncio.CancelledError, Exception): - pass - await asyncio.to_thread(bridge.close) - - -# Starlette's TestClient reports the peer as "testclient"; treat it as -# loopback so tests don't need to rewrite request scope. -_LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1", "localhost", "testclient"}) - - -def _ws_client_reason(ws: "WebSocket") -> Optional[str]: - """Return a rejection reason for the client IP, or None when allowed. - - Reasons are short machine-parseable tokens logged on the rejection path - so a "WS keeps closing" report can be diagnosed from agent.log without a - repro. ``None`` means the peer IP passed this gate. - - See :func:`_ws_client_is_allowed` for the full policy rationale. - """ - if getattr(app.state, "auth_required", False): - return None - bound_host = (getattr(app.state, "bound_host", "") or "").strip().lower() - if bound_host and bound_host not in _LOOPBACK_HOSTS: - return None - client_host = ws.client.host if ws.client else "" - if not client_host: - # Fail-closed: a loopback-bound dashboard with auth disabled must - # not accept a WebSocket with no identifiable peer. ASGI servers - # behind a misconfigured proxy or unix socket can deliver - # ws.client == None or "" — treating that as "allowed" would let - # an unidentified peer reach a loopback-only surface. - return f"missing_or_empty_peer bound={bound_host or '?'}" - if client_host in _LOOPBACK_HOSTS: - return None - return f"peer_not_loopback peer={client_host} bound={bound_host or '?'}" - - -def _ws_client_is_allowed(ws: "WebSocket") -> bool: - """Check if the WebSocket client IP is acceptable. - - Loopback bind: only loopback clients allowed — the legacy - ``?token=<_SESSION_TOKEN>`` path is the only auth we have, so we - don't want LAN hosts guessing tokens. - - Explicit non-loopback bind (``--host 0.0.0.0``, ``--host ::``, or a - specific address such as a Tailscale/LAN IP, always with - ``--insecure``): allow any peer. The operator explicitly opted into - non-loopback exposure, so the loopback-only peer restriction does not - apply. DNS-rebinding is still blocked by the Host/Origin guard in - :func:`_ws_host_origin_is_allowed`, which mirrors the HTTP layer and - requires the Host header to match the bound interface — the same - defence ``_is_accepted_host`` applies to non-loopback HTTP requests. - - Gated mode: any peer is allowed — uvicorn's ``proxy_headers=True`` - (enabled when the OAuth gate is active so cookies can pick up - ``X-Forwarded-Proto``) rewrites ``ws.client.host`` to the - X-Forwarded-For value, which is the real internet client IP. The - OAuth gate + single-use ``?ticket=`` is the auth at that point; the - Host/Origin guard in :func:`_ws_host_origin_is_allowed` is what - blocks DNS-rebinding here, not the peer IP. - """ - if getattr(app.state, "auth_required", False): - return True - # Any explicit non-loopback bind (0.0.0.0, ::, or a specific LAN / - # Tailscale address) means the operator opted into non-loopback - # access via --insecure. The loopback-only peer gate only applies to - # an actual loopback bind; otherwise the WS handshake is rejected even - # though same-bind HTTP requests pass _is_accepted_host. - bound_host = (getattr(app.state, "bound_host", "") or "").strip().lower() - if bound_host and bound_host not in _LOOPBACK_HOSTS: - return True - client_host = ws.client.host if ws.client else "" - if not client_host: - # Fail-closed: see _ws_client_reason for rationale. An empty - # client_host on a loopback-bound dashboard with auth disabled - # must be rejected, not accepted as a default-allow. - return False - return client_host in _LOOPBACK_HOSTS - - -def _ws_host_origin_reason(ws: "WebSocket") -> Optional[str]: - """Return a Host/Origin rejection reason, or None when allowed. - - Mirrors :func:`_ws_host_origin_is_allowed` but yields a short - machine-parseable token (``host_mismatch …`` / ``origin_mismatch …``) - on rejection so the close path can log *why* the upgrade was refused. - """ - bound_host = getattr(app.state, "bound_host", None) - if not bound_host: - return None - - trusted_public_hosts = getattr( - app.state, "trusted_public_hosts", frozenset() - ) - - host_header = ws.headers.get("host", "") - if not _is_accepted_host( - host_header, bound_host, trusted_public_hosts - ): - return f"host_mismatch host={host_header or '?'} bound={bound_host}" - - origin = ws.headers.get("origin", "") - if not origin: - return None - - parsed = urllib.parse.urlparse(origin) - if parsed.scheme not in {"http", "https"}: - # Non-web origin (packaged Electron: file://, null, app://). The - # upstream credential check is the real auth boundary; trust it. - # See _ws_host_origin_is_allowed for the full rationale. - return None - - if not parsed.netloc: - return f"origin_mismatch origin={origin} bound={bound_host}" - - if not _is_accepted_host( - parsed.netloc, bound_host, trusted_public_hosts - ): - return f"origin_mismatch origin={origin} bound={bound_host}" - return None - - -def _ws_host_origin_is_allowed(ws: "WebSocket") -> bool: - """Apply the dashboard Host/Origin guard to WebSocket upgrades. - - FastAPI HTTP middleware does not run for WebSocket routes, so the - DNS-rebinding Host check used for normal dashboard HTTP requests must be - repeated here before accepting the upgrade. Browsers also send an Origin - header on WebSocket handshakes; when present, require it to target the - same bound dashboard host. - """ - return _ws_host_origin_reason(ws) is None - - -def _ws_request_is_allowed(ws: "WebSocket") -> bool: - """Return True when the WebSocket upgrade matches dashboard boundaries.""" - return _ws_host_origin_is_allowed(ws) and _ws_client_is_allowed(ws) - - -_GATEWAY_WS_PROTOCOL = "hermes-gateway-v1" -_GATEWAY_WS_TICKET_PROTOCOL_PREFIX = "hermes-gateway-ticket." - - -def _gateway_ws_ticket_from_subprotocol(ws: "WebSocket") -> tuple[str, str]: - """Return ``(ticket, reason)`` from an unambiguous gateway protocol set.""" - raw = str(ws.headers.get("sec-websocket-protocol", "") or "") - protocols = [value.strip() for value in raw.split(",") if value.strip()] - ticket_protocols = [ - value for value in protocols - if value.startswith(_GATEWAY_WS_TICKET_PROTOCOL_PREFIX) - ] - if not ticket_protocols: - return "", "none" - if _GATEWAY_WS_PROTOCOL not in protocols or len(ticket_protocols) != 1: - return "", "invalid" - ticket = ticket_protocols[0][len(_GATEWAY_WS_TICKET_PROTOCOL_PREFIX):] - return (ticket, "ok") if ticket else ("", "invalid") - - -def _ws_auth_reason(ws: "WebSocket") -> tuple[Optional[str], str]: - """Validate WS-upgrade auth; return ``(reason, credential)``. - - ``reason`` is None when the credential is accepted, else a short - machine-parseable token explaining the rejection (``no_credential``, - ``token_mismatch``, ``ticket_invalid``, ``internal_invalid``). - ``credential`` names which credential type was presented (``ticket``, - ``internal``, ``token``, or ``none``) so the accepted path can log *how* - a peer authed, not just that it did. - - Loopback / ``--insecure``: legacy ``?token=<_SESSION_TOKEN>`` query - parameter, constant-time compared. - - Gated (public bind, no ``--insecure``): one of two credentials — - - * ``?ticket=`` — a browser-minted, single-use, 30s-TTL ticket - consumed against the dashboard-auth ticket store. This is what the SPA - (and native clients) use. - * ``?internal=`` — the process-lifetime internal - credential, used only by WS clients the server spawns itself (the - embedded-TUI PTY child attaching to ``/api/ws`` and ``/api/pub``). It - is multi-use and never expires so the child can reconnect, and is never - injected into the SPA — see ``dashboard_auth.ws_tickets`` for the - threat model. - - The legacy ``?token=`` path is unconditionally rejected in gated mode - (the SPA bundle isn't carrying the token any longer, and a leaked - ``_SESSION_TOKEN`` must not grant WS access once the gate is engaged). - - Audit-logs the rejection so operators can debug "WS keeps closing" - issues from the log. - """ - auth_required = bool(getattr(app.state, "auth_required", False)) - if auth_required: - # Lazy import — keeps this function importable in test harnesses - # that don't bring in the dashboard_auth layer. - from hermes_cli.dashboard_auth.audit import AuditEvent, audit_log - from hermes_cli.dashboard_auth.ws_tickets import ( - TicketInvalid, - consume_internal_credential, - consume_ticket, - ) - - # Server-spawned children (PTY child → /api/ws, /api/pub) present the - # multi-use internal credential rather than a single-use ticket, so - # they survive reconnects and slow cold boots. - internal = ws.query_params.get("internal", "") - if internal: - try: - info = consume_internal_credential(internal) - # Stamp the server-minted identity onto the WS object so the - # connection (and any transport built from it) can never be - # impersonated by RPC params. Internal peers are marked - # ``server-internal`` and are excluded from privileged - # controller registration downstream. - ws._hermes_auth_identity = { - "user_id": info.get("user_id"), - "provider": info.get("provider"), - } - return None, "internal" - except TicketInvalid as exc: - audit_log( - AuditEvent.WS_TICKET_REJECTED, - reason=f"internal: {exc}", - ip=(ws.client.host if ws.client else ""), - path=ws.url.path, - ) - return "internal_invalid", "internal" - - protocol_ticket, protocol_reason = _gateway_ws_ticket_from_subprotocol(ws) - if protocol_reason == "invalid": - return "ticket_invalid", "ticket-subprotocol" - ticket = protocol_ticket or ws.query_params.get("ticket", "") - if not ticket: - return "no_credential", "none" - - try: - info = consume_ticket(ticket) - # The ticket binds a server-minted {user_id, provider}; stamp it - # onto the WS object so ``gateway_ws`` can hand it to the gateway - # transport, where it is the sole identity authority for - # browser-controller registration. A client can never supply or - # spoof this value through RPC params. Only the two identity - # fields are carried — bookkeeping (e.g. ``minted_at``) is not - # part of the identity contract. - ws._hermes_auth_identity = { - "user_id": info.get("user_id"), - "provider": info.get("provider"), - } - if protocol_ticket: - # Select only the stable public protocol during accept. The - # ticket-bearing protocol is a credential and must never be - # reflected back to the browser or retained after admission. - ws._hermes_ws_subprotocol = _GATEWAY_WS_PROTOCOL - return None, "ticket-subprotocol" - return None, "ticket" - except TicketInvalid as exc: - audit_log( - AuditEvent.WS_TICKET_REJECTED, - reason=str(exc), - ip=(ws.client.host if ws.client else ""), - path=ws.url.path, - ) - return "ticket_invalid", "ticket" - - token = ws.query_params.get("token", "") - if not token: - return "no_credential", "none" - if hmac.compare_digest(token.encode(), _SESSION_TOKEN.encode()): - return None, "token" - return "token_mismatch", "token" - - -def _ws_auth_ok(ws: "WebSocket") -> bool: - """True when the WS-upgrade credential is accepted. See _ws_auth_reason.""" - return _ws_auth_reason(ws)[0] is None - -# Per-channel subscriber registry used by /api/pub (PTY-side gateway → dashboard) -# and /api/events (dashboard → browser sidebar). Keyed by an opaque channel id -# the chat tab generates on mount; entries auto-evict when the last subscriber -# drops AND the publisher has disconnected. -# (Channel state and the chat-argv lock are initialised in _lifespan on app -# startup — see _get_event_state / _get_chat_argv_lock above.) - - -def _resolve_chat_argv( - resume: Optional[str] = None, - sidecar_url: Optional[str] = None, - profile: Optional[str] = None, - active_session_file: Optional[str] = None, -) -> tuple[list[str], Optional[str], Optional[dict]]: - """Resolve the argv + cwd + env for the chat PTY. - - Default: whatever ``hermes --tui`` would run. Tests monkeypatch this - function to inject a tiny fake command (``cat``, ``sh -c 'printf …'``) - so nothing has to build Node or the TUI bundle. - - Session resume is propagated via the ``HERMES_TUI_RESUME`` env var — - matching what ``hermes_cli.main._launch_tui`` does for the CLI path. - Appending ``--resume `` to argv doesn't work because ``ui-tui`` does - not parse its argv. - - ``HERMES_TUI_GATEWAY_URL`` is injected so the PTY child can attach to - this process's in-memory ``tui_gateway`` instance instead of spawning - its own Python gateway subprocess. - - `sidecar_url` (when set) is forwarded as ``HERMES_TUI_SIDECAR_URL`` so - the spawned ``tui_gateway.entry`` can mirror dispatcher emits to the - dashboard's ``/api/pub`` endpoint (see :func:`pub_ws`). - - `active_session_file` (when set) is forwarded as - ``HERMES_TUI_ACTIVE_SESSION_FILE``. The TUI writes the current session id - there whenever it creates/resumes/switches sessions, giving the dashboard a - small cross-process breadcrumb for reconnecting after an unexpected browser - WebSocket close. - - `profile` (when set) scopes the ENTIRE chat to that profile by pointing - ``HERMES_HOME`` at the profile dir in the child env. Every spawned - process (the TUI and the ``tui_gateway.entry`` it launches) resolves - ``get_hermes_home()`` from that env var at its own import, so the child - binds the profile's config, skills, memory, and state.db from the start - — the same propagation ``hermes -p `` performs. The in-process - ``HERMES_TUI_GATEWAY_URL`` attach is SKIPPED for scoped chats: the - dashboard's in-memory gateway runs under the dashboard's own profile, - so a profile-scoped chat must spawn its own gateway subprocess. - """ - from hermes_cli.main import PROJECT_ROOT, _apply_tui_python_env, _make_tui_argv - - profile_dir: Optional[Path] = None - requested = (profile or "").strip() - if requested and requested.lower() != "current": - profile_dir = _resolve_profile_dir(requested) - - argv, cwd = _make_tui_argv(PROJECT_ROOT / "ui-tui", tui_dev=False) - # Hermes TUI child: build via the single spawn-env factory (profile-home - # contract applied; secrets kept — the spawned agent needs provider creds). - # An explicit profile scope still overrides HERMES_HOME before config is - # bridged into the child environment. - from tools.environments.local import build_subprocess_env - env = build_subprocess_env(scrub_secrets=False, inherit_profile_home=True) - if profile_dir is not None: - env["HERMES_HOME"] = str(profile_dir) - try: - from hermes_cli.config import ( - apply_terminal_config_to_env, - read_raw_config, - terminal_config_owned_env_vars, - ) - - if profile_dir is not None: - # The dashboard process already bridged its own terminal config - # into os.environ at startup. Remove only keys explicitly owned by - # that launch profile before applying the selected profile. Values - # exported by the operator for keys omitted from the launch profile - # remain valid fallbacks, matching apply_terminal_config_to_env(). - raw_launch_terminal = read_raw_config().get("terminal") - for env_var in terminal_config_owned_env_vars(raw_launch_terminal): - env.pop(env_var, None) - with _config_profile_scope(requested): - apply_terminal_config_to_env(env=env) - else: - apply_terminal_config_to_env(env=env) - except Exception: - _log.warning("Failed to apply terminal config bridge for dashboard chat", exc_info=True) - _apply_tui_python_env(env) - env.setdefault("NODE_ENV", "production") - # Browser-embedded chat should prefer stable wheel-based scrollback over - # native terminal mouse tracking. When mouse tracking is enabled, wheel - # events are consumed by the TUI and forwarded as terminal input, which - # makes browser-side transcript scrolling feel broken. Keep the terminal - # build unchanged for native CLI usage; only disable mouse tracking for - # the dashboard PTY path. - env.setdefault("HERMES_TUI_DISABLE_MOUSE", "1") - env.setdefault("HERMES_TUI_INLINE", "1") - # The dashboard terminal is xterm.js, which always renders 24-bit RGB. - # But chalk inside the TUI child decides its color depth from the - # SERVER process env — and hosted/cloud deploys run the dashboard under - # a process manager (container init, systemd) with no COLORTERM, so - # chalk downgrades every hex color to the xterm 256 palette. The skin's - # bronze border #CD7F32 snaps to palette 173 (#D7875F, salmon-red) and - # the banner reads red/yellow instead of gold. Local launches dodge - # this only because the operator's interactive terminal leaks - # COLORTERM=truecolor into os.environ. Backfill it for the PTY child; - # setdefault so an explicit operator value still wins. - env.setdefault("COLORTERM", "truecolor") - env["HERMES_TUI_DASHBOARD"] = "1" - - if resume: - _resume_db = _open_session_db_for_profile( - requested if profile_dir is not None else None, - read_only=True, - ) - try: - latest_resume, _latest_path = _session_latest_descendant(resume, _resume_db) - finally: - _resume_db.close() - if latest_resume: - resume = latest_resume - env["HERMES_TUI_RESUME"] = resume - - if sidecar_url: - env["HERMES_TUI_SIDECAR_URL"] = sidecar_url - - if active_session_file: - env["HERMES_TUI_ACTIVE_SESSION_FILE"] = active_session_file - - # Profile-scoped chats must NOT attach to the dashboard's in-memory - # gateway — it runs under the dashboard's own profile. Without the - # attach URL, gatewayClient spawns its own `tui_gateway.entry`, which - # inherits the profile HERMES_HOME set above. - if profile_dir is None: - if gateway_ws_url := _build_gateway_ws_url(): - env["HERMES_TUI_GATEWAY_URL"] = gateway_ws_url - - return list(argv), str(cwd) if cwd else None, env - - -# Hosts that mean "listen on every interface" — the server should bind to -# them, but an in-container client must NOT dial them: dialing 0.0.0.0 -# resolves to "any local interface", which on most platforms routes through -# the kernel's wildcard stack and behind a forward proxy (HTTPS_PROXY with -# a NO_PROXY that doesn't list 0.0.0.0) gets MITM'd into a failed handshake -# (issue #58993). The fix is to use a loopback address for the client -# netloc while leaving the bind host alone. -_WILDCARD_HOSTS = frozenset({"0.0.0.0", "::"}) - - -def _resolve_client_ws_host() -> Optional[str]: - """Return the host the in-container WS client should dial. - - Resolution order: - - 1. Explicit ``HERMES_DASHBOARD_WS_HOST`` env var — wins always. Operators - running the dashboard behind a forward proxy can pin a routable host - (e.g. ``127.0.0.1``, the container's internal IP, or a sidecar DNS - name) and bypass auto-detection entirely. - 2. The configured bind host — if it's a wildcard (``0.0.0.0`` / ``::``), - substitute ``127.0.0.1`` since both the dashboard and its TUI child - run in the same container. - 3. Any other bind host (loopback or LAN IP) — preserved verbatim. - """ - explicit = os.environ.get("HERMES_DASHBOARD_WS_HOST", "").strip() - if explicit: - return explicit - - host = getattr(app.state, "bound_host", None) - if not host: - return None - - if host in _WILDCARD_HOSTS: - return "127.0.0.1" - - return host - - -def _build_gateway_ws_url() -> Optional[str]: - """ws:// URL the PTY child should attach to for JSON-RPC gateway traffic. - - Loopback / ``--insecure``: ``?token=<_SESSION_TOKEN>``. - - Gated mode: the legacy token path is rejected by ``_ws_auth_ok``, so the - server-spawned PTY child authenticates with the process-lifetime internal - credential (``?internal=``). It must NOT use a single-use browser ticket: - the child reads this URL once at startup and reuses it on every reconnect, - and a 30s-TTL ticket can expire before a slow cold boot even dials. - """ - host = _resolve_client_ws_host() - port = getattr(app.state, "bound_port", None) - - if not host or not port: - return None - - netloc = ( - f"[{host}]:{port}" - if ":" in host and not host.startswith("[") - else f"{host}:{port}" - ) - - if getattr(app.state, "auth_required", False): - from hermes_cli.dashboard_auth.ws_tickets import internal_ws_credential - - qs = urllib.parse.urlencode({"internal": internal_ws_credential()}) - else: - qs = urllib.parse.urlencode({"token": _SESSION_TOKEN}) - - return f"ws://{netloc}/api/ws?{qs}" - - -async def _resolve_chat_argv_async( - resume: Optional[str] = None, - sidecar_url: Optional[str] = None, - profile: Optional[str] = None, - active_session_file: Optional[str] = None, -) -> tuple[list[str], Optional[str], Optional[dict]]: - """Resolve chat argv without blocking the dashboard event loop. - - ``_resolve_chat_argv`` may run ``npm install`` / ``npm run build`` through - ``_make_tui_argv``. Keep that synchronous work off the WebSocket event - loop so reverse proxies and existing dashboard connections can continue - to exchange keepalives while the TUI launch command is prepared. The - async lock preserves the previous one-build-at-a-time behavior when - multiple browser tabs connect at once without occupying worker threads - while queued connections wait. - """ - kwargs = { - "resume": resume, - "sidecar_url": sidecar_url, - "profile": profile, - } - if active_session_file is not None: - kwargs["active_session_file"] = active_session_file - - async with _get_chat_argv_lock(app): - return await asyncio.to_thread( - _resolve_chat_argv, - **kwargs, - ) - - -def _build_sidecar_url(channel: str) -> Optional[str]: - """ws:// URL the PTY child should publish events to, or None when unbound. - - Loopback / ``--insecure``: uses ``?token=<_SESSION_TOKEN>``. - - Gated mode: authenticates with the process-lifetime internal credential - (``?internal=``), the same one ``_build_gateway_ws_url`` uses. The PTY - child is a server-spawned process we trust; the credential is multi-use - and never expires, so the child can reconnect ``/api/pub`` without a new - URL. (This previously minted a single-use 30s ticket, which meant the - child could not reconnect and could miss the window on a slow cold boot.) - Connections authenticated this way are recorded under the - ``server-internal`` identity in the audit log. - """ - host = _resolve_client_ws_host() - port = getattr(app.state, "bound_port", None) - - if not host or not port: - return None - - netloc = f"[{host}]:{port}" if ":" in host and not host.startswith("[") else f"{host}:{port}" - - if getattr(app.state, "auth_required", False): - # Gated mode — use the internal credential so the WS upgrade survives - # _ws_auth_ok and the child can reconnect. - from hermes_cli.dashboard_auth.ws_tickets import internal_ws_credential - - qs = urllib.parse.urlencode( - {"internal": internal_ws_credential(), "channel": channel} - ) - else: - qs = urllib.parse.urlencode({"token": _SESSION_TOKEN, "channel": channel}) - - return f"ws://{netloc}/api/pub?{qs}" - - -def _active_session_file_for_channel(app: "FastAPI", channel: str) -> Path: - """Return the per-channel file where a dashboard TUI writes its active sid.""" - files = _get_pty_active_session_files(app) - existing = files.get(channel) - if existing is not None: - return existing - - fd, raw_path = tempfile.mkstemp(prefix="hermes-pty-active-", suffix=".json") - os.close(fd) - path = Path(raw_path) - files[channel] = path - return path - - from hermes_cli.web_routers import chat_ws as _chat_ws_routes # noqa: E402 app.include_router(_chat_ws_routes.router) @@ -7720,37 +7038,6 @@ from hermes_cli.web_routers.chat_ws import ( # noqa: E402,F401 — legacy re-ex ) -# Console commands run in a worker thread. On a timeout, asyncio.wait_for cancels -# the *awaitable*, but Python threads aren't preemptible, so a genuinely stuck -# worker keeps running to completion. To keep that from exhausting the shared -# default thread pool (asyncio.to_thread), we run console commands on a small -# dedicated, bounded pool: a leaked worker is capped, and concurrent console -# execution is bounded to a fixed number of threads regardless of reconnects. -_CONSOLE_EXECUTOR_MAX_WORKERS = 4 -_console_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None -_console_executor_lock = threading.Lock() - - -def _get_console_executor() -> concurrent.futures.ThreadPoolExecutor: - """Lazily create the bounded console worker pool (once per process).""" - global _console_executor - if _console_executor is None: - with _console_executor_lock: - if _console_executor is None: - _console_executor = concurrent.futures.ThreadPoolExecutor( - max_workers=_CONSOLE_EXECUTOR_MAX_WORKERS, - thread_name_prefix="hermes-console", - ) - # Ensure the pool is torn down on interpreter exit. Don't wait on - # in-flight workers: a stuck 60s console command must not block - # shutdown (cancel_futures drops anything not yet started). - atexit.register( - lambda: _console_executor - and _console_executor.shutdown(wait=False, cancel_futures=True) - ) - return _console_executor - - from hermes_cli.web_server_dashboard import ( # noqa: E402,F401 — re-exported; routers/tests reach these via web_server. _BUILTIN_DASHBOARD_THEMES, _THEME_COMPONENT_BUCKETS, diff --git a/hermes_cli/web_server_chat.py b/hermes_cli/web_server_chat.py new file mode 100644 index 0000000000..53a21fdc32 --- /dev/null +++ b/hermes_cli/web_server_chat.py @@ -0,0 +1,778 @@ +"""Chat/terminal WebSocket plumbing: PTY bridge selection and registry, WS client/origin/auth gates, chat argv resolution, gateway/sidecar URL building. + +Split out of ``hermes_cli.web_server``; every externally used name is re-imported +there, so ``web_server.`` keeps resolving (and monkeypatching) as before. +Helpers that tests patch on ``web_server`` are reached lazily through it. +""" + +import logging +import asyncio +import atexit +import concurrent.futures +import hmac +import os +import re +import sys +import tempfile +import threading +import urllib.request +from fastapi import FastAPI, WebSocket, WebSocketDisconnect +from pathlib import Path +from typing import Optional +from hermes_cli.pty_session import PtySessionRegistry + +# Same logger the code used before extraction (record parity). +_log = logging.getLogger("hermes_cli.web_server") + + +# --------------------------------------------------------------------------- +# /api/pty — PTY-over-WebSocket bridge for the dashboard "Chat" tab. +# +# The endpoint spawns the same ``hermes --tui`` binary the CLI uses, behind +# a POSIX pseudo-terminal, and forwards bytes + resize escapes across a +# WebSocket. The browser renders the ANSI through xterm.js (see +# web/src/pages/ChatPage.tsx). +# +# Auth: ``?token=`` query param (browsers can't set +# Authorization on the WS upgrade). Same ephemeral ``_SESSION_TOKEN`` as +# REST. Localhost-only — we defensively reject non-loopback clients even +# though uvicorn binds to 127.0.0.1. +# --------------------------------------------------------------------------- + +# PTY bridge: POSIX uses pty_bridge (fcntl/termios/ptyprocess); native Windows +# uses win_pty_bridge (pywinpty/ConPTY, already a declared dependency). Both +# expose the same public surface — spawn/read/write/resize/close/is_available — +# so the /api/pty WebSocket handler needs no platform guards. +if sys.platform.startswith("win"): + try: + from hermes_cli.win_pty_bridge import WinPtyBridge as PtyBridge, PtyUnavailableError + _PTY_BRIDGE_AVAILABLE = True + except ImportError: # pragma: no cover - pywinpty missing + PtyBridge = None # type: ignore[assignment] + _PTY_BRIDGE_AVAILABLE = False + + class PtyUnavailableError(RuntimeError): # type: ignore[no-redef] + """Stub when win_pty_bridge cannot be imported.""" + pass +else: + try: + from hermes_cli.pty_bridge import PtyBridge, PtyUnavailableError + _PTY_BRIDGE_AVAILABLE = True + except ImportError: # pragma: no cover - dev env without ptyprocess + PtyBridge = None # type: ignore[assignment] + _PTY_BRIDGE_AVAILABLE = False + + class PtyUnavailableError(RuntimeError): # type: ignore[no-redef] + """Stub on platforms where pty_bridge can't be imported.""" + pass +_RESIZE_RE = re.compile(rb"\x1b\[RESIZE:(\d+);(\d+)\]") +_PTY_READ_CHUNK_TIMEOUT = 0.2 + +# Back-off delay between idle PTY reads so a quiet terminal does not spin +# the event loop. A positive sleep lets other coroutines run and keeps +# dashboard idle CPU low (#42627). +_PTY_IDLE_BACKOFF = 0.05 +PTY_REGISTRY = PtySessionRegistry( + ttl=30 * 60, + max_sessions=16, + buffer_cap=1 * 1024 * 1024, + read_timeout=_PTY_READ_CHUNK_TIMEOUT, +) + + +async def _legacy_pump(ws: "WebSocket", bridge) -> None: + """Original 1:1 socket<->PTY pump: stream until disconnect, then close the + bridge. Used when no ``?attach=`` token is supplied (keep-alive opt-in). + + Behavior is identical to the pre-keep-alive ``pty_ws`` body, including the + #54028 half-open-socket protection (reader EOF → close the WS so the + writer's ``ws.receive()`` unparks) and the #53227 ``to_thread`` offloads + for the blocking ``bridge.close()``. + """ + loop = asyncio.get_running_loop() + + # --- reader task: PTY master → WebSocket ---------------------------- + async def pump_pty_to_ws() -> None: + try: + while True: + chunk = await loop.run_in_executor( + None, bridge.read, _PTY_READ_CHUNK_TIMEOUT + ) + if chunk is None: # EOF + return + if not chunk: # no data this tick; yield control and retry + await asyncio.sleep(_PTY_IDLE_BACKOFF) + continue + try: + await ws.send_bytes(chunk) + except Exception: + return + finally: + # The child has exited (EOF) or the send side broke. Close the + # WebSocket so the writer loop's ``ws.receive()`` returns instead + # of blocking forever — otherwise, when the browser's socket is + # half-open (no FIN delivered, common on macOS/launchd) the + # handler never reaches its ``finally`` and the PTY's fds leak. + # With dashboard auto-reconnect (#52962) every dropped socket then + # stacks a fresh PTY on top of the orphaned one, exhausting fds. + # + # Reap the bridge here too (close() is idempotent): on child EOF the + # writer loop's ``finally`` is the usual closer, but if the handler + # task is cancelled the instant we close the WS, that ``finally`` + # can be skipped, leaking the PTY. Closing from the EOF path makes + # the reap independent of that cancellation race (#54028). + try: + await asyncio.to_thread(bridge.close) + except Exception: + pass + try: + await ws.close() + except Exception: + pass + + reader_task = asyncio.create_task(pump_pty_to_ws()) + + # --- writer loop: WebSocket → PTY master ---------------------------- + try: + while True: + try: + msg = await ws.receive() + except RuntimeError: + # Raised when ws.receive() is called after the socket is + # already disconnected (e.g. closed by the reader task above). + break + if msg.get("type") == "websocket.disconnect": + break + raw = msg.get("bytes") + if raw is None: + text = msg.get("text") + raw = text.encode("utf-8") if isinstance(text, str) else b"" + if not raw: + continue + # Resize escape is consumed locally, never written to the PTY. + match = _RESIZE_RE.match(raw) + if match and match.end() == len(raw): + bridge.resize(cols=int(match.group(1)), rows=int(match.group(2))) + continue + bridge.write(raw) + except WebSocketDisconnect: + pass + finally: + reader_task.cancel() + try: + await reader_task + except (asyncio.CancelledError, Exception): + pass + await asyncio.to_thread(bridge.close) + + +# Starlette's TestClient reports the peer as "testclient"; treat it as +# loopback so tests don't need to rewrite request scope. +_LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1", "localhost", "testclient"}) + + +def _ws_client_reason(ws: "WebSocket") -> Optional[str]: + """Return a rejection reason for the client IP, or None when allowed. + + Reasons are short machine-parseable tokens logged on the rejection path + so a "WS keeps closing" report can be diagnosed from agent.log without a + repro. ``None`` means the peer IP passed this gate. + + See :func:`_ws_client_is_allowed` for the full policy rationale. + """ + from hermes_cli.web_server import app + if getattr(app.state, "auth_required", False): + return None + bound_host = (getattr(app.state, "bound_host", "") or "").strip().lower() + if bound_host and bound_host not in _LOOPBACK_HOSTS: + return None + client_host = ws.client.host if ws.client else "" + if not client_host: + # Fail-closed: a loopback-bound dashboard with auth disabled must + # not accept a WebSocket with no identifiable peer. ASGI servers + # behind a misconfigured proxy or unix socket can deliver + # ws.client == None or "" — treating that as "allowed" would let + # an unidentified peer reach a loopback-only surface. + return f"missing_or_empty_peer bound={bound_host or '?'}" + if client_host in _LOOPBACK_HOSTS: + return None + return f"peer_not_loopback peer={client_host} bound={bound_host or '?'}" + + +def _ws_client_is_allowed(ws: "WebSocket") -> bool: + """Check if the WebSocket client IP is acceptable. + + Loopback bind: only loopback clients allowed — the legacy + ``?token=<_SESSION_TOKEN>`` path is the only auth we have, so we + don't want LAN hosts guessing tokens. + + Explicit non-loopback bind (``--host 0.0.0.0``, ``--host ::``, or a + specific address such as a Tailscale/LAN IP, always with + ``--insecure``): allow any peer. The operator explicitly opted into + non-loopback exposure, so the loopback-only peer restriction does not + apply. DNS-rebinding is still blocked by the Host/Origin guard in + :func:`_ws_host_origin_is_allowed`, which mirrors the HTTP layer and + requires the Host header to match the bound interface — the same + defence ``_is_accepted_host`` applies to non-loopback HTTP requests. + + Gated mode: any peer is allowed — uvicorn's ``proxy_headers=True`` + (enabled when the OAuth gate is active so cookies can pick up + ``X-Forwarded-Proto``) rewrites ``ws.client.host`` to the + X-Forwarded-For value, which is the real internet client IP. The + OAuth gate + single-use ``?ticket=`` is the auth at that point; the + Host/Origin guard in :func:`_ws_host_origin_is_allowed` is what + blocks DNS-rebinding here, not the peer IP. + """ + from hermes_cli.web_server import app + if getattr(app.state, "auth_required", False): + return True + # Any explicit non-loopback bind (0.0.0.0, ::, or a specific LAN / + # Tailscale address) means the operator opted into non-loopback + # access via --insecure. The loopback-only peer gate only applies to + # an actual loopback bind; otherwise the WS handshake is rejected even + # though same-bind HTTP requests pass _is_accepted_host. + bound_host = (getattr(app.state, "bound_host", "") or "").strip().lower() + if bound_host and bound_host not in _LOOPBACK_HOSTS: + return True + client_host = ws.client.host if ws.client else "" + if not client_host: + # Fail-closed: see _ws_client_reason for rationale. An empty + # client_host on a loopback-bound dashboard with auth disabled + # must be rejected, not accepted as a default-allow. + return False + return client_host in _LOOPBACK_HOSTS + + +def _ws_host_origin_reason(ws: "WebSocket") -> Optional[str]: + """Return a Host/Origin rejection reason, or None when allowed. + + Mirrors :func:`_ws_host_origin_is_allowed` but yields a short + machine-parseable token (``host_mismatch …`` / ``origin_mismatch …``) + on rejection so the close path can log *why* the upgrade was refused. + """ + from hermes_cli.web_server import _is_accepted_host, app + bound_host = getattr(app.state, "bound_host", None) + if not bound_host: + return None + + trusted_public_hosts = getattr( + app.state, "trusted_public_hosts", frozenset() + ) + + host_header = ws.headers.get("host", "") + if not _is_accepted_host( + host_header, bound_host, trusted_public_hosts + ): + return f"host_mismatch host={host_header or '?'} bound={bound_host}" + + origin = ws.headers.get("origin", "") + if not origin: + return None + + parsed = urllib.parse.urlparse(origin) + if parsed.scheme not in {"http", "https"}: + # Non-web origin (packaged Electron: file://, null, app://). The + # upstream credential check is the real auth boundary; trust it. + # See _ws_host_origin_is_allowed for the full rationale. + return None + + if not parsed.netloc: + return f"origin_mismatch origin={origin} bound={bound_host}" + + if not _is_accepted_host( + parsed.netloc, bound_host, trusted_public_hosts + ): + return f"origin_mismatch origin={origin} bound={bound_host}" + return None + + +def _ws_host_origin_is_allowed(ws: "WebSocket") -> bool: + """Apply the dashboard Host/Origin guard to WebSocket upgrades. + + FastAPI HTTP middleware does not run for WebSocket routes, so the + DNS-rebinding Host check used for normal dashboard HTTP requests must be + repeated here before accepting the upgrade. Browsers also send an Origin + header on WebSocket handshakes; when present, require it to target the + same bound dashboard host. + """ + from hermes_cli.web_server import _ws_host_origin_reason + return _ws_host_origin_reason(ws) is None + + +def _ws_request_is_allowed(ws: "WebSocket") -> bool: + """Return True when the WebSocket upgrade matches dashboard boundaries.""" + return _ws_host_origin_is_allowed(ws) and _ws_client_is_allowed(ws) + + +_GATEWAY_WS_PROTOCOL = "hermes-gateway-v1" +_GATEWAY_WS_TICKET_PROTOCOL_PREFIX = "hermes-gateway-ticket." + + +def _gateway_ws_ticket_from_subprotocol(ws: "WebSocket") -> tuple[str, str]: + """Return ``(ticket, reason)`` from an unambiguous gateway protocol set.""" + raw = str(ws.headers.get("sec-websocket-protocol", "") or "") + protocols = [value.strip() for value in raw.split(",") if value.strip()] + ticket_protocols = [ + value for value in protocols + if value.startswith(_GATEWAY_WS_TICKET_PROTOCOL_PREFIX) + ] + if not ticket_protocols: + return "", "none" + if _GATEWAY_WS_PROTOCOL not in protocols or len(ticket_protocols) != 1: + return "", "invalid" + ticket = ticket_protocols[0][len(_GATEWAY_WS_TICKET_PROTOCOL_PREFIX):] + return (ticket, "ok") if ticket else ("", "invalid") + + +def _ws_auth_reason(ws: "WebSocket") -> tuple[Optional[str], str]: + """Validate WS-upgrade auth; return ``(reason, credential)``. + + ``reason`` is None when the credential is accepted, else a short + machine-parseable token explaining the rejection (``no_credential``, + ``token_mismatch``, ``ticket_invalid``, ``internal_invalid``). + ``credential`` names which credential type was presented (``ticket``, + ``internal``, ``token``, or ``none``) so the accepted path can log *how* + a peer authed, not just that it did. + + Loopback / ``--insecure``: legacy ``?token=<_SESSION_TOKEN>`` query + parameter, constant-time compared. + + Gated (public bind, no ``--insecure``): one of two credentials — + + * ``?ticket=`` — a browser-minted, single-use, 30s-TTL ticket + consumed against the dashboard-auth ticket store. This is what the SPA + (and native clients) use. + * ``?internal=`` — the process-lifetime internal + credential, used only by WS clients the server spawns itself (the + embedded-TUI PTY child attaching to ``/api/ws`` and ``/api/pub``). It + is multi-use and never expires so the child can reconnect, and is never + injected into the SPA — see ``dashboard_auth.ws_tickets`` for the + threat model. + + The legacy ``?token=`` path is unconditionally rejected in gated mode + (the SPA bundle isn't carrying the token any longer, and a leaked + ``_SESSION_TOKEN`` must not grant WS access once the gate is engaged). + + Audit-logs the rejection so operators can debug "WS keeps closing" + issues from the log. + """ + from hermes_cli.web_server import _SESSION_TOKEN, app + auth_required = bool(getattr(app.state, "auth_required", False)) + if auth_required: + # Lazy import — keeps this function importable in test harnesses + # that don't bring in the dashboard_auth layer. + from hermes_cli.dashboard_auth.audit import AuditEvent, audit_log + from hermes_cli.dashboard_auth.ws_tickets import ( + TicketInvalid, + consume_internal_credential, + consume_ticket, + ) + + # Server-spawned children (PTY child → /api/ws, /api/pub) present the + # multi-use internal credential rather than a single-use ticket, so + # they survive reconnects and slow cold boots. + internal = ws.query_params.get("internal", "") + if internal: + try: + info = consume_internal_credential(internal) + # Stamp the server-minted identity onto the WS object so the + # connection (and any transport built from it) can never be + # impersonated by RPC params. Internal peers are marked + # ``server-internal`` and are excluded from privileged + # controller registration downstream. + ws._hermes_auth_identity = { + "user_id": info.get("user_id"), + "provider": info.get("provider"), + } + return None, "internal" + except TicketInvalid as exc: + audit_log( + AuditEvent.WS_TICKET_REJECTED, + reason=f"internal: {exc}", + ip=(ws.client.host if ws.client else ""), + path=ws.url.path, + ) + return "internal_invalid", "internal" + + protocol_ticket, protocol_reason = _gateway_ws_ticket_from_subprotocol(ws) + if protocol_reason == "invalid": + return "ticket_invalid", "ticket-subprotocol" + ticket = protocol_ticket or ws.query_params.get("ticket", "") + if not ticket: + return "no_credential", "none" + + try: + info = consume_ticket(ticket) + # The ticket binds a server-minted {user_id, provider}; stamp it + # onto the WS object so ``gateway_ws`` can hand it to the gateway + # transport, where it is the sole identity authority for + # browser-controller registration. A client can never supply or + # spoof this value through RPC params. Only the two identity + # fields are carried — bookkeeping (e.g. ``minted_at``) is not + # part of the identity contract. + ws._hermes_auth_identity = { + "user_id": info.get("user_id"), + "provider": info.get("provider"), + } + if protocol_ticket: + # Select only the stable public protocol during accept. The + # ticket-bearing protocol is a credential and must never be + # reflected back to the browser or retained after admission. + ws._hermes_ws_subprotocol = _GATEWAY_WS_PROTOCOL + return None, "ticket-subprotocol" + return None, "ticket" + except TicketInvalid as exc: + audit_log( + AuditEvent.WS_TICKET_REJECTED, + reason=str(exc), + ip=(ws.client.host if ws.client else ""), + path=ws.url.path, + ) + return "ticket_invalid", "ticket" + + token = ws.query_params.get("token", "") + if not token: + return "no_credential", "none" + if hmac.compare_digest(token.encode(), _SESSION_TOKEN.encode()): + return None, "token" + return "token_mismatch", "token" + + +def _ws_auth_ok(ws: "WebSocket") -> bool: + """True when the WS-upgrade credential is accepted. See _ws_auth_reason.""" + from hermes_cli.web_server import _ws_auth_reason + return _ws_auth_reason(ws)[0] is None + + +# Per-channel subscriber registry used by /api/pub (PTY-side gateway → dashboard) +# and /api/events (dashboard → browser sidebar). Keyed by an opaque channel id +# the chat tab generates on mount; entries auto-evict when the last subscriber +# drops AND the publisher has disconnected. +# (Channel state and the chat-argv lock are initialised in _lifespan on app +# startup — see _get_event_state / _get_chat_argv_lock above.) + + +def _resolve_chat_argv( + resume: Optional[str] = None, + sidecar_url: Optional[str] = None, + profile: Optional[str] = None, + active_session_file: Optional[str] = None, +) -> tuple[list[str], Optional[str], Optional[dict]]: + """Resolve the argv + cwd + env for the chat PTY. + + Default: whatever ``hermes --tui`` would run. Tests monkeypatch this + function to inject a tiny fake command (``cat``, ``sh -c 'printf …'``) + so nothing has to build Node or the TUI bundle. + + Session resume is propagated via the ``HERMES_TUI_RESUME`` env var — + matching what ``hermes_cli.main._launch_tui`` does for the CLI path. + Appending ``--resume `` to argv doesn't work because ``ui-tui`` does + not parse its argv. + + ``HERMES_TUI_GATEWAY_URL`` is injected so the PTY child can attach to + this process's in-memory ``tui_gateway`` instance instead of spawning + its own Python gateway subprocess. + + `sidecar_url` (when set) is forwarded as ``HERMES_TUI_SIDECAR_URL`` so + the spawned ``tui_gateway.entry`` can mirror dispatcher emits to the + dashboard's ``/api/pub`` endpoint (see :func:`pub_ws`). + + `active_session_file` (when set) is forwarded as + ``HERMES_TUI_ACTIVE_SESSION_FILE``. The TUI writes the current session id + there whenever it creates/resumes/switches sessions, giving the dashboard a + small cross-process breadcrumb for reconnecting after an unexpected browser + WebSocket close. + + `profile` (when set) scopes the ENTIRE chat to that profile by pointing + ``HERMES_HOME`` at the profile dir in the child env. Every spawned + process (the TUI and the ``tui_gateway.entry`` it launches) resolves + ``get_hermes_home()`` from that env var at its own import, so the child + binds the profile's config, skills, memory, and state.db from the start + — the same propagation ``hermes -p `` performs. The in-process + ``HERMES_TUI_GATEWAY_URL`` attach is SKIPPED for scoped chats: the + dashboard's in-memory gateway runs under the dashboard's own profile, + so a profile-scoped chat must spawn its own gateway subprocess. + """ + from hermes_cli.web_server import ( + _config_profile_scope, + _open_session_db_for_profile, + _resolve_profile_dir, + _session_latest_descendant, + ) + from hermes_cli.main import PROJECT_ROOT, _apply_tui_python_env, _make_tui_argv + + profile_dir: Optional[Path] = None + requested = (profile or "").strip() + if requested and requested.lower() != "current": + profile_dir = _resolve_profile_dir(requested) + + argv, cwd = _make_tui_argv(PROJECT_ROOT / "ui-tui", tui_dev=False) + # Hermes TUI child: build via the single spawn-env factory (profile-home + # contract applied; secrets kept — the spawned agent needs provider creds). + # An explicit profile scope still overrides HERMES_HOME before config is + # bridged into the child environment. + from tools.environments.local import build_subprocess_env + env = build_subprocess_env(scrub_secrets=False, inherit_profile_home=True) + if profile_dir is not None: + env["HERMES_HOME"] = str(profile_dir) + try: + from hermes_cli.config import ( + apply_terminal_config_to_env, + read_raw_config, + terminal_config_owned_env_vars, + ) + + if profile_dir is not None: + # The dashboard process already bridged its own terminal config + # into os.environ at startup. Remove only keys explicitly owned by + # that launch profile before applying the selected profile. Values + # exported by the operator for keys omitted from the launch profile + # remain valid fallbacks, matching apply_terminal_config_to_env(). + raw_launch_terminal = read_raw_config().get("terminal") + for env_var in terminal_config_owned_env_vars(raw_launch_terminal): + env.pop(env_var, None) + with _config_profile_scope(requested): + apply_terminal_config_to_env(env=env) + else: + apply_terminal_config_to_env(env=env) + except Exception: + _log.warning("Failed to apply terminal config bridge for dashboard chat", exc_info=True) + _apply_tui_python_env(env) + env.setdefault("NODE_ENV", "production") + # Browser-embedded chat should prefer stable wheel-based scrollback over + # native terminal mouse tracking. When mouse tracking is enabled, wheel + # events are consumed by the TUI and forwarded as terminal input, which + # makes browser-side transcript scrolling feel broken. Keep the terminal + # build unchanged for native CLI usage; only disable mouse tracking for + # the dashboard PTY path. + env.setdefault("HERMES_TUI_DISABLE_MOUSE", "1") + env.setdefault("HERMES_TUI_INLINE", "1") + # The dashboard terminal is xterm.js, which always renders 24-bit RGB. + # But chalk inside the TUI child decides its color depth from the + # SERVER process env — and hosted/cloud deploys run the dashboard under + # a process manager (container init, systemd) with no COLORTERM, so + # chalk downgrades every hex color to the xterm 256 palette. The skin's + # bronze border #CD7F32 snaps to palette 173 (#D7875F, salmon-red) and + # the banner reads red/yellow instead of gold. Local launches dodge + # this only because the operator's interactive terminal leaks + # COLORTERM=truecolor into os.environ. Backfill it for the PTY child; + # setdefault so an explicit operator value still wins. + env.setdefault("COLORTERM", "truecolor") + env["HERMES_TUI_DASHBOARD"] = "1" + + if resume: + _resume_db = _open_session_db_for_profile( + requested if profile_dir is not None else None, + read_only=True, + ) + try: + latest_resume, _latest_path = _session_latest_descendant(resume, _resume_db) + finally: + _resume_db.close() + if latest_resume: + resume = latest_resume + env["HERMES_TUI_RESUME"] = resume + + if sidecar_url: + env["HERMES_TUI_SIDECAR_URL"] = sidecar_url + + if active_session_file: + env["HERMES_TUI_ACTIVE_SESSION_FILE"] = active_session_file + + # Profile-scoped chats must NOT attach to the dashboard's in-memory + # gateway — it runs under the dashboard's own profile. Without the + # attach URL, gatewayClient spawns its own `tui_gateway.entry`, which + # inherits the profile HERMES_HOME set above. + if profile_dir is None: + if gateway_ws_url := _build_gateway_ws_url(): + env["HERMES_TUI_GATEWAY_URL"] = gateway_ws_url + + return list(argv), str(cwd) if cwd else None, env + + +# Hosts that mean "listen on every interface" — the server should bind to +# them, but an in-container client must NOT dial them: dialing 0.0.0.0 +# resolves to "any local interface", which on most platforms routes through +# the kernel's wildcard stack and behind a forward proxy (HTTPS_PROXY with +# a NO_PROXY that doesn't list 0.0.0.0) gets MITM'd into a failed handshake +# (issue #58993). The fix is to use a loopback address for the client +# netloc while leaving the bind host alone. +_WILDCARD_HOSTS = frozenset({"0.0.0.0", "::"}) + + +def _resolve_client_ws_host() -> Optional[str]: + """Return the host the in-container WS client should dial. + + Resolution order: + + 1. Explicit ``HERMES_DASHBOARD_WS_HOST`` env var — wins always. Operators + running the dashboard behind a forward proxy can pin a routable host + (e.g. ``127.0.0.1``, the container's internal IP, or a sidecar DNS + name) and bypass auto-detection entirely. + 2. The configured bind host — if it's a wildcard (``0.0.0.0`` / ``::``), + substitute ``127.0.0.1`` since both the dashboard and its TUI child + run in the same container. + 3. Any other bind host (loopback or LAN IP) — preserved verbatim. + """ + from hermes_cli.web_server import app + explicit = os.environ.get("HERMES_DASHBOARD_WS_HOST", "").strip() + if explicit: + return explicit + + host = getattr(app.state, "bound_host", None) + if not host: + return None + + if host in _WILDCARD_HOSTS: + return "127.0.0.1" + + return host + + +def _build_gateway_ws_url() -> Optional[str]: + """ws:// URL the PTY child should attach to for JSON-RPC gateway traffic. + + Loopback / ``--insecure``: ``?token=<_SESSION_TOKEN>``. + + Gated mode: the legacy token path is rejected by ``_ws_auth_ok``, so the + server-spawned PTY child authenticates with the process-lifetime internal + credential (``?internal=``). It must NOT use a single-use browser ticket: + the child reads this URL once at startup and reuses it on every reconnect, + and a 30s-TTL ticket can expire before a slow cold boot even dials. + """ + from hermes_cli.web_server import _SESSION_TOKEN, app + host = _resolve_client_ws_host() + port = getattr(app.state, "bound_port", None) + + if not host or not port: + return None + + netloc = ( + f"[{host}]:{port}" + if ":" in host and not host.startswith("[") + else f"{host}:{port}" + ) + + if getattr(app.state, "auth_required", False): + from hermes_cli.dashboard_auth.ws_tickets import internal_ws_credential + + qs = urllib.parse.urlencode({"internal": internal_ws_credential()}) + else: + qs = urllib.parse.urlencode({"token": _SESSION_TOKEN}) + + return f"ws://{netloc}/api/ws?{qs}" + + +async def _resolve_chat_argv_async( + resume: Optional[str] = None, + sidecar_url: Optional[str] = None, + profile: Optional[str] = None, + active_session_file: Optional[str] = None, +) -> tuple[list[str], Optional[str], Optional[dict]]: + """Resolve chat argv without blocking the dashboard event loop. + + ``_resolve_chat_argv`` may run ``npm install`` / ``npm run build`` through + ``_make_tui_argv``. Keep that synchronous work off the WebSocket event + loop so reverse proxies and existing dashboard connections can continue + to exchange keepalives while the TUI launch command is prepared. The + async lock preserves the previous one-build-at-a-time behavior when + multiple browser tabs connect at once without occupying worker threads + while queued connections wait. + """ + from hermes_cli.web_server import _get_chat_argv_lock, _resolve_chat_argv, app + kwargs = { + "resume": resume, + "sidecar_url": sidecar_url, + "profile": profile, + } + if active_session_file is not None: + kwargs["active_session_file"] = active_session_file + + async with _get_chat_argv_lock(app): + return await asyncio.to_thread( + _resolve_chat_argv, + **kwargs, + ) + + +def _build_sidecar_url(channel: str) -> Optional[str]: + """ws:// URL the PTY child should publish events to, or None when unbound. + + Loopback / ``--insecure``: uses ``?token=<_SESSION_TOKEN>``. + + Gated mode: authenticates with the process-lifetime internal credential + (``?internal=``), the same one ``_build_gateway_ws_url`` uses. The PTY + child is a server-spawned process we trust; the credential is multi-use + and never expires, so the child can reconnect ``/api/pub`` without a new + URL. (This previously minted a single-use 30s ticket, which meant the + child could not reconnect and could miss the window on a slow cold boot.) + Connections authenticated this way are recorded under the + ``server-internal`` identity in the audit log. + """ + from hermes_cli.web_server import _SESSION_TOKEN, app + host = _resolve_client_ws_host() + port = getattr(app.state, "bound_port", None) + + if not host or not port: + return None + + netloc = f"[{host}]:{port}" if ":" in host and not host.startswith("[") else f"{host}:{port}" + + if getattr(app.state, "auth_required", False): + # Gated mode — use the internal credential so the WS upgrade survives + # _ws_auth_ok and the child can reconnect. + from hermes_cli.dashboard_auth.ws_tickets import internal_ws_credential + + qs = urllib.parse.urlencode( + {"internal": internal_ws_credential(), "channel": channel} + ) + else: + qs = urllib.parse.urlencode({"token": _SESSION_TOKEN, "channel": channel}) + + return f"ws://{netloc}/api/pub?{qs}" + + +def _active_session_file_for_channel(app: "FastAPI", channel: str) -> Path: + """Return the per-channel file where a dashboard TUI writes its active sid.""" + from hermes_cli.web_server import _get_pty_active_session_files + files = _get_pty_active_session_files(app) + existing = files.get(channel) + if existing is not None: + return existing + + fd, raw_path = tempfile.mkstemp(prefix="hermes-pty-active-", suffix=".json") + os.close(fd) + path = Path(raw_path) + files[channel] = path + return path + + +# Console commands run in a worker thread. On a timeout, asyncio.wait_for cancels +# the *awaitable*, but Python threads aren't preemptible, so a genuinely stuck +# worker keeps running to completion. To keep that from exhausting the shared +# default thread pool (asyncio.to_thread), we run console commands on a small +# dedicated, bounded pool: a leaked worker is capped, and concurrent console +# execution is bounded to a fixed number of threads regardless of reconnects. +_CONSOLE_EXECUTOR_MAX_WORKERS = 4 +_console_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None +_console_executor_lock = threading.Lock() + + +def _get_console_executor() -> concurrent.futures.ThreadPoolExecutor: + """Lazily create the bounded console worker pool (once per process).""" + global _console_executor + if _console_executor is None: + with _console_executor_lock: + if _console_executor is None: + _console_executor = concurrent.futures.ThreadPoolExecutor( + max_workers=_CONSOLE_EXECUTOR_MAX_WORKERS, + thread_name_prefix="hermes-console", + ) + # Ensure the pool is torn down on interpreter exit. Don't wait on + # in-flight workers: a stuck 60s console command must not block + # shutdown (cancel_futures drops anything not yet started). + atexit.register( + lambda: _console_executor + and _console_executor.shutdown(wait=False, cancel_futures=True) + ) + return _console_executor diff --git a/tests/hermes_cli/test_web_server_pty_import.py b/tests/hermes_cli/test_web_server_pty_import.py index f2fea5326d..ff7db6bc95 100644 --- a/tests/hermes_cli/test_web_server_pty_import.py +++ b/tests/hermes_cli/test_web_server_pty_import.py @@ -48,10 +48,13 @@ def test_web_server_uses_posix_pty_bridge_on_posix(): def test_pty_bridge_import_block_is_platform_branched(): """Source-level guard: a future refactor must not collapse the branch - back to a single POSIX import. Reads web_server.py directly so this - fails the same way on every OS — the runtime symbol checks above can - pass even when the branch shape is wrong on the current platform.""" - src = pytest.importorskip("inspect").getsource(web_server) + back to a single POSIX import. Reads the module that owns the import + block (``web_server_chat``) directly so this fails the same way on every + OS — the runtime symbol checks above can pass even when the branch shape + is wrong on the current platform.""" + from hermes_cli import web_server_chat + + src = pytest.importorskip("inspect").getsource(web_server_chat) # The shape we expect (from PR #39913): # # if sys.platform.startswith("win"):