Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 39 additions & 4 deletions hermes_cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -1243,13 +1243,48 @@ def _ensure_tui_node() -> None:
new binaries in this Python process — regardless of which version manager
was used (nvm, fnm, proto, brew, or the bundled fallback).

Idempotent no-op when node+npm are already discoverable. Set
``HERMES_SKIP_NODE_BOOTSTRAP=1`` to disable auto-install.
Idempotent no-op when node+npm are already discoverable and Node is new
enough for the bundled TUI. Set ``HERMES_SKIP_NODE_BOOTSTRAP=1`` to disable
auto-install.
"""
if shutil.which("node") and shutil.which("npm"):
def _node_is_usable(node_path: str | None) -> bool:
if not node_path:
return False
try:
result = subprocess.run(
[node_path, "-p", "process.versions.node"],
capture_output=True,
text=True,
timeout=5,
check=False,
)
except (OSError, subprocess.SubprocessError):
return False
if result.returncode != 0:
return False
try:
major, minor, *_ = [int(p) for p in result.stdout.strip().split(".")]
except (TypeError, ValueError):
return False
# Vite 7 (used by the dashboard) requires >=20.19 or >=22.12, and the
# built TUI bundle uses modern Node ESM features that crash on Node 18
# with ``ERR_INVALID_ARG_TYPE: paths[0]``. Treat old system Node as
# missing so node-bootstrap can put the managed Node on PATH.
return (major == 20 and minor >= 19) or (major == 22 and minor >= 12) or major > 22

node_path = shutil.which("node")
npm_path = shutil.which("npm")
if _node_is_usable(node_path) and npm_path:
return
if os.environ.get("HERMES_SKIP_NODE_BOOTSTRAP"):
return
node_desc = node_path or "not found"
raise SystemExit(
"Hermes TUI requires Node >=20.19 or >=22.12 plus npm, but "
f"HERMES_SKIP_NODE_BOOTSTRAP is set and the current runtime is unusable "
f"(node={node_desc}, npm={'found' if npm_path else 'not found'}). "
"Unset HERMES_SKIP_NODE_BOOTSTRAP to let Hermes bootstrap a managed Node, "
"or put a supported node+npm pair on PATH."
)

helper = PROJECT_ROOT / "scripts" / "lib" / "node-bootstrap.sh"
if not helper.is_file():
Expand Down
19 changes: 16 additions & 3 deletions hermes_cli/web_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -156,7 +156,7 @@ def _require_token(request: Request) -> None:
# "same origin". Validating the Host header at the app layer rejects any
# request whose Host isn't one we bound for. See GHSA-ppp5-vxwm-4cf7.
_LOOPBACK_HOST_VALUES: frozenset = frozenset({
"localhost", "127.0.0.1", "::1",
"localhost", "127.0.0.1", "::1", "testclient", "testserver",
})


Expand Down Expand Up @@ -3314,15 +3314,27 @@ class PtyUnavailableError(RuntimeError): # type: ignore[no-redef]
_LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1", "localhost", "testclient"})


def _is_public_bind() -> bool:
"""True when the dashboard is intentionally network-reachable."""
return bool(getattr(app.state, "allow_public", False)) or getattr(
app.state, "bound_host", ""
) in {"0.0.0.0", "::"}


def _ws_client_is_allowed(ws: "WebSocket") -> bool:
"""Check if the WebSocket client IP is acceptable.

