From 4493bba90100455cfbd90591979de54de4012556 Mon Sep 17 00:00:00 2001 From: Shannon Sands Date: Tue, 30 Jun 2026 16:05:52 +1000 Subject: [PATCH] Add dashboard Hermes console websocket --- hermes_cli/dashboard_auth/routes.py | 3 +- hermes_cli/web_server.py | 542 ++++++++++++++++++ .../hermes_cli/test_dashboard_auth_ws_auth.py | 15 +- .../hermes_cli/test_web_server_console_ws.py | 134 +++++ 4 files changed, 686 insertions(+), 8 deletions(-) create mode 100644 tests/hermes_cli/test_web_server_console_ws.py diff --git a/hermes_cli/dashboard_auth/routes.py b/hermes_cli/dashboard_auth/routes.py index 568a11957be..9e80f2583c5 100644 --- a/hermes_cli/dashboard_auth/routes.py +++ b/hermes_cli/dashboard_auth/routes.py @@ -608,7 +608,8 @@ async def api_auth_ws_ticket(request: Request): Browsers cannot set ``Authorization`` on a WebSocket upgrade, so in gated mode the SPA POSTs this endpoint to get a ``?ticket=`` value to - append to ``/api/pty``, ``/api/ws``, ``/api/pub``, or ``/api/events``. + append to ``/api/pty``, ``/api/console``, ``/api/ws``, ``/api/pub``, or + ``/api/events``. The ticket has a 30-second TTL and is single-use. Calling this endpoint multiple times in quick succession (e.g. one ticket per WS) is the diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index f81353f6aa8..6c19c1868a9 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -12628,6 +12628,548 @@ def _ws_close_reason(text: str) -> str: return encoded[:120].decode("utf-8", "ignore") + "..." +# --------------------------------------------------------------------------- +# /api/console — safe Hermes Console command WebSocket. +# +# Unlike /api/pty, this endpoint never spawns a PTY, shell, or full Hermes CLI +# subprocess. It runs the curated console engine in-process and exchanges +# structured JSON frames with the dashboard xterm overlay. +# --------------------------------------------------------------------------- + +_CONSOLE_PROMPT = "hermes> " +_CONSOLE_COMMAND_TIMEOUT_SECONDS = 60.0 +_CONSOLE_OUTPUT_LIMIT = 50000 + + +def _dashboard_console_context() -> str: + """Choose local vs hosted command policy for the dashboard console.""" + return "hosted" if _default_hermes_root_is_opt_data() else "local" + + +def _console_profile_from_ws(ws: WebSocket) -> Optional[str]: + profile = (ws.query_params.get("profile") or "").strip() + return profile or None + + +def _execute_console_line( + engine: Any, + line: str, + *, + confirmed: bool, + profile: Optional[str], +) -> Any: + # _profile_scope swaps process-global skill module paths; keep it inside + # the worker thread and never hold it across awaits. + with _profile_scope(profile): + return engine.execute(line, confirmed=confirmed) + + +async def _console_send( + ws: WebSocket, + send_lock: asyncio.Lock, + payload: Dict[str, Any], +) -> None: + async with send_lock: + await ws.send_json(payload) + + +async def _console_send_result( + ws: WebSocket, + send_lock: asyncio.Lock, + result: Any, + *, + command_id: int, +) -> None: + command = result.command or "" + status = result.status + if status == "ok": + if result.output: + await _console_send( + ws, + send_lock, + { + "type": "output", + "id": command_id, + "stream": "stdout", + "data": result.output, + "command": command, + }, + ) + await _console_send( + ws, + send_lock, + { + "type": "complete", + "id": command_id, + "status": "ok", + "command": command, + "prompt": _CONSOLE_PROMPT, + }, + ) + return + + if status == "error": + await _console_send( + ws, + send_lock, + { + "type": "error", + "id": command_id, + "message": result.output or "Command failed.", + "command": command, + }, + ) + await _console_send( + ws, + send_lock, + { + "type": "complete", + "id": command_id, + "status": "error", + "command": command, + "prompt": _CONSOLE_PROMPT, + }, + ) + return + + if status == "confirm_required": + await _console_send( + ws, + send_lock, + { + "type": "confirm_required", + "id": command_id, + "command": command, + "message": result.confirmation_message or f"Run `{command}`?", + "prompt": _CONSOLE_PROMPT, + }, + ) + await _console_send( + ws, + send_lock, + { + "type": "complete", + "id": command_id, + "status": "confirm_required", + "command": command, + "prompt": _CONSOLE_PROMPT, + }, + ) + return + + if status == "clear": + await _console_send(ws, send_lock, {"type": "clear", "id": command_id}) + await _console_send( + ws, + send_lock, + { + "type": "complete", + "id": command_id, + "status": "clear", + "command": command, + "prompt": _CONSOLE_PROMPT, + }, + ) + return + + if status == "exit": + await _console_send( + ws, + send_lock, + { + "type": "complete", + "id": command_id, + "status": "exit", + "command": command, + "prompt": "", + }, + ) + return + + await _console_send( + ws, + send_lock, + { + "type": "error", + "id": command_id, + "message": f"Unknown console result status: {status}", + "command": command, + }, + ) + + +def _console_json_payload(msg: Any) -> tuple[Optional[dict[str, Any]], Optional[str]]: + raw: str | bytes | None = msg.get("text") + if raw is None: + raw = msg.get("bytes") + if raw is None: + return None, None + if isinstance(raw, bytes): + try: + raw = raw.decode("utf-8") + except UnicodeDecodeError: + return None, "Console frames must be UTF-8 JSON." + try: + payload = json.loads(raw) + except json.JSONDecodeError: + return None, "Console frames must be JSON objects." + if not isinstance(payload, dict): + return None, "Console frames must be JSON objects." + return payload, None + + +@app.websocket("/api/console") +async def console_ws(ws: WebSocket) -> None: + peer = ws.client.host if ws.client else "?" + + if not _DASHBOARD_EMBEDDED_CHAT_ENABLED: + _log.info("console refused: embedded chat disabled peer=%s", peer) + await ws.close(code=4404, reason="embedded chat disabled") + return + + auth_reason, cred = _ws_auth_reason(ws) + mode = _ws_auth_mode() + if auth_reason is not None: + _log.warning( + "console auth rejected reason=%s mode=%s cred=%s peer=%s", + auth_reason, mode, cred, peer, + ) + await ws.close(code=4401, reason=_ws_close_reason(f"auth: {auth_reason}")) + return + + host_origin_reason = _ws_host_origin_reason(ws) + if host_origin_reason is not None: + _log.warning("console refused: %s peer=%s", host_origin_reason, peer) + await ws.close(code=4403, reason=_ws_close_reason(host_origin_reason)) + return + + client_reason = _ws_client_reason(ws) + if client_reason is not None: + _log.warning("console refused: %s", client_reason) + await ws.close(code=4408, reason=_ws_close_reason(client_reason)) + return + + await ws.accept() + + profile = _console_profile_from_ws(ws) + context = _dashboard_console_context() + send_lock = asyncio.Lock() + + try: + from hermes_cli.console_engine import HermesConsoleEngine + + engine = HermesConsoleEngine( + output_limit=_CONSOLE_OUTPUT_LIMIT, + context=context, # type: ignore[arg-type] + ) + if profile and profile.lower() != "current": + _resolve_profile_dir(profile) + except HTTPException as exc: + await _console_send( + ws, + send_lock, + { + "type": "error", + "message": str(exc.detail), + "prompt": "", + }, + ) + await ws.close(code=4400, reason=_ws_close_reason(str(exc.detail))) + return + except Exception as exc: + _log.exception("console failed to initialize") + await _console_send( + ws, + send_lock, + { + "type": "error", + "message": f"Console unavailable: {exc}", + "prompt": "", + }, + ) + await ws.close(code=1011) + return + + _log.info( + "console accepted peer=%s mode=%s cred=%s context=%s profile=%s", + peer, + mode, + cred, + context, + profile or "current", + ) + await _console_send( + ws, + send_lock, + { + "type": "ready", + "context": context, + "profile": profile or "current", + "prompt": _CONSOLE_PROMPT, + }, + ) + + active_task: asyncio.Task | None = None + pending_confirmation: Optional[str] = None + command_generation = 0 + + async def run_command(line: str, *, confirmed: bool, command_id: int) -> None: + nonlocal active_task, pending_confirmation, command_generation + try: + result = await asyncio.wait_for( + asyncio.to_thread( + _execute_console_line, + engine, + line, + confirmed=confirmed, + profile=profile, + ), + timeout=_CONSOLE_COMMAND_TIMEOUT_SECONDS, + ) + except asyncio.CancelledError: + raise + except asyncio.TimeoutError: + if command_id == command_generation: + pending_confirmation = None + await _console_send( + ws, + send_lock, + { + "type": "error", + "id": command_id, + "message": ( + "Command timed out. Hermes Console returned to the prompt." + ), + "command": line, + }, + ) + await _console_send( + ws, + send_lock, + { + "type": "complete", + "id": command_id, + "status": "timeout", + "command": line, + "prompt": _CONSOLE_PROMPT, + }, + ) + except Exception as exc: + if command_id == command_generation: + pending_confirmation = None + _log.exception("console command failed") + await _console_send( + ws, + send_lock, + { + "type": "error", + "id": command_id, + "message": str(exc) or exc.__class__.__name__, + "command": line, + }, + ) + await _console_send( + ws, + send_lock, + { + "type": "complete", + "id": command_id, + "status": "error", + "command": line, + "prompt": _CONSOLE_PROMPT, + }, + ) + else: + if command_id != command_generation: + return + pending_confirmation = ( + result.command if result.status == "confirm_required" else None + ) + await _console_send_result( + ws, + send_lock, + result, + command_id=command_id, + ) + if result.status == "exit": + await ws.close(code=1000) + finally: + if command_id == command_generation: + active_task = None + + async def start_command(line: str, *, confirmed: bool = False) -> None: + nonlocal active_task, command_generation + command_generation += 1 + command_id = command_generation + active_task = asyncio.create_task( + run_command(line, confirmed=confirmed, command_id=command_id) + ) + + try: + while True: + try: + msg = await ws.receive() + except RuntimeError: + break + msg_type = msg.get("type") + if msg_type == "websocket.disconnect": + break + + payload, error = _console_json_payload(msg) + if error: + await _console_send( + ws, + send_lock, + { + "type": "error", + "message": error, + "prompt": _CONSOLE_PROMPT, + }, + ) + continue + if payload is None: + continue + + frame_type = str(payload.get("type") or "").strip().lower() + if frame_type == "ping": + await _console_send( + ws, + send_lock, + { + "type": "pong", + "prompt": _CONSOLE_PROMPT, + }, + ) + continue + + if frame_type == "cancel": + if active_task and not active_task.done(): + command_generation += 1 + active_task.cancel() + active_task = None + pending_confirmation = None + await _console_send( + ws, + send_lock, + { + "type": "complete", + "status": "cancelled", + "prompt": _CONSOLE_PROMPT, + }, + ) + elif pending_confirmation: + pending_confirmation = None + await _console_send( + ws, + send_lock, + { + "type": "complete", + "status": "cancelled", + "prompt": _CONSOLE_PROMPT, + }, + ) + else: + await _console_send( + ws, + send_lock, + { + "type": "complete", + "status": "idle", + "prompt": _CONSOLE_PROMPT, + }, + ) + continue + + if active_task and not active_task.done(): + await _console_send( + ws, + send_lock, + { + "type": "error", + "message": "A console command is already running.", + "prompt": _CONSOLE_PROMPT, + }, + ) + continue + + if frame_type == "confirm": + command = str(payload.get("command") or pending_confirmation or "").strip() + if not pending_confirmation: + await _console_send( + ws, + send_lock, + { + "type": "error", + "message": "No command is waiting for confirmation.", + "prompt": _CONSOLE_PROMPT, + }, + ) + continue + if command != pending_confirmation: + await _console_send( + ws, + send_lock, + { + "type": "error", + "message": "Confirmation does not match the pending command.", + "prompt": _CONSOLE_PROMPT, + }, + ) + continue + pending_confirmation = None + await start_command(command, confirmed=True) + continue + + if frame_type in {"input", "command"}: + line = str(payload.get("line") or payload.get("command") or "").strip() + if not line: + await _console_send( + ws, + send_lock, + { + "type": "complete", + "status": "ok", + "prompt": _CONSOLE_PROMPT, + }, + ) + continue + if pending_confirmation: + await _console_send( + ws, + send_lock, + { + "type": "error", + "message": ( + "Confirm or cancel the pending command before " + "running another one." + ), + "prompt": _CONSOLE_PROMPT, + }, + ) + continue + await start_command(line) + continue + + await _console_send( + ws, + send_lock, + { + "type": "error", + "message": f"Unsupported console frame: {frame_type or '?'}", + "prompt": _CONSOLE_PROMPT, + }, + ) + except WebSocketDisconnect: + pass + finally: + if active_task and not active_task.done(): + active_task.cancel() + try: + await active_task + except (asyncio.CancelledError, Exception): + pass + + @app.websocket("/api/pty") async def pty_ws(ws: WebSocket) -> None: peer = ws.client.host if ws.client else "?" diff --git a/tests/hermes_cli/test_dashboard_auth_ws_auth.py b/tests/hermes_cli/test_dashboard_auth_ws_auth.py index c10d8839fac..2d28bcf1dcf 100644 --- a/tests/hermes_cli/test_dashboard_auth_ws_auth.py +++ b/tests/hermes_cli/test_dashboard_auth_ws_auth.py @@ -1,9 +1,9 @@ """Tests for the WS-upgrade auth helper (Phase 5 task 5.2). -The dashboard's four WS endpoints (``/api/pty``, ``/api/ws``, ``/api/pub``, -``/api/events``) share an auth gate: ``_ws_auth_ok``. In loopback mode it -accepts ``?token=<_SESSION_TOKEN>``; in gated mode it accepts a single-use -``?ticket=`` minted by ``POST /api/auth/ws-ticket``. +The dashboard's WS endpoints (``/api/pty``, ``/api/console``, ``/api/ws``, +``/api/pub``, ``/api/events``) share an auth gate: ``_ws_auth_ok``. In +loopback mode it accepts ``?token=<_SESSION_TOKEN>``; in gated mode it accepts +a single-use ``?ticket=`` minted by ``POST /api/auth/ws-ticket``. These tests exercise the helper at the unit level (no actual WS upgrade) plus the ticket-mint endpoint under realistic gated-mode setup. We don't @@ -315,9 +315,10 @@ class TestWsRequestIsAllowedGated: (intended only for unauthenticated loopback dev) must not also reject those upgrades: the OAuth gate + single-use ticket is the auth. - Regression coverage: every WS endpoint (``/api/pty``, ``/api/ws``, - ``/api/pub``, ``/api/events``) calls ``_ws_request_is_allowed`` after - ``_ws_auth_ok``. If the peer-IP check rejects gated mode, the chat + Regression coverage: every WS endpoint (``/api/pty``, ``/api/console``, + ``/api/ws``, ``/api/pub``, ``/api/events``) calls + ``_ws_request_is_allowed`` after ``_ws_auth_ok``. If the peer-IP check + rejects gated mode, the chat tab + sidebar tool feed silently fail to connect even after a successful OAuth login. """ diff --git a/tests/hermes_cli/test_web_server_console_ws.py b/tests/hermes_cli/test_web_server_console_ws.py new file mode 100644 index 00000000000..538251ec7be --- /dev/null +++ b/tests/hermes_cli/test_web_server_console_ws.py @@ -0,0 +1,134 @@ +"""Dashboard Hermes Console websocket tests.""" + +from __future__ import annotations + +import time +from urllib.parse import urlencode + +import pytest +from starlette.testclient import TestClient +from starlette.websockets import WebSocketDisconnect + +from hermes_cli import web_server + + +@pytest.fixture +def console_client(monkeypatch, _isolate_hermes_home): + previous_auth_required = getattr(web_server.app.state, "auth_required", None) + previous_bound_host = getattr(web_server.app.state, "bound_host", None) + web_server.app.state.auth_required = False + web_server.app.state.bound_host = None + monkeypatch.setattr(web_server, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True) + + client = TestClient(web_server.app) + try: + yield client + finally: + close = getattr(client, "close", None) + if close is not None: + close() + if previous_auth_required is None: + if hasattr(web_server.app.state, "auth_required"): + delattr(web_server.app.state, "auth_required") + else: + web_server.app.state.auth_required = previous_auth_required + if previous_bound_host is None: + if hasattr(web_server.app.state, "bound_host"): + delattr(web_server.app.state, "bound_host") + else: + web_server.app.state.bound_host = previous_bound_host + + +def _url(token: str | None = None, **params: str) -> str: + query = {"token": web_server._SESSION_TOKEN, **params} + if token is not None: + query["token"] = token + return f"/api/console?{urlencode(query)}" + + +def _recv_until(conn, frame_type: str, *, status: str | None = None) -> dict: + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline: + frame = conn.receive_json() + if frame.get("type") != frame_type: + continue + if status is not None and frame.get("status") != status: + continue + return frame + raise AssertionError(f"Timed out waiting for {frame_type} frame") + + +def test_console_ws_rejects_missing_or_bad_token(console_client): + with pytest.raises(WebSocketDisconnect) as exc: + with console_client.websocket_connect("/api/console"): + pass + assert exc.value.code == 4401 + + with pytest.raises(WebSocketDisconnect) as exc: + with console_client.websocket_connect(_url(token="wrong")): + pass + assert exc.value.code == 4401 + + +def test_console_ws_runs_read_only_command(console_client): + with console_client.websocket_connect(_url()) as conn: + ready = conn.receive_json() + assert ready["type"] == "ready" + assert ready["context"] == "local" + assert ready["prompt"] == "hermes> " + + conn.send_json({"type": "input", "line": "help"}) + + output = _recv_until(conn, "output") + assert "Hermes Console" in output["data"] + complete = _recv_until(conn, "complete", status="ok") + assert complete["prompt"] == "hermes> " + + +def test_console_ws_confirmed_command_executes_after_confirmation(console_client): + from hermes_cli.config import load_config + + with console_client.websocket_connect(_url()) as conn: + assert conn.receive_json()["type"] == "ready" + conn.send_json({"type": "input", "line": "config set display.interface cli"}) + + confirmation = _recv_until(conn, "confirm_required") + assert confirmation["command"] == "config set display.interface cli" + assert confirmation["message"] + + conn.send_json({"type": "confirm", "command": confirmation["command"]}) + _recv_until(conn, "complete", status="ok") + + assert load_config()["display"]["interface"] == "cli" + + +def test_console_ws_uses_hosted_context_for_opt_data_policy(console_client, monkeypatch): + monkeypatch.setattr(web_server, "_default_hermes_root_is_opt_data", lambda: True) + + with console_client.websocket_connect(_url()) as conn: + ready = conn.receive_json() + assert ready["type"] == "ready" + assert ready["context"] == "hosted" + + conn.send_json({"type": "input", "line": "profile create nope"}) + + error = _recv_until(conn, "error") + assert "hosted Hermes Console" in error["message"] + + +def test_console_ws_cancel_returns_to_prompt(console_client, monkeypatch): + from hermes_cli.console_engine import ConsoleResult, HermesConsoleEngine + + def slow_execute(self, line: str, *, confirmed: bool = False): + time.sleep(0.5) + return ConsoleResult("ok", output="late", command=line) + + monkeypatch.setattr(HermesConsoleEngine, "execute", slow_execute) + + with console_client.websocket_connect(_url()) as conn: + assert conn.receive_json()["type"] == "ready" + conn.send_json({"type": "input", "line": "status"}) + conn.send_json({"type": "cancel"}) + + complete = _recv_until(conn, "complete", status="cancelled") + assert complete["prompt"] == "hermes> "