diff --git a/hermes_cli/web_routers/chat_ws.py b/hermes_cli/web_routers/chat_ws.py index 8d38a1a591..81825bcced 100644 --- a/hermes_cli/web_routers/chat_ws.py +++ b/hermes_cli/web_routers/chat_ws.py @@ -1,21 +1,22 @@ -"""Chat-tab WebSocket routes: /api/console, /api/pty, the /api/ws gateway sidecar and /api/pub + /api/events broadcast. +"""Chat-tab WebSocket routes: /api/console, /api/pty, the /api/ws gateway +sidecar and /api/pub + /api/events broadcast. -Extracted from ``hermes_cli.web_server``; helpers/state that tests monkeypatch on -``web_server`` stay there and are imported lazily at call time (cycle-safe). +Helpers/state that tests monkeypatch on ``web_server`` stay there and are +reached through the late-binding seam (cycle-safe). """ import asyncio import functools -import logging import json -from fastapi import APIRouter -from hermes_cli.web_deps import late -from fastapi import HTTPException, WebSocket, WebSocketDisconnect -from hermes_cli.pty_session import RegistryFull +import logging +import re from pathlib import Path from typing import Any, Dict, Optional -import re -from fastapi import FastAPI + +from fastapi import APIRouter, FastAPI, HTTPException, WebSocket, WebSocketDisconnect + +from hermes_cli.pty_session import RegistryFull +from hermes_cli.web_deps import LateState, late _log = logging.getLogger("hermes_cli.web_server") router = APIRouter() @@ -33,17 +34,13 @@ _ws_auth_reason = late("_ws_auth_reason") _ws_client_reason = late("_ws_client_reason") _ws_host_origin_reason = late("_ws_host_origin_reason") _ws_request_is_allowed = late("_ws_request_is_allowed") +_DASHBOARD_EMBEDDED_CHAT_ENABLED = LateState("_DASHBOARD_EMBEDDED_CHAT_ENABLED") def _get_event_state(app: "FastAPI"): - """Return (event_channels, event_lock) from app.state. - - Lazily initialises the state if the lifespan hasn't run (e.g. when - TestClient is constructed without a ``with`` block). The lifespan - path is preferred because it guarantees the Lock is created on the - correct event loop, but the lazy path lets existing non-``with`` - TestClient usages keep working. - """ + """(event_channels, event_lock) from app.state, lazily initialised when the + lifespan hasn't run (TestClient without a ``with`` block). The lifespan path + is preferred because it creates the Lock on the correct event loop.""" try: return app.state.event_channels, app.state.event_lock except AttributeError: @@ -71,20 +68,17 @@ async def _broadcast_event(app: Any, channel: str, payload: str) -> None: event_channels, event_lock = _get_event_state(app) async with event_lock: subs = list(event_channels.get(channel, ())) - for sub in subs: try: await sub.send_text(payload) except Exception: - # Subscriber went away mid-send; the /api/events finally clause - # will remove it from the registry on its next iteration. + # Subscriber went away mid-send; /api/events' finally removes it. _log.warning("broadcast send failed for subscriber on %s", channel, exc_info=True) def _channel_or_close_code(ws: WebSocket) -> Optional[str]: - """Return the channel id from the query string or None if invalid.""" + """Channel id from the query string, or None if invalid.""" channel = ws.query_params.get("channel", "") - return channel if _VALID_CHANNEL_RE.match(channel) else None @@ -93,198 +87,142 @@ def _read_active_session_file(path: Path) -> Optional[str]: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None - - session_id = str(data.get("session_id") or "").strip() - return session_id or None - - -def _forget_active_session_file(path: Path) -> None: - try: - path.unlink(missing_ok=True) - except OSError: - pass + return str(data.get("session_id") or "").strip() or None def _ws_close_reason(text: str) -> str: - """Clamp a WS close reason to the protocol's 123-byte UTF-8 limit. - - RFC 6455 caps the close-frame reason at 123 bytes; uvicorn raises if a - longer string is passed. Our reasons embed an attacker-controlled origin, - so truncate defensively rather than crash the close handler. - """ + """Clamp to RFC 6455's 123-byte close-reason limit (uvicorn raises past it); + reasons embed an attacker-controlled origin, so truncate rather than crash.""" encoded = text.encode("utf-8", "replace") if len(encoded) <= 123: return text 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. -# --------------------------------------------------------------------------- +async def _ws_gate(ws: WebSocket, kind: str) -> Optional[tuple[str, str, str]]: + """Run the pre-accept gates for /api/console and /api/pty. + + Each gate maps to a distinct close code so the log and the browser banner + agree on the cause: 4404 chat disabled, 4401 bad credential, 4403 + host/origin mismatch, 4408 peer not allowed. Returns ``(peer, mode, cred)`` + once every gate passes, or None after closing the socket. + """ + peer = ws.client.host if ws.client else "?" + if not _DASHBOARD_EMBEDDED_CHAT_ENABLED: + _log.info("%s refused: embedded chat disabled peer=%s", kind, peer) + await ws.close(code=4404, reason="embedded chat disabled") + return None + + auth_reason, cred = _ws_auth_reason(ws) + mode = _ws_auth_mode() + if auth_reason is not None: + _log.warning("%s auth rejected reason=%s mode=%s cred=%s peer=%s", kind, auth_reason, mode, cred, peer) + await ws.close(code=4401, reason=_ws_close_reason(f"auth: {auth_reason}")) + return None + + host_origin_reason = _ws_host_origin_reason(ws) + if host_origin_reason is not None: + _log.warning("%s refused: %s peer=%s", kind, host_origin_reason, peer) + await ws.close(code=4403, reason=_ws_close_reason(host_origin_reason)) + return None + + client_reason = _ws_client_reason(ws) + if client_reason is not None: + _log.warning("%s refused: %s", kind, client_reason) + await ws.close(code=4408, reason=_ws_close_reason(client_reason)) + return None + return peer, mode, cred + + +async def _close_unless_sidecar_allowed(ws: WebSocket) -> bool: + """Pre-accept gates for the /api/ws, /api/pub and /api/events sidecars: + 4403 when chat is disabled or the request isn't allowed, 4401 on bad auth.""" + if not _DASHBOARD_EMBEDDED_CHAT_ENABLED: + await ws.close(code=4403) + return False + if not _ws_auth_ok(ws): + await ws.close(code=4401) + return False + if not _ws_request_is_allowed(ws): + await ws.close(code=4403) + return False + return True + + +# --- /api/console: the curated console engine, in-process, exchanging JSON +# frames with the dashboard xterm overlay. Never spawns a PTY, shell or CLI. _CONSOLE_PROMPT = "hermes> " - - _CONSOLE_COMMAND_TIMEOUT_SECONDS = 60.0 - - _CONSOLE_OUTPUT_LIMIT = 50000 -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: +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) +class _ConsoleSender: + """Serialises frames onto one console socket and owns the prompt suffix.""" + def __init__(self, ws: WebSocket) -> None: + self.ws = ws + self.lock = asyncio.Lock() -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, - }, + async def send(self, payload: Dict[str, Any]) -> None: + async with self.lock: + await self.ws.send_json(payload) + + async def prompt(self, **payload: Any) -> None: + await self.send({**payload, "prompt": _CONSOLE_PROMPT}) + + async def error(self, message: str, *, id: Optional[int] = None, command: Optional[str] = None, + prompt: Optional[str] = None) -> None: + # Key order matches the historical frames: type, id, message, command, prompt. + frame: Dict[str, Any] = {"type": "error"} + if id is not None: + frame["id"] = id + frame["message"] = message + if command is not None: + frame["command"] = command + if prompt is not None: + frame["prompt"] = prompt + await self.send(frame) + + async def complete(self, status: str, command: str, command_id: int, *, prompt: str = _CONSOLE_PROMPT) -> None: + await self.send({"type": "complete", "id": command_id, "status": status, "command": command, "prompt": prompt}) + + async def error_then_complete(self, message: str, command: str, command_id: int, status: str) -> None: + await self.error(message, id=command_id, command=command) + await self.complete(status, command, command_id) + + async def send_result(self, result: Any, *, command_id: int) -> None: + command = result.command or "" + status = result.status + if status == "ok": + if result.output: + await self.send({ + "type": "output", "id": command_id, "stream": "stdout", + "data": result.output, "command": command, + }) + await self.complete("ok", command, command_id) + elif status == "error": + await self.error_then_complete(result.output or "Command failed.", command, command_id, "error") + elif status == "confirm_required": + await self.prompt( + type="confirm_required", id=command_id, command=command, + message=result.confirmation_message or f"Run `{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, - }, - ) + await self.complete("confirm_required", command, command_id) + elif status == "clear": + await self.send({"type": "clear", "id": command_id}) + await self.complete("clear", command, command_id) + elif status == "exit": + await self.complete("exit", command, command_id, prompt="") + else: + await self.error(f"Unknown console result status: {status}", id=command_id, command=command) def _console_json_payload(msg: Any) -> tuple[Optional[dict[str, Any]], Optional[str]]: @@ -309,40 +247,14 @@ def _console_json_payload(msg: Any) -> tuple[Optional[dict[str, Any]], Optional[ @router.websocket("/api/console") async def console_ws(ws: WebSocket) -> None: - from hermes_cli.web_server import _DASHBOARD_EMBEDDED_CHAT_ENABLED - 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") + gate = await _ws_gate(ws, "console") + if gate is None: 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 - + peer, mode, cred = gate await ws.accept() - profile = _console_profile_from_ws(ws) - send_lock = asyncio.Lock() + profile = (ws.query_params.get("profile") or "").strip() or None + out = _ConsoleSender(ws) try: from hermes_cli.console_engine import HermesConsoleEngine @@ -351,47 +263,17 @@ async def console_ws(ws: WebSocket) -> None: 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 out.error(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 out.error(f"Console unavailable: {exc}", prompt="") await ws.close(code=1011) return - _log.info( - "console accepted peer=%s mode=%s cred=%s profile=%s", - peer, - mode, - cred, - profile or "current", - ) - await _console_send( - ws, - send_lock, - { - "type": "ready", - "profile": profile or "current", - "prompt": _CONSOLE_PROMPT, - }, - ) + _log.info("console accepted peer=%s mode=%s cred=%s profile=%s", peer, mode, cred, profile or "current") + await out.prompt(type="ready", profile=profile or "current") active_task: asyncio.Task | None = None pending_confirmation: Optional[str] = None @@ -404,13 +286,7 @@ async def console_ws(ws: WebSocket) -> None: result = await asyncio.wait_for( loop.run_in_executor( _get_console_executor(), - functools.partial( - _execute_console_line, - engine, - line, - confirmed=confirmed, - profile=profile, - ), + functools.partial(_execute_console_line, engine, line, confirmed=confirmed, profile=profile), ), timeout=_CONSOLE_COMMAND_TIMEOUT_SECONDS, ) @@ -419,79 +295,29 @@ async def console_ws(ws: WebSocket) -> None: 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, - }, + await out.error_then_complete( + "Command timed out. Hermes Console returned to the prompt.", line, command_id, "timeout", ) 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, - }, - ) + await out.error_then_complete(str(exc) or exc.__class__.__name__, line, command_id, "error") 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, - ) + pending_confirmation = result.command if result.status == "confirm_required" else None + await out.send_result(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: + 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) - ) + active_task = asyncio.create_task(run_command(line, confirmed=confirmed, command_id=command_generation)) try: while True: @@ -499,35 +325,19 @@ async def console_ws(ws: WebSocket) -> None: msg = await ws.receive() except RuntimeError: break - msg_type = msg.get("type") - if msg_type == "websocket.disconnect": + if msg.get("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, - }, - ) + await out.prompt(type="error", message=error) 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, - }, - ) + await out.prompt(type="pong") continue if frame_type == "cancel": @@ -536,117 +346,45 @@ async def console_ws(ws: WebSocket) -> None: active_task.cancel() active_task = None pending_confirmation = None - await _console_send( - ws, - send_lock, - { - "type": "complete", - "status": "cancelled", - "prompt": _CONSOLE_PROMPT, - }, - ) + await out.prompt(type="complete", status="cancelled") elif pending_confirmation: pending_confirmation = None - await _console_send( - ws, - send_lock, - { - "type": "complete", - "status": "cancelled", - "prompt": _CONSOLE_PROMPT, - }, - ) + await out.prompt(type="complete", status="cancelled") else: - await _console_send( - ws, - send_lock, - { - "type": "complete", - "status": "idle", - "prompt": _CONSOLE_PROMPT, - }, - ) + await out.prompt(type="complete", status="idle") 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, - }, - ) + await out.prompt(type="error", message="A console command is already running.") 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, - }, - ) + await out.prompt(type="error", message="No command is waiting for confirmation.") continue if command != pending_confirmation: - await _console_send( - ws, - send_lock, - { - "type": "error", - "message": "Confirmation does not match the pending command.", - "prompt": _CONSOLE_PROMPT, - }, - ) + await out.prompt(type="error", message="Confirmation does not match the pending command.") continue pending_confirmation = None - await start_command(command, confirmed=True) + 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, - }, - ) + await out.prompt(type="complete", status="ok") 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, - }, + await out.prompt( + type="error", + message="Confirm or cancel the pending command before running another one.", ) continue - await start_command(line) + start_command(line) continue - await _console_send( - ws, - send_lock, - { - "type": "error", - "message": f"Unsupported console frame: {frame_type or '?'}", - "prompt": _CONSOLE_PROMPT, - }, - ) + await out.prompt(type="error", message=f"Unsupported console frame: {frame_type or '?'}") except WebSocketDisconnect: pass finally: @@ -658,56 +396,28 @@ async def console_ws(ws: WebSocket) -> None: pass +async def _pty_fail(ws: WebSocket, text: str) -> None: + await ws.send_text(f"\r\n\x1b[31m{text}\x1b[0m\r\n") + await ws.close(code=1011) + + @router.websocket("/api/pty") async def pty_ws(ws: WebSocket) -> None: from hermes_cli.web_server import ( PTY_REGISTRY, PtyBridge, PtyUnavailableError, - _DASHBOARD_EMBEDDED_CHAT_ENABLED, _PTY_BRIDGE_AVAILABLE, _RESIZE_RE, ) - peer = ws.client.host if ws.client else "?" - - if not _DASHBOARD_EMBEDDED_CHAT_ENABLED: - _log.info("pty refused: embedded chat disabled peer=%s", peer) - await ws.close(code=4404, reason="embedded chat disabled") + gate = await _ws_gate(ws, "pty") + if gate is None: return - - # --- auth + host/origin/peer check (before accept so we can close - # cleanly AND tell the client WHY via the close code + reason). - # Each gate maps to a distinct close code so the log and the - # browser banner agree on the cause: - # 4401 bad credential 4403 host/origin mismatch - # 4408 peer not allowed 4404 chat disabled - auth_reason, cred = _ws_auth_reason(ws) - mode = _ws_auth_mode() - if auth_reason is not None: - _log.warning( - "pty 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("pty 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("pty refused: %s", client_reason) - await ws.close(code=4408, reason=_ws_close_reason(client_reason)) - return - + peer, mode, cred = gate await ws.accept() _log.info("pty accepted peer=%s mode=%s cred=%s", peer, mode, cred) - # On native Windows, the POSIX PTY bridge can't be imported. Tell the - # client and close cleanly rather than pretending the feature works. + # Native Windows can't import the POSIX PTY bridge: say so and close cleanly. if not _PTY_BRIDGE_AVAILABLE: await ws.send_text( "\r\n\x1b[31mChat unavailable: the embedded terminal requires a " @@ -718,56 +428,43 @@ async def pty_ws(ws: WebSocket) -> None: await ws.close(code=1011) return - # --- spawn PTY ------------------------------------------------------ raw_resume = ws.query_params.get("resume") or None resume = raw_resume profile = ws.query_params.get("profile") or None channel = _channel_or_close_code(ws) sidecar_url = _build_sidecar_url(channel) if channel else None - force_fresh = (ws.query_params.get("fresh") or "").strip().lower() in { - "1", - "true", - "yes", - "on", - } + force_fresh = (ws.query_params.get("fresh") or "").strip().lower() in {"1", "true", "yes", "on"} active_session_file: Optional[Path] = None if channel: active_session_file = _active_session_file_for_channel(ws.app, channel) if force_fresh: resume = None - _forget_active_session_file(active_session_file) + try: + active_session_file.unlink(missing_ok=True) + except OSError: + pass elif not resume: resume = _read_active_session_file(active_session_file) if resume: - # The client only knows to pin the viewport to the bottom - # when it requested `?resume=`. Tell it a replay is coming - # anyway so the implicit active-session fallback gets the - # same follow-scroll treatment as an explicit resume (#93518). + # The client only pins the viewport to the bottom when it asked + # for `?resume=`; announce the implicit active-session replay so + # it gets the same follow-scroll treatment. await ws.send_json({"type": "resume", "id": resume}) - resolve_kwargs = { - "resume": resume, - "sidecar_url": sidecar_url, - "profile": profile, - } + resolve_kwargs = {"resume": resume, "sidecar_url": sidecar_url, "profile": profile} if active_session_file is not None: resolve_kwargs["active_session_file"] = str(active_session_file) try: argv, cwd, env = await _resolve_chat_argv_async(**resolve_kwargs) - except HTTPException as exc: - # Unknown/invalid profile from _resolve_profile_dir. - await ws.send_text(f"\r\n\x1b[31mChat unavailable: {exc.detail}\x1b[0m\r\n") - await ws.close(code=1011) + except HTTPException as exc: # unknown/invalid profile + await _pty_fail(ws, f"Chat unavailable: {exc.detail}") return - except SystemExit as exc: - # _make_tui_argv calls sys.exit(1) when node/npm is missing. - await ws.send_text(f"\r\n\x1b[31mChat unavailable: {exc}\x1b[0m\r\n") - await ws.close(code=1011) + except SystemExit as exc: # _make_tui_argv sys.exit(1)s when node/npm is missing + await _pty_fail(ws, f"Chat unavailable: {exc}") return - attach_token = ws.query_params.get("attach") or None registry_resume = raw_resume if raw_resume and env: @@ -780,52 +477,38 @@ async def pty_ws(ws: WebSocket) -> None: return PtyBridge.spawn(argv, cwd=cwd, env=env) if attach_token is None: - # Legacy path: 1:1 socket<->PTY, killed on disconnect (unchanged). + # Legacy path: 1:1 socket<->PTY, killed on disconnect. try: bridge = _spawn() except PtyUnavailableError as exc: - await ws.send_text(f"\r\n\x1b[31mChat unavailable: {exc}\x1b[0m\r\n") - await ws.close(code=1011) + await _pty_fail(ws, f"Chat unavailable: {exc}") return except (FileNotFoundError, OSError) as exc: - await ws.send_text(f"\r\n\x1b[31mChat failed to start: {exc}\x1b[0m\r\n") - await ws.close(code=1011) + await _pty_fail(ws, f"Chat failed to start: {exc}") return await _legacy_pump(ws, bridge) return # Keep-alive path: the PTY outlives this socket; reattach by token. try: - session, _created = await PTY_REGISTRY.attach_or_spawn( - attach_token, spawn=_spawn - ) - except PtyUnavailableError as exc: - await ws.send_text(f"\r\n\x1b[31mChat unavailable: {exc}\x1b[0m\r\n") - await ws.close(code=1011) - return - except (FileNotFoundError, OSError, RegistryFull) as exc: - await ws.send_text(f"\r\n\x1b[31mChat unavailable: {exc}\x1b[0m\r\n") - await ws.close(code=1011) + session, _created = await PTY_REGISTRY.attach_or_spawn(attach_token, spawn=_spawn) + except (PtyUnavailableError, FileNotFoundError, OSError, RegistryFull) as exc: + await _pty_fail(ws, f"Chat unavailable: {exc}") return - # A fresh xterm cannot reliably reconstruct the TUI from an arbitrary - # bounded tail of alternate-screen, differential ANSI output. Reused PTYs - # emit a complete frame after replay so reconnects never reopen blank. + # A fresh xterm can't rebuild the TUI from an arbitrary tail of alternate- + # screen differential output; reused PTYs emit a full frame after replay. await session.attach(ws, force_redraw=not _created) - # --- writer loop: WebSocket → PTY master ---------------------------- - # No reader task here: the session's drain task (spawned once per PTY, - # inside the registry) forwards PTY output to whichever socket is - # attached and rings-buffers it while detached. On child EOF the drain - # closes the attached socket with 4410, which unparks ``ws.receive()`` - # below — same half-open-socket protection the legacy pump has (#54028). + # Writer loop only: the session's drain task (one per PTY, inside the + # registry) forwards output to whichever socket is attached and ring-buffers + # it while detached. On child EOF it closes the attached socket with 4410, + # which unparks ws.receive() — same half-open protection as the legacy pump. try: while True: try: msg = await ws.receive() - except RuntimeError: - # ws.receive() after the socket is already disconnected - # (e.g. closed by the drain task on process exit). + except RuntimeError: # receive() after the drain task already closed us break if msg.get("type") == "websocket.disconnect": break @@ -835,13 +518,11 @@ async def pty_ws(ws: WebSocket) -> None: 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): session.bridge.resize(cols=int(match.group(1)), rows=int(match.group(2))) continue - session.bridge.write(raw) except WebSocketDisconnect: pass @@ -851,37 +532,20 @@ async def pty_ws(ws: WebSocket) -> None: PTY_REGISTRY.detach(attach_token, ws) -# --------------------------------------------------------------------------- -# /api/ws — JSON-RPC WebSocket sidecar for the dashboard "Chat" tab. -# -# Drives the same `tui_gateway.dispatch` surface Ink uses over stdio, so the -# dashboard can render structured metadata (model badge, tool-call sidebar, -# slash launcher, session info) alongside the xterm.js terminal that PTY -# already paints. Both transports bind to the same session id when one is -# active, so a tool.start emitted by the agent fans out to both sinks. -# --------------------------------------------------------------------------- +# --- /api/ws: JSON-RPC sidecar for the Chat tab. Drives the same +# tui_gateway.dispatch surface Ink uses over stdio so the dashboard can render +# structured metadata next to the xterm; both transports bind to the same +# session id, so agent emits fan out to both sinks. @router.websocket("/api/ws") async def gateway_ws(ws: WebSocket) -> None: - from hermes_cli.web_server import _DASHBOARD_EMBEDDED_CHAT_ENABLED - if not _DASHBOARD_EMBEDDED_CHAT_ENABLED: - await ws.close(code=4403) + if not await _close_unless_sidecar_allowed(ws): return - - if not _ws_auth_ok(ws): - await ws.close(code=4401) - return - - if not _ws_request_is_allowed(ws): - await ws.close(code=4403) - return - from tui_gateway.ws import handle_ws - # The authenticated identity (ticket / internal credential) was stamped - # onto the WS object by _ws_auth_reason; carry it into the gateway - # transport where it becomes the identity authority for privileged RPCs + # The authenticated identity (ticket / internal credential) stamped by + # _ws_auth_reason becomes the identity authority for privileged RPCs # (browser.controller.register). None on the legacy token path. await handle_ws( ws, @@ -890,40 +554,29 @@ async def gateway_ws(ws: WebSocket) -> None: ) -# --------------------------------------------------------------------------- -# /api/pub + /api/events — chat-tab event broadcast. -# -# The PTY-side ``tui_gateway.entry`` opens /api/pub at startup (driven by -# HERMES_TUI_SIDECAR_URL set in /api/pty's PTY env) and writes every -# dispatcher emit through it. The dashboard fans those frames out to any -# subscriber that opened /api/events on the same channel id. This is what -# gives the React sidebar its tool-call feed without breaking the PTY +# --- /api/pub + /api/events: the PTY-side tui_gateway.entry opens /api/pub +# (HERMES_TUI_SIDECAR_URL from /api/pty's env) and writes every dispatcher emit +# through it; the dashboard fans frames out to /api/events subscribers on the +# same channel — the React sidebar's tool-call feed without touching the PTY # child's stdio handshake with Ink. -# --------------------------------------------------------------------------- + + +async def _accept_channel_ws(ws: WebSocket) -> Optional[str]: + if not await _close_unless_sidecar_allowed(ws): + return None + channel = _channel_or_close_code(ws) + if not channel: + await ws.close(code=4400) + return None + await ws.accept() + return channel @router.websocket("/api/pub") async def pub_ws(ws: WebSocket) -> None: - from hermes_cli.web_server import _DASHBOARD_EMBEDDED_CHAT_ENABLED - if not _DASHBOARD_EMBEDDED_CHAT_ENABLED: - await ws.close(code=4403) + channel = await _accept_channel_ws(ws) + if channel is None: return - - if not _ws_auth_ok(ws): - await ws.close(code=4401) - return - - if not _ws_request_is_allowed(ws): - await ws.close(code=4403) - return - - channel = _channel_or_close_code(ws) - if not channel: - await ws.close(code=4400) - return - - await ws.accept() - try: while True: await _broadcast_event(ws.app, channel, await ws.receive_text()) @@ -933,44 +586,22 @@ async def pub_ws(ws: WebSocket) -> None: @router.websocket("/api/events") async def events_ws(ws: WebSocket) -> None: - from hermes_cli.web_server import _DASHBOARD_EMBEDDED_CHAT_ENABLED - if not _DASHBOARD_EMBEDDED_CHAT_ENABLED: - await ws.close(code=4403) + channel = await _accept_channel_ws(ws) + if channel is None: return - - if not _ws_auth_ok(ws): - await ws.close(code=4401) - return - - if not _ws_request_is_allowed(ws): - await ws.close(code=4403) - return - - channel = _channel_or_close_code(ws) - if not channel: - await ws.close(code=4400) - return - - await ws.accept() - event_channels, event_lock = _get_event_state(ws.app) async with event_lock: event_channels.setdefault(channel, set()).add(ws) - try: while True: - # Subscribers don't speak — the receive() just blocks until - # disconnect so the connection stays open as long as the - # browser holds it. + # Subscribers don't speak — receive() just blocks until disconnect. await ws.receive_text() except WebSocketDisconnect: pass finally: async with event_lock: subs = event_channels.get(channel) - if subs is not None: subs.discard(ws) - if not subs: event_channels.pop(channel, None) diff --git a/hermes_cli/web_routers/cron.py b/hermes_cli/web_routers/cron.py index 2cc466462d..a9784e199a 100644 --- a/hermes_cli/web_routers/cron.py +++ b/hermes_cli/web_routers/cron.py @@ -8,20 +8,15 @@ late-binding seam so ``monkeypatch.setattr(web_server, ...)`` keeps working. import asyncio import functools import time -from typing import Optional +from pathlib import Path +from typing import Any, Dict, List, Optional from fastapi import APIRouter, HTTPException, Request from fastapi.responses import JSONResponse from hermes_cli.web_deps import late -from hermes_cli.web_models import ( - CronJobCreate, - CronJobUpdate, - AutomationBlueprintInstantiate, -) +from hermes_cli.web_models import AutomationBlueprintInstantiate, CronJobCreate, CronJobUpdate from hermes_cli.web_routers._common import log as _log -from typing import Any, Dict, List -from pathlib import Path router = APIRouter() @@ -46,47 +41,51 @@ _cron_optional_text = late("_cron_optional_text") _cron_string_list = late("_cron_string_list") _normalize_dashboard_cron_script = late("_normalize_dashboard_cron_script") +def _job_not_found() -> HTTPException: + return HTTPException(status_code=404, detail="Job not found") -def _normalize_dashboard_cron_updates( - updates: Dict[str, Any], - profile_home: Path, -) -> Dict[str, Any]: + +def _normalize_dashboard_cron_updates(updates: Dict[str, Any], profile_home: Path) -> Dict[str, Any]: """Normalize dashboard JSON into cron.jobs.update_job's storage shape. - This intentionally stays in the dashboard adapter layer: cron/jobs.py is the - source of truth for scheduling behaviour; the dashboard only translates form - payloads into the shapes that existing core functions already accept. + Stays in the dashboard adapter layer on purpose: cron/jobs.py is the source + of truth for scheduling; this only translates form payloads into shapes the + core functions already accept. """ normalized = dict(updates or {}) - for key in ("model", "provider", "workdir"): if key in normalized: normalized[key] = _cron_optional_text(normalized[key]) if "script" in normalized: - normalized["script"] = _normalize_dashboard_cron_script( - normalized["script"], - profile_home, - ) + normalized["script"] = _normalize_dashboard_cron_script(normalized["script"], profile_home) if "base_url" in normalized: - normalized["base_url"] = _cron_optional_text( - normalized["base_url"], strip_trailing_slash=True - ) + normalized["base_url"] = _cron_optional_text(normalized["base_url"], strip_trailing_slash=True) if "deliver" in normalized: normalized["deliver"] = _cron_optional_text(normalized["deliver"]) or "local" if "failure_deliver" in normalized: - # Same text normalization as deliver, but empty CLEARS the override - # (failures fall back to deliver) rather than coalescing to a target - # — the field is optional by design (NS-788). - normalized["failure_deliver"] = _cron_optional_text( - normalized["failure_deliver"] - ) - if "context_from" in normalized: - normalized["context_from"] = _cron_string_list(normalized["context_from"]) - if "enabled_toolsets" in normalized: - normalized["enabled_toolsets"] = _cron_string_list(normalized["enabled_toolsets"]) + # Same normalization as deliver, but empty CLEARS the override (failures + # fall back to deliver) rather than coalescing — the field is optional. + normalized["failure_deliver"] = _cron_optional_text(normalized["failure_deliver"]) + for key in ("context_from", "enabled_toolsets"): + if key in normalized: + normalized[key] = _cron_string_list(normalized[key]) return normalized +def _job_profile(job_id: str, profile: Optional[str]) -> str: + """Profile owning ``job_id`` (explicit or discovered); 404 when none.""" + selected = profile or _find_cron_job_profile(job_id) + if not selected: + raise _job_not_found() + return selected + + +def _found(job): + if not job: + raise _job_not_found() + return job + + def _list_cron_jobs_sync(profile: str = "all"): requested = (profile or "all").strip() if requested.lower() != "all": @@ -105,29 +104,17 @@ def _list_cron_jobs_sync(profile: str = "all"): def _get_cron_job_sync(job_id: str, profile: Optional[str] = None): - selected = profile or _find_cron_job_profile(job_id) - if not selected: - raise HTTPException(status_code=404, detail="Job not found") - job = _call_cron_for_profile(selected, "get_job", job_id) - if not job: - raise HTTPException(status_code=404, detail="Job not found") - return job + return _found(_call_cron_for_profile(_job_profile(job_id, profile), "get_job", job_id)) def _list_cron_job_runs_sync(job_id: str, profile: Optional[str] = None, limit: int = 20): """Run sessions produced by a cron job, newest first. - Cron runs are stored as ordinary sessions whose id is - ``cron_{job_id}_{timestamp}`` (see cron/scheduler.run_job). A job's history - is therefore every session whose id carries that prefix; ``source='cron'`` - narrows it and the id prefix binds it to this job. Powers the run-history - list under each job in the desktop cron detail. Same row shape as - ``/api/sessions`` so the frontend can reuse SessionInfo. - - Backed by ``SessionDB.list_cron_job_runs`` — a bounded ``[prefix, hi)`` - id-range scan, not the compression-chain CTE used for the recents list, - so the cost scales with the requested window and not the (unbounded) total - cron history. + Runs are ordinary sessions with id ``cron_{job_id}_{timestamp}`` (see + cron/scheduler.run_job); ``source='cron'`` plus the id prefix binds them to + this job. Same row shape as ``/api/sessions`` so the frontend reuses + SessionInfo. Backed by ``SessionDB.list_cron_job_runs`` — a bounded id-range + scan, so cost scales with the requested window, not total cron history. """ selected = profile or _find_cron_job_profile(job_id) # job_id may be a human name; resolve to the canonical id used in run-session ids. @@ -147,10 +134,7 @@ def _list_cron_job_runs_sync(job_id: str, profile: Optional[str] = None, limit: runs = db.list_cron_job_runs(canonical, limit=limit_n, offset=0) now = time.time() for s in runs: - s["is_active"] = ( - s.get("ended_at") is None - and (now - s.get("last_active", s.get("started_at", 0))) < 300 - ) + s["is_active"] = s.get("ended_at") is None and (now - s.get("last_active", s.get("started_at", 0))) < 300 s["archived"] = bool(s.get("archived")) if selected: s["profile"] = selected @@ -159,26 +143,18 @@ def _list_cron_job_runs_sync(job_id: str, profile: Optional[str] = None, limit: db.close() +_EXECUTION_FIELDS = {"prompt", "skill", "skills", "script", "no_agent"} + + def _update_cron_job_sync(job_id: str, body: CronJobUpdate, profile: Optional[str] = None): - selected = profile or _find_cron_job_profile(job_id) - if not selected: - raise HTTPException(status_code=404, detail="Job not found") + selected = _job_profile(job_id, profile) try: profile_name, profile_home = _cron_profile_home(selected) - existing = _call_cron_for_profile(profile_name, "get_job", job_id) - if not existing: - raise HTTPException(status_code=404, detail="Job not found") - updates = _normalize_dashboard_cron_updates( - body.updates, - profile_home, - ) + existing = _found(_call_cron_for_profile(profile_name, "get_job", job_id)) + updates = _normalize_dashboard_cron_updates(body.updates, profile_home) if "context_from" in updates: - _validate_dashboard_cron_context_from( - updates.get("context_from"), - profile_name, - ) - execution_fields = {"prompt", "skill", "skills", "script", "no_agent"} - if execution_fields.intersection(updates): + _validate_dashboard_cron_context_from(updates.get("context_from"), profile_name) + if _EXECUTION_FIELDS.intersection(updates): effective = {**existing, **updates} if "skills" in updates and "skill" not in updates: effective["skill"] = None @@ -188,80 +164,53 @@ def _update_cron_job_sync(job_id: str, body: CronJobUpdate, profile: Optional[st raise except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc - if not job: - raise HTTPException(status_code=404, detail="Job not found") - return job + return _found(job) def _pause_cron_job_sync(job_id: str, profile: Optional[str] = None): - selected = profile or _find_cron_job_profile(job_id) - if not selected: - raise HTTPException(status_code=404, detail="Job not found") - job = _mutate_cron_for_profile(selected, "pause_job", job_id) - if not job: - raise HTTPException(status_code=404, detail="Job not found") - return job + return _found(_mutate_cron_for_profile(_job_profile(job_id, profile), "pause_job", job_id)) def _resume_cron_job_sync(job_id: str, profile: Optional[str] = None): - selected = profile or _find_cron_job_profile(job_id) - if not selected: - raise HTTPException(status_code=404, detail="Job not found") - job = _mutate_cron_for_profile(selected, "resume_job", job_id) - if not job: - raise HTTPException(status_code=404, detail="Job not found") - return job + return _found(_mutate_cron_for_profile(_job_profile(job_id, profile), "resume_job", job_id)) def _trigger_cron_job_sync(job_id: str, profile: Optional[str] = None): - selected = profile or _find_cron_job_profile(job_id) - if not selected: - raise HTTPException(status_code=404, detail="Job not found") - job = _call_cron_for_profile(selected, "resolve_job_ref", job_id) - if not job: - raise HTTPException(status_code=404, detail="Job not found") - # Do not expose the job as due before claiming it: the built-in ticker and - # external/manual fire paths share the same durable claim, so only one can - # execute this selected run even if they race across processes. Active jobs - # keep the legacy provider call shape; paused jobs need the explicit force - # flag to resume and claim atomically. + selected = _job_profile(job_id, profile) + job = _found(_call_cron_for_profile(selected, "resolve_job_ref", job_id)) + # Never expose the job as due before claiming it: the built-in ticker and + # external/manual fire paths share one durable claim, so only one executes + # this run even racing across processes. Active jobs keep the legacy call + # shape; paused jobs need the explicit force flag to resume + claim atomically. force = not job.get("enabled", True) or job.get("state") == "paused" ran = _fire_cron_job_for_profile(selected, job["id"], force=force) refreshed = _call_cron_for_profile(selected, "get_job", job["id"]) if refreshed and refreshed.get("last_run_at") != job.get("last_run_at"): return refreshed if not ran: - raise HTTPException( - status_code=409, - detail="Job is already running or was claimed by another scheduler", - ) + raise HTTPException(status_code=409, detail="Job is already running or was claimed by another scheduler") if refreshed: return refreshed - # A one-shot may remove itself after exhausting repeat=1. Keep the response - # shape compatible without inventing an outcome that is no longer present - # in the job store; authoritative list refresh removes the completed row. - return { - **job, - "enabled": False, - "state": "completed", - } + # A one-shot may remove itself after exhausting repeat=1: keep the response + # shape without inventing an outcome the store no longer holds; the list + # refresh removes the completed row. + return {**job, "enabled": False, "state": "completed"} def _delete_cron_job_sync(job_id: str, profile: Optional[str] = None): - selected = profile or _find_cron_job_profile(job_id) - if not selected: - raise HTTPException(status_code=404, detail="Job not found") + selected = _job_profile(job_id, profile) try: removed = _mutate_cron_for_profile(selected, "remove_job", job_id) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc if not removed: - raise HTTPException(status_code=404, detail="Job not found") + raise _job_not_found() return {"ok": True} -# Retry-After hint (seconds) on retryable cron-fire 503s: sized to clear a -# scale-to-zero wake or gateway restart so a scheduler that honors it spaces -# its next attempt past the outage instead of burning its retry budget in it. + +# Retry-After (seconds) on retryable cron-fire 503s: sized to clear a +# scale-to-zero wake or gateway restart so a scheduler that honors it spaces its +# next attempt past the outage instead of burning its retry budget in it. _CRON_FIRE_RETRY_AFTER_SECONDS = 60 @@ -290,14 +239,7 @@ async def get_cron_delivery_targets(): """Delivery targets for the cron dropdown: implicit ``local`` plus the configured gateway platforms (a platform without a cron home channel is still listed with ``home_target_set: false`` so the UI can say so).""" - targets = [ - { - "id": "local", - "name": "Local (save only)", - "home_target_set": True, - "home_env_var": None, - } - ] + targets = [{"id": "local", "name": "Local (save only)", "home_target_set": True, "home_env_var": None}] try: from cron.scheduler import cron_delivery_targets @@ -336,13 +278,12 @@ async def delete_cron_job(job_id: str, profile: Optional[str] = None): async def cron_fire_webhook(request: Request): """Chronos managed-cron fire webhook (NAS -> agent) — gateway forwarder. - Gated by the NAS-minted JWT (this path is in ``PUBLIC_API_PATHS``), not the - dashboard cookie. The dashboard is only the public door: execution belongs - to the GATEWAY process (it owns the live platform adapters relay-fronted - and E2EE targets need), so the fire is forwarded to the gateway - api_server's own ``/api/cron/fire`` on loopback and its response passed - through (the gateway re-verifies the JWT). Gateway unreachable -> 503 so - NAS retries; deliberately NO local-execution fallback. + Gated by the NAS-minted JWT (path is in ``PUBLIC_API_PATHS``), not the + dashboard cookie. Execution belongs to the GATEWAY process (it owns the live + platform adapters relay-fronted and E2EE targets need), so the fire is + forwarded to the gateway api_server's own ``/api/cron/fire`` on loopback + and its response passed through (the gateway re-verifies the JWT). Gateway + unreachable -> 503 so NAS retries; deliberately NO local-execution fallback. """ from plugins.cron_providers.chronos.verify import get_fire_verifier @@ -369,15 +310,14 @@ async def cron_fire_webhook(request: Request): # Walks every profile's job list (file I/O) — off the event loop. profile = await _run_cron_dashboard_io(_find_cron_job_profile, job_id) - if not profile: - # Job is gone (cancelled / completed): 200 so NAS does not retry. + if not profile: # job is gone (cancelled / completed): 200 so NAS does not retry return JSONResponse({"status": "gone", "job_id": job_id}, status_code=200) forwarded = await _forward_cron_fire_to_gateway(profile, job_id, auth) if forwarded is None: # Stamp the miss on the job record (last_fire_error) so the dead hop is - # visible in `cronjob list` / the dashboard, not just gui.log. - # Best-effort: visibility must never break the retry contract below. + # visible in `cronjob list` / the dashboard. Best-effort: visibility + # must never break the retry contract below. try: await _run_cron_dashboard_io( _call_cron_for_profile, @@ -394,7 +334,7 @@ async def cron_fire_webhook(request: Request): # Split by operator intent: a deliberately stopped gateway (durable # desired_state == "stopped") can never be reached by retrying, so drop # with 200 + a structured log line — the Chronos provider re-arms every - # job on the next gateway start. A transient window (wake, restart, + # job on the next gateway start. A transient window (wake, restart, # crash loop) keeps the retryable 503 with a Retry-After hint. if await _run_cron_dashboard_io(_gateway_intentionally_stopped, profile): _log.info( @@ -406,27 +346,21 @@ async def cron_fire_webhook(request: Request): return JSONResponse( { "status": "gateway_stopped", - "detail": "gateway deliberately stopped; fire dropped, " - "jobs re-arm on next gateway start", + "detail": "gateway deliberately stopped; fire dropped, jobs re-arm on next gateway start", "job_id": job_id, "profile": profile, }, status_code=200, ) return JSONResponse( - { - "error": "gateway unreachable; retry", - "job_id": job_id, - "profile": profile, - }, + {"error": "gateway unreachable; retry", "job_id": job_id, "profile": profile}, status_code=503, headers={"Retry-After": str(_CRON_FIRE_RETRY_AFTER_SECONDS)}, ) status_code, gateway_body = forwarded if isinstance(gateway_body, dict): gateway_body.setdefault("job_id", job_id) - # The gateway's own 503s (draining, admission failure) are equally - # transient — same spacing hint. + # The gateway's own 503s (draining, admission failure) are equally transient. headers = {"Retry-After": str(_CRON_FIRE_RETRY_AFTER_SECONDS)} if status_code == 503 else None return JSONResponse(gateway_body, status_code=status_code, headers=headers) @@ -465,21 +399,20 @@ async def list_cron_blueprints(): async def instantiate_blueprint(body: AutomationBlueprintInstantiate, profile: str = "default"): """Fill a blueprint's slots and create the cron job (form-submit path).""" try: - from cron.blueprint_catalog import fill_blueprint, get_blueprint, BlueprintFillError + from cron.blueprint_catalog import BlueprintFillError, fill_blueprint, get_blueprint blueprint = get_blueprint(body.blueprint) if blueprint is None: raise HTTPException(status_code=404, detail=f"Unknown blueprint: {body.blueprint}") try: spec = fill_blueprint(blueprint, body.values) - except BlueprintFillError as exc: - # Field-level validation error — 422 so the form can show it inline. + except BlueprintFillError as exc: # field-level error — 422 so the form shows it inline raise HTTPException(status_code=422, detail=str(exc)) from exc - # Blueprint jobs deliver to the dashboard's configured target by - # default; the form's deliver slot overrides via spec["deliver"]. + # Blueprint jobs deliver to the dashboard's configured target by default; + # the form's deliver slot overrides via spec["deliver"]. spec.pop("origin", None) - # Off the event loop like the sibling endpoints; partial keeps **spec - # keys from colliding with the wrapper's own parameters. + # Off-loop like the siblings; partial keeps **spec keys from colliding + # with the wrapper's own parameters. _create = functools.partial(_call_cron_for_profile, profile, "create_job", **spec) created = await _run_cron_dashboard_io(_create) # Reconcile the profile-scoped provider (file I/O + NAS calls) off-loop. diff --git a/hermes_cli/web_routers/files.py b/hermes_cli/web_routers/files.py index 3600e922b7..0daa58fb4e 100644 --- a/hermes_cli/web_routers/files.py +++ b/hermes_cli/web_routers/files.py @@ -1,31 +1,35 @@ """Managed-files, chat image upload, /api/media and /api/fs dashboard routes. -Extracted from ``hermes_cli.web_server``; helpers/state that tests monkeypatch on -``web_server`` stay there and are imported lazily at call time (cycle-safe). +Helpers/state that tests monkeypatch on ``web_server`` stay there and are +reached through the late-binding seam (cycle-safe). """ +import asyncio import base64 import binascii +import contextlib import mimetypes import os import re -import stat -import tempfile -import asyncio import secrets import shutil +import stat import subprocess import sys +import tempfile from datetime import datetime -from fastapi import APIRouter -from hermes_cli.web_deps import late -from fastapi import File, Form, HTTPException, Request, UploadFile -from fastapi.responses import FileResponse -from hermes_cli._subprocess_compat import windows_hide_flags -from hermes_cli.web_models import ManagedFileUpload, ChatImageUpload, ManagedDirectoryCreate, ManagedFileDelete, FsWriteText from pathlib import Path from typing import Any, Dict, Optional +from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile +from fastapi.responses import FileResponse + +from hermes_cli._subprocess_compat import windows_hide_flags +from hermes_cli.web_deps import late +from hermes_cli.web_models import ( + ChatImageUpload, FsWriteText, ManagedDirectoryCreate, ManagedFileDelete, ManagedFileUpload, +) + router = APIRouter() # web_server helpers, late-bound so monkeypatch.setattr(web_server, ...) stays authoritative. @@ -38,114 +42,48 @@ get_hermes_home = late("get_hermes_home") load_config = late("load_config") -# Image MIME types this endpoint will serve. Extension-allowlisted so an -# authenticated caller can't pull non-image files through it. +# Image types GET /api/media serves — extension-allowlisted so an authenticated +# caller can't pull non-image files through it. _MEDIA_CONTENT_TYPES = { - ".png": "image/png", - ".jpg": "image/jpeg", - ".jpeg": "image/jpeg", - ".gif": "image/gif", - ".webp": "image/webp", - ".svg": "image/svg+xml", - ".bmp": "image/bmp", - ".ico": "image/x-icon", + ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".gif": "image/gif", + ".webp": "image/webp", ".svg": "image/svg+xml", ".bmp": "image/bmp", ".ico": "image/x-icon", } - - _MEDIA_MAX_BYTES = 25 * 1024 * 1024 - -_STREAMABLE_MEDIA_EXTENSIONS = frozenset( - { - ".avi", - ".flac", - ".m4a", - ".mkv", - ".mov", - ".mp3", - ".mp4", - ".ogg", - ".opus", - ".wav", - ".webm", - } -) - +_STREAMABLE_MEDIA_EXTENSIONS = frozenset({ + ".avi", ".flac", ".m4a", ".mkv", ".mov", ".mp3", ".mp4", ".ogg", ".opus", ".wav", ".webm", +}) _FS_READDIR_HIDDEN = { - ".git", - ".hg", - ".svn", - ".cache", - ".next", - ".turbo", - ".venv", - "__pycache__", - "build", - "dist", - "node_modules", - "target", - "venv", + ".git", ".hg", ".svn", ".cache", ".next", ".turbo", ".venv", "__pycache__", + "build", "dist", "node_modules", "target", "venv", } - -# Filenames that must never be listed, read, or downloaded through the -# managed-files API. These typically contain credentials (API keys, tokens) -# and exposing them through the dashboard file browser is a security leak — -# see issue #57505. The set mirrors the credential-file basenames of the two -# canonical credential guards elsewhere in the codebase -# (agent.file_safety.get_read_block_error and -# gateway.platforms.base._ROOT_CREDENTIAL_FILES) so the dashboard Files tab -# doesn't lag behind them — an operator can point the managed root at -# HERMES_HOME itself, at which point every one of these basenames is a live -# secret store sitting in the browsable tree. +# Basenames the managed-files API must never list, read or download: credential +# stores that become live secrets in the browsable tree the moment an operator +# points the managed root at HERMES_HOME. Mirrors the two canonical guards +# (agent.file_safety.get_read_block_error, gateway.platforms.base +# ._ROOT_CREDENTIAL_FILES) so the Files tab never lags behind them. _SENSITIVE_MANAGED_FILE_BASENAMES = frozenset({ - "auth.json", - "auth.lock", - "credentials", - "config.yaml", - ".anthropic_oauth.json", - "google_token.json", - "google_oauth_pending.json", - "google_oauth.json", - "webhook_subscriptions.json", - "bws_cache.json", - "bws_cache.enc.json", - # git's credential-store helper cache (agent.file_safety blocks this too). - ".git-credentials", + "auth.json", "auth.lock", "credentials", "config.yaml", ".anthropic_oauth.json", + "google_token.json", "google_oauth_pending.json", "google_oauth.json", + "webhook_subscriptions.json", "bws_cache.json", "bws_cache.enc.json", + ".git-credentials", # git's credential-store cache (file_safety blocks it too) }) - -# Directory names whose entire subtree is credential material. Both canonical -# guards deny these as directory trees, not basenames: -# * gateway.platforms.base._ROOT_CREDENTIAL_DIRS = {"pairing", "mcp-tokens"} -# * agent.file_safety.get_read_block_error (mcp-tokens/ prefix match) -# The managed-files API lets the browser descend into subdirs, so a -# basename-only guard would still expose e.g. ``mcp-tokens/.json`` -# (live MCP OAuth tokens) and ``pairing/``. We match on ANY path component -# so these trees are blocked wherever they appear under the browsable root, -# without needing to resolve them relative to HERMES_HOME. -_SENSITIVE_MANAGED_DIR_NAMES = frozenset({ - "mcp-tokens", - "pairing", -}) +# Directory names whose whole subtree is credential material (the canonical +# guards deny these as trees: _ROOT_CREDENTIAL_DIRS and the mcp-tokens/ prefix +# match). The browser can descend into subdirs, so a basename-only guard would +# still expose ``mcp-tokens/.json``; match on ANY path component so the +# trees are blocked wherever they sit under the root, no HERMES_HOME resolution. +_SENSITIVE_MANAGED_DIR_NAMES = frozenset({"mcp-tokens", "pairing"}) def _is_sensitive_filename(name: str) -> bool: - """Return True for a basename the managed-files API must never expose. - - Covers ``.env`` / ``.env.`` / ``.envrc`` variants plus the - canonical Hermes credential-store basenames (see - ``_SENSITIVE_MANAGED_FILE_BASENAMES`` above). - - Case-insensitive so ``.ENV`` / ``.Env.local`` / ``Auth.JSON`` on - case-insensitive filesystems (macOS/Windows mounts) can't slip past - the guard. - - Basename-only: for the directory-tree credential stores - (``mcp-tokens/``, ``pairing/``) that the canonical guards also deny, - use :func:`_is_sensitive_path`, which the API call sites route through. - """ + """Basename denylist: ``.env`` / ``.env.`` / ``.envrc`` plus the + credential-store basenames. Case-insensitive so ``.ENV`` / ``Auth.JSON`` + on case-insensitive mounts can't slip past. Basename-only — call sites use + :func:`_is_sensitive_path`, which adds the credential-directory check.""" lowered = name.lower() if lowered == ".env" or lowered.startswith(".env.") or lowered == ".envrc": return True @@ -153,94 +91,35 @@ def _is_sensitive_filename(name: str) -> bool: def _is_sensitive_path(path: Path) -> bool: - """Return True for any path the managed-files API must never expose. - - Combines the basename denylist (:func:`_is_sensitive_filename`) with a - credential-directory-tree check: a path is sensitive if its own basename - is sensitive OR any of its path components is a credential directory - (``mcp-tokens`` / ``pairing``). The component match is case-insensitive - and needs no HERMES_HOME resolution, so it blocks these trees wherever - they sit under the operator-configured managed root — closing the gap - the canonical guards cover as directory trees but a basename-only check - would miss. - - Read-side only: this guards list/read/download (the #57505 exfil surface). - The write endpoints (upload/mkdir/delete) are a separate threat class - handled by the write-path checks; extending this guard to them is out of - scope for this fix. - """ + """True when the basename is sensitive OR any path component (case- + insensitive) is a credential directory. Read-side guard (list/read/ + download); the write endpoints are a separate threat class.""" if _is_sensitive_filename(path.name): return True return any(part.lower() in _SENSITIVE_MANAGED_DIR_NAMES for part in path.parts) _FS_TEXT_SOURCE_MAX_BYTES = 64 * 1024 * 1024 - - _FS_TEXT_PREVIEW_MAX_BYTES = 512 * 1024 - - -# Upper bound for the in-app spot editor's save. The editor only opens -# non-truncated text (<= the preview cap), so this is a safety ceiling against -# a pasted-in megablob, not the expected payload size. +# Spot-editor save ceiling: the editor only opens non-truncated text (<= the +# preview cap), so this guards against a pasted megablob, not expected payloads. _FS_TEXT_WRITE_MAX_BYTES = 8 * 1024 * 1024 - _FS_PREVIEW_LANGUAGE_BY_EXT = { - ".c": "c", - ".conf": "ini", - ".cpp": "cpp", - ".css": "css", - ".csv": "csv", - ".go": "go", - ".graphql": "graphql", - ".h": "c", - ".hpp": "cpp", - ".html": "html", - ".java": "java", - ".js": "javascript", - ".json": "json", - ".jsx": "jsx", - ".kt": "kotlin", - ".lua": "lua", - ".md": "markdown", - ".mjs": "javascript", - ".py": "python", - ".rb": "ruby", - ".rs": "rust", - ".sh": "shell", - ".sql": "sql", - ".svg": "xml", - ".toml": "toml", - ".ts": "typescript", - ".tsx": "tsx", - ".txt": "text", - ".xml": "xml", - ".yaml": "yaml", - ".yml": "yaml", - ".zsh": "shell", + ".c": "c", ".conf": "ini", ".cpp": "cpp", ".css": "css", ".csv": "csv", ".go": "go", + ".graphql": "graphql", ".h": "c", ".hpp": "cpp", ".html": "html", ".java": "java", + ".js": "javascript", ".json": "json", ".jsx": "jsx", ".kt": "kotlin", ".lua": "lua", + ".md": "markdown", ".mjs": "javascript", ".py": "python", ".rb": "ruby", ".rs": "rust", + ".sh": "shell", ".sql": "sql", ".svg": "xml", ".toml": "toml", ".ts": "typescript", + ".tsx": "tsx", ".txt": "text", ".xml": "xml", ".yaml": "yaml", ".yml": "yaml", ".zsh": "shell", } - _FS_MIME_TYPES = { - ".avi": "video/x-msvideo", - ".bmp": "image/bmp", - ".flac": "audio/flac", - ".gif": "image/gif", - ".jpeg": "image/jpeg", - ".jpg": "image/jpeg", - ".m4a": "audio/mp4", - ".mkv": "video/x-matroska", - ".mov": "video/quicktime", - ".mp3": "audio/mpeg", - ".mp4": "video/mp4", - ".ogg": "audio/ogg", - ".opus": "audio/ogg; codecs=opus", - ".png": "image/png", - ".svg": "image/svg+xml", - ".wav": "audio/wav", - ".webm": "video/webm", - ".webp": "image/webp", + ".avi": "video/x-msvideo", ".bmp": "image/bmp", ".flac": "audio/flac", ".gif": "image/gif", + ".jpeg": "image/jpeg", ".jpg": "image/jpeg", ".m4a": "audio/mp4", ".mkv": "video/x-matroska", + ".mov": "video/quicktime", ".mp3": "audio/mpeg", ".mp4": "video/mp4", ".ogg": "audio/ogg", + ".opus": "audio/ogg; codecs=opus", ".png": "image/png", ".svg": "image/svg+xml", + ".wav": "audio/wav", ".webm": "video/webm", ".webp": "image/webp", } @@ -261,13 +140,22 @@ def _fs_looks_binary(data: bytes) -> bool: return suspicious / len(data) > 0.12 +@contextlib.contextmanager +def _io_errors(denied: str, failed: str): + """PermissionError -> 403 ``denied``; other OSError -> 500 ``": "``.""" + try: + yield + except PermissionError: + raise HTTPException(status_code=403, detail=denied) + except OSError as exc: + raise HTTPException(status_code=500, detail=f"{failed}: {exc}") + + def _fs_regular_file(path: Path) -> tuple[Path, os.stat_result]: target = _fs_path(str(path)) try: st = target.stat() - except FileNotFoundError: - raise HTTPException(status_code=404, detail="File not found") - except NotADirectoryError: + except (FileNotFoundError, NotADirectoryError): raise HTTPException(status_code=404, detail="File not found") except PermissionError: raise HTTPException(status_code=403, detail="File is not readable") @@ -280,6 +168,19 @@ def _fs_regular_file(path: Path) -> tuple[Path, os.stat_result]: return target, st +def _fs_read_bytes(target: Path, limit: Optional[int] = None) -> bytes: + """Read (a prefix of) ``target``; 403/400 on failure.""" + try: + if limit is None: + return target.read_bytes() + with target.open("rb") as handle: + return handle.read(limit) + except PermissionError: + raise HTTPException(status_code=403, detail="File is not readable") + except OSError as exc: + raise HTTPException(status_code=400, detail=str(exc) or "File read failed") + + def _fs_find_git_root(start: Path) -> str | None: directory = start for _ in range(50): @@ -310,36 +211,22 @@ def _fs_default_cwd() -> str: def _fs_git_branch(cwd: str) -> str: try: - run_kwargs: Dict[str, Any] = { - "capture_output": True, - "text": True, - "timeout": 2, - "check": False, - } + run_kwargs: Dict[str, Any] = {"capture_output": True, "text": True, "timeout": 2, "check": False} if sys.platform == "win32": run_kwargs["creationflags"] = windows_hide_flags() - result = subprocess.run( - ["git", "-C", cwd, "branch", "--show-current"], - **run_kwargs, - ) + result = subprocess.run(["git", "-C", cwd, "branch", "--show-current"], **run_kwargs) return result.stdout.strip() if result.returncode == 0 else "" except Exception: return "" def _media_serve_roots() -> list[Path]: - """Directories ``GET /api/media`` is allowed to read from. - - Confined to where the agent and attach pipeline actually write media on the - gateway host — its images dir and cache subtree. This stops an authenticated - client from reading image-extension files anywhere on disk (e.g. a renamed - key or a screenshot outside the cache) merely because the suffix passes the - allowlist. - """ + """Directories GET /api/media may read: where the agent and attach pipeline + actually write media (images, screenshots, cache). Stops an authenticated + client reading image-suffixed files anywhere else on disk.""" home = get_hermes_home() - roots = [home / "images", home / "screenshots", home / "cache"] out: list[Path] = [] - for root in roots: + for root in (home / "images", home / "screenshots", home / "cache"): try: out.append(root.resolve()) except (OSError, RuntimeError): @@ -349,16 +236,9 @@ def _media_serve_roots() -> list[Path]: @router.get("/api/media") async def get_media(path: str): - """Return a gateway-local image file as a base64 data URL. - - Lets remote clients (the desktop app over the network, or the web dashboard - in a browser) display images the agent wrote to *this* machine's filesystem - — they can't read the gateway's local disk directly. - - Auth-gated by the session token like every other /api route. Restricted to - an image-extension allowlist, a size cap, AND the gateway's own media roots - (resolved, symlink-safe) so it can't be used to read arbitrary files. - """ + """Return a gateway-local image as a base64 data URL for remote clients + that can't read this machine's disk. Auth-gated; restricted to the image + allowlist, a size cap AND the resolved (symlink-safe) media roots.""" try: target = Path(path).expanduser().resolve() except (OSError, RuntimeError): @@ -366,11 +246,9 @@ async def get_media(path: str): if target.suffix.lower() not in _MEDIA_CONTENT_TYPES: raise HTTPException(status_code=415, detail="Unsupported media type") - roots = _media_serve_roots() if not any(target == root or root in target.parents for root in roots): raise HTTPException(status_code=403, detail="Path outside media roots") - if not target.is_file(): raise HTTPException(status_code=404, detail="File not found") if target.stat().st_size > _MEDIA_MAX_BYTES: @@ -380,15 +258,6 @@ async def get_media(path: str): return {"data_url": f"data:{_MEDIA_CONTENT_TYPES[target.suffix.lower()]};base64,{encoded}"} -def _local_dashboard_request(request: Request) -> bool: - if getattr(request.app.state, "auth_required", False): - return False - host = (request.url.hostname or "").lower() - client_host = (request.client.host if request.client else "").lower() - local_hosts = {"", "localhost", "127.0.0.1", "::1", "testserver", "testclient"} - return host in local_hosts or client_host in local_hosts - - def _decode_data_url(data_url: str) -> tuple[bytes, str]: from hermes_cli.web_server import _MANAGED_FILE_MAX_BYTES text = (data_url or "").strip() @@ -408,25 +277,17 @@ def _decode_data_url(data_url: str) -> tuple[bytes, str]: _CHAT_IMAGE_UPLOAD_MAX_BYTES = 25 * 1024 * 1024 - - _CHAT_IMAGE_ALLOWED_EXTENSIONS = frozenset({".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp"}) - - _CHAT_IMAGE_MAGIC: tuple[tuple[bytes, str], ...] = ( - (b"\x89PNG\r\n\x1a\n", ".png"), - (b"\xff\xd8\xff", ".jpg"), - (b"GIF87a", ".gif"), - (b"GIF89a", ".gif"), - (b"BM", ".bmp"), + (b"\x89PNG\r\n\x1a\n", ".png"), (b"\xff\xd8\xff", ".jpg"), + (b"GIF87a", ".gif"), (b"GIF89a", ".gif"), (b"BM", ".bmp"), ) def _sanitize_chat_image_filename(filename: str | None) -> str: candidate = Path(str(filename or "").strip()).name candidate = re.sub(r"[\x00-\x1f]+", "_", candidate) - candidate = candidate.strip().strip(".") - return candidate or "pasted-image" + return candidate.strip().strip(".") or "pasted-image" def _chat_image_extension(data: bytes) -> str | None: @@ -446,7 +307,6 @@ def _decode_chat_image_upload(payload: ChatImageUpload) -> tuple[bytes, str, str if len(data) > _CHAT_IMAGE_UPLOAD_MAX_BYTES: mb = _CHAT_IMAGE_UPLOAD_MAX_BYTES // (1024 * 1024) raise HTTPException(status_code=413, detail=f"Image is too large; cap is {mb} MB") - ext = _chat_image_extension(data) if ext not in _CHAT_IMAGE_ALLOWED_EXTENSIONS: raise HTTPException(status_code=400, detail="Unsupported image type") @@ -455,38 +315,26 @@ def _decode_chat_image_upload(payload: ChatImageUpload) -> tuple[bytes, str, str @router.post("/api/chat/image-upload") async def upload_chat_image(payload: ChatImageUpload, profile: Optional[str] = None): - """Persist a browser-provided chat image where the embedded TUI can read it. + """Persist a browser clipboard image where the embedded TUI can read it. - The dashboard /chat page runs Hermes inside an xterm.js PTY. Browser - clipboard image bytes are not visible to the server-side clipboard, so the - page uploads them here, then drives the TUI's ``/image `` command - with the returned gateway-visible path. Files land under - ``HERMES_HOME/images/`` — the same directory ``clipboard.paste`` / - ``image.attach`` already use. + Browser clipboard bytes aren't visible to the server-side clipboard, so the + /chat page uploads them here and drives the TUI's ``/image `` with + the returned gateway-visible path under ``HERMES_HOME/images/`` (the same + dir ``clipboard.paste`` / ``image.attach`` use). """ def _run(): data, mime_type, ext = _decode_chat_image_upload(payload) with _profile_scope(profile) as scoped_home: - home = scoped_home or get_hermes_home() - img_dir = Path(home) / "images" - try: + img_dir = Path(scoped_home or get_hermes_home()) / "images" + with _io_errors("Image directory is not writable", "Could not create image directory"): img_dir.mkdir(parents=True, exist_ok=True) - except PermissionError: - raise HTTPException(status_code=403, detail="Image directory is not writable") - except OSError as exc: - raise HTTPException(status_code=500, detail=f"Could not create image directory: {exc}") stem = Path(_sanitize_chat_image_filename(payload.filename)).stem or "pasted-image" stem = re.sub(r"[^A-Za-z0-9_.-]+", "_", stem).strip("._-") or "pasted-image" ts = datetime.now().strftime("%Y%m%d_%H%M%S") target = img_dir / f"dashboard_{ts}_{secrets.token_hex(4)}_{stem}{ext}" - - try: + with _io_errors("Image directory is not writable", "Could not write image"): target.write_bytes(data) - except PermissionError: - raise HTTPException(status_code=403, detail="Image directory is not writable") - except OSError as exc: - raise HTTPException(status_code=500, detail=f"Could not write image: {exc}") return { "ok": True, @@ -496,9 +344,9 @@ async def upload_chat_image(payload: ChatImageUpload, profile: Optional[str] = N "mime_type": mime_type, } - # _profile_scope acquires _SKILLS_PROFILE_LOCK and the body does file I/O — - # keep both off the event loop (asyncio.to_thread copies the contextvar - # context, so the profile override stays scoped to the worker thread). + # _profile_scope takes _SKILLS_PROFILE_LOCK and the body does file I/O — both + # off the loop; to_thread copies the contextvar context so the override + # stays scoped to the worker thread. return await asyncio.to_thread(_run) @@ -510,33 +358,25 @@ async def list_managed_files(request: Request, path: Optional[str] = None): if not target.is_dir(): raise HTTPException(status_code=400, detail="Path is not a directory") - try: - with os.scandir(target) as scan: - entries = [ - _managed_file_entry(policy, Path(entry.path)) - for entry in scan - if not _is_sensitive_path(Path(entry.path)) - ] - except PermissionError: - raise HTTPException(status_code=403, detail="Directory is not readable") - except OSError as exc: - raise HTTPException(status_code=500, detail=f"Could not read directory: {exc}") + with _io_errors("Directory is not readable", "Could not read directory"), os.scandir(target) as scan: + entries = [ + _managed_file_entry(policy, Path(entry.path)) + for entry in scan + if not _is_sensitive_path(Path(entry.path)) + ] entries.sort(key=lambda item: (not item["is_directory"], str(item["name"]).lower())) locked_root = policy.locked_root parent = None if target.parent != target and (locked_root is None or target != locked_root): parent = str(target.parent) - return { - "path": display_path, - "parent": parent, - "entries": entries, - **_managed_response_meta(policy), - } + return {"path": display_path, "parent": parent, "entries": entries, **_managed_response_meta(policy)} -@router.get("/api/files/read") -async def read_managed_file(request: Request, path: str): +def _managed_readable_file(request: Request, path: str) -> tuple[Any, Path, str, int, str]: + """Resolve + guard a managed file for reading: existence, regular file, + sensitive-path denylist, size cap. Returns (policy, target, display_path, + size, mime_type).""" from hermes_cli.web_server import _MANAGED_FILE_MAX_BYTES policy, target, display_path = _resolve_managed_path(path, request) if not target.exists(): @@ -545,22 +385,26 @@ async def read_managed_file(request: Request, path: str): raise HTTPException(status_code=400, detail="Path is not a file") if _is_sensitive_path(target): raise HTTPException(status_code=403, detail="Access to sensitive files is not allowed") + mime_type = mimetypes.guess_type(target.name)[0] or "application/octet-stream" + return policy, target, display_path, _MANAGED_FILE_MAX_BYTES, mime_type + +def _managed_file_size(target: Path, max_bytes: int) -> int: try: size = target.stat().st_size except OSError as exc: raise HTTPException(status_code=500, detail=f"Could not stat file: {exc}") - if size > _MANAGED_FILE_MAX_BYTES: + if size > max_bytes: raise HTTPException(status_code=413, detail="File is too large") + return size - mime_type = mimetypes.guess_type(target.name)[0] or "application/octet-stream" - try: + +@router.get("/api/files/read") +async def read_managed_file(request: Request, path: str): + policy, target, display_path, max_bytes, mime_type = _managed_readable_file(request, path) + size = _managed_file_size(target, max_bytes) + with _io_errors("File is not readable", "Could not read file"): encoded = base64.b64encode(target.read_bytes()).decode("ascii") - except PermissionError: - raise HTTPException(status_code=403, detail="File is not readable") - except OSError as exc: - raise HTTPException(status_code=500, detail=f"Could not read file: {exc}") - return { "name": target.name, "path": display_path, @@ -578,27 +422,11 @@ def _managed_file_response( content_disposition_type: str, media_only: bool = False, ) -> FileResponse: - """Build a range-aware response after applying managed-file policy.""" - from hermes_cli.web_server import _MANAGED_FILE_MAX_BYTES - policy, target, _display_path = _resolve_managed_path(path, request) - if not target.exists(): - raise HTTPException(status_code=404, detail="File not found") - if not target.is_file(): - raise HTTPException(status_code=400, detail="Path is not a file") - if _is_sensitive_path(target): - raise HTTPException(status_code=403, detail="Access to sensitive files is not allowed") + """Range-aware response after applying managed-file policy.""" + _policy, target, _display_path, max_bytes, mime_type = _managed_readable_file(request, path) if media_only and target.suffix.lower() not in _STREAMABLE_MEDIA_EXTENSIONS: raise HTTPException(status_code=415, detail="Unsupported media type") - - try: - size = target.stat().st_size - except OSError as exc: - raise HTTPException(status_code=500, detail=f"Could not stat file: {exc}") - if size > _MANAGED_FILE_MAX_BYTES: - raise HTTPException(status_code=413, detail="File is too large") - - mime_type = mimetypes.guess_type(target.name)[0] or "application/octet-stream" - + _managed_file_size(target, max_bytes) return FileResponse( path=str(target), media_type=mime_type, @@ -612,16 +440,11 @@ def _managed_file_response( async def download_managed_file(request: Request, path: str): """Stream a managed file as an attachment download. - Remote clients (desktop app, browser dashboard) open agent-written files - that live on *this* gateway's disk, not theirs. Auth-gated like every other - managed-files route — ``auth_middleware`` additionally accepts the session - token as a ``?token=`` query param here so a shell/browser-opened download - (which can't set the session header) still authenticates. See ``/api/pty`` - for the same query-token precedent. Chromium identifies ``