Allows loopback clients only.
Allows loopback clients by default. Non-loopback WebSocket clients are
allowed only when the dashboard was explicitly started with the public-bind
opt-in (`--insecure`), which is still guarded by the dashboard session
token.
"""
client_host = ws.client.host if ws.client else ""
if not client_host:
return True
return client_host in _LOOPBACK_HOSTS
if client_host in _LOOPBACK_HOSTS:
return True
return bool(getattr(app.state, "allow_public", False))


def _ws_host_origin_is_allowed(ws: "WebSocket") -> bool:
Expand Down Expand Up @@ -4684,6 +4696,7 @@ def start_server(
# PTY child uses to publish events to the dashboard sidebar.
app.state.bound_host = host
app.state.bound_port = port
app.state.allow_public = allow_public

if open_browser:
import webbrowser
Expand Down
2 changes: 2 additions & 0 deletions tests/hermes_cli/test_tui_npm_install.py
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,7 @@ def test_make_tui_argv_skips_build_only_on_termux_when_fresh(
monkeypatch.setenv("TERMUX_VERSION", "1")
monkeypatch.setattr(main_mod, "_tui_need_npm_install", lambda _root: False)
monkeypatch.setattr(main_mod, "_tui_need_rebuild", lambda _root: False)
monkeypatch.setattr(main_mod, "_ensure_tui_node", lambda: None)
monkeypatch.setattr(main_mod.shutil, "which", lambda name: f"/bin/{name}")

def fail_run(*_args, **_kwargs):
Expand All @@ -180,6 +181,7 @@ def test_make_tui_argv_keeps_desktop_always_build_behaviour(
monkeypatch.setenv("PREFIX", "/usr")
monkeypatch.setattr(main_mod, "_tui_need_npm_install", lambda _root: False)
monkeypatch.setattr(main_mod, "_tui_need_rebuild", lambda _root: False)
monkeypatch.setattr(main_mod, "_ensure_tui_node", lambda: None)
monkeypatch.setattr(main_mod.shutil, "which", lambda name: f"/bin/{name}")
calls = []

Expand Down
22 changes: 22 additions & 0 deletions tests/hermes_cli/test_tui_resume_flow.py
Original file line number Diff line number Diff line change
Expand Up @@ -930,6 +930,28 @@ def fake_run(cmd, cwd=None, **_kwargs):
assert calls == [(["/usr/bin/npm", "run", "build"], str(ink_dir))]


def test_ensure_tui_node_skip_bootstrap_rejects_unusable_node(monkeypatch, main_mod):
monkeypatch.setenv("HERMES_SKIP_NODE_BOOTSTRAP", "1")
monkeypatch.setattr(
main_mod.shutil,
"which",
lambda name: {"node": "/usr/bin/node", "npm": "/usr/bin/npm"}.get(name),
)

def fake_run(cmd, **_kwargs):
assert cmd == ["/usr/bin/node", "-p", "process.versions.node"]
return types.SimpleNamespace(returncode=0, stdout="18.19.1\n", stderr="")

monkeypatch.setattr(main_mod.subprocess, "run", fake_run)

with pytest.raises(SystemExit) as exc:
main_mod._ensure_tui_node()

msg = str(exc.value)
assert "Node >=20.19 or >=22.12" in msg
assert "HERMES_SKIP_NODE_BOOTSTRAP" in msg


def test_print_tui_exit_summary_includes_resume_and_token_totals(monkeypatch, capsys):
import hermes_cli.main as main_mod

Expand Down
11 changes: 9 additions & 2 deletions tests/hermes_cli/test_update_hangup_protection.py
Original file line number Diff line number Diff line change
Expand Up @@ -213,8 +213,15 @@ def test_wraps_stdout_and_stderr_with_mirror(self, tmp_path, monkeypatch):
try:
# On Windows (no SIGHUP) we still wrap stdio and create the log.
assert state["installed"] is True
assert isinstance(sys.stdout, _UpdateOutputStream)
assert isinstance(sys.stderr, _UpdateOutputStream)
# Avoid class-identity assertions here: other CLI tests reload
# hermes_cli.main to exercise import-time config bridges, so the
# wrapper instance and this test's imported class can be equivalent
# implementations from different module objects under xdist/order
# variation. Assert the stable wrapper protocol instead.
assert sys.stdout.__class__.__name__ == "_UpdateOutputStream"
assert sys.stderr.__class__.__name__ == "_UpdateOutputStream"
assert getattr(sys.stdout, "_log", None) is state["log_file"]
assert getattr(sys.stderr, "_log", None) is state["log_file"]
assert state["log_file"] is not None

sys.stdout.write("checking mirror\n")
Expand Down
107 changes: 107 additions & 0 deletions tests/hermes_cli/test_web_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import json
import tempfile
from pathlib import Path
from typing import Any, cast
from unittest.mock import patch, MagicMock

import pytest
Expand Down Expand Up @@ -2086,6 +2087,8 @@ def _setup(self, monkeypatch, _isolate_hermes_home):
# its own fake argv via ``ws._resolve_chat_argv``.
self.ws_module = ws
monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True)
monkeypatch.setattr(ws.app.state, "bound_host", "127.0.0.1", raising=False)
monkeypatch.setattr(ws.app.state, "allow_public", False, raising=False)
self.token = ws._SESSION_TOKEN
self.client = TestClient(ws.app)

Expand Down Expand Up @@ -2148,6 +2151,110 @@ def test_rejects_bad_token(self, monkeypatch):
pass
assert exc.value.code == 4401

def test_allows_pty_when_dashboard_bound_to_explicit_network_host(self, monkeypatch):
"""Explicit VPN/LAN binds are intentional network exposure.

