ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

refactor(web_server): extract chat/PTY/WS-auth cluster to web_server_chat

Browse Source
This commit is contained in:
Teknium
2026-09-02 15:36:53 -07:00
parent 2497ec5126
commit 5608fe7cdf
3 changed files with 812 additions and 744 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/web_server.py
+27 -740
View File
@@ -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=<session_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=<token>`` 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.<name>
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=<single-use>`` — 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=<process-credential>`` — 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 <id>`` 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 <name>`` 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.<name>
_BUILTIN_DASHBOARD_THEMES,
_THEME_COMPONENT_BUCKETS,
hermes_cli/web_server_chat.py
+778
View File
@@ -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.<name>`` 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=<session_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=<single-use>`` — 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=<process-credential>`` — 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 <id>`` 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 <name>`` 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
tests/hermes_cli/test_web_server_pty_import.py
+7 -4
View File
@@ -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"):