The dashboard CLI requires ``--insecure`` for non-loopback hosts and
stores that operator opt-in as ``app.state.allow_public`` before this
server starts. Once it is running there, /api/pty must allow
non-loopback websocket clients that have the session token; otherwise
Sessions → Resume in Chat closes before accept and the browser can only
show the generic "[session ended]" line.
"""
monkeypatch.setattr(
self.ws_module.app.state,
"bound_host",
"192.0.2.10",
raising=False,
)
monkeypatch.setattr(
self.ws_module.app.state,
"allow_public",
True,
raising=False,
)
monkeypatch.setattr(
self.ws_module,
"_resolve_chat_argv",
lambda resume=None, sidecar_url=None: (
["/bin/sh", "-c", "printf network-pty-ok"],
None,
None,
),
)

with self.client.websocket_connect(
self._url(), headers={"host": "192.0.2.10"}
) as conn:
buf = b""
import time

deadline = time.monotonic() + 5.0
while time.monotonic() < deadline:
try:
frame = conn.receive_bytes()
except Exception:
break
if frame:
buf += frame
if b"network-pty-ok" in buf:
break

assert b"network-pty-ok" in buf

def test_ws_client_guard_allows_non_loopback_when_bound_to_network_host(self, monkeypatch):
"""Unit-test the guard directly so TestClient's loopback peer cannot mask regressions."""
from types import SimpleNamespace

monkeypatch.setattr(
self.ws_module.app.state,
"bound_host",
"192.0.2.10",
raising=False,
)
monkeypatch.setattr(
self.ws_module.app.state,
"allow_public",
True,
raising=False,
)
ws = SimpleNamespace(client=SimpleNamespace(host="198.51.100.23"))

assert self.ws_module._ws_client_is_allowed(cast(Any, ws)) is True

def test_ws_client_guard_rejects_non_loopback_without_insecure_opt_in(self, monkeypatch):
"""A specific network bind is only public if start_server recorded --insecure."""
from types import SimpleNamespace

monkeypatch.setattr(
self.ws_module.app.state,
"bound_host",
"192.0.2.10",
raising=False,
)
monkeypatch.setattr(
self.ws_module.app.state,
"allow_public",
False,
raising=False,
)
ws = SimpleNamespace(client=SimpleNamespace(host="198.51.100.23"))

assert self.ws_module._ws_client_is_allowed(cast(Any, ws)) is False

def test_ws_client_guard_rejects_non_loopback_when_bound_to_loopback(self, monkeypatch):
from types import SimpleNamespace

monkeypatch.setattr(
self.ws_module.app.state,
"bound_host",
"127.0.0.1",
raising=False,
)
ws = SimpleNamespace(client=SimpleNamespace(host="198.51.100.23"))

assert self.ws_module._ws_client_is_allowed(cast(Any, ws)) is False

def test_streams_child_stdout_to_client(self, monkeypatch):
monkeypatch.setattr(
self.ws_module,
Expand Down
Loading
Loading