From b5eb5ee7a8a38ac96b62f44e0febf563922839e1 Mon Sep 17 00:00:00 2001 From: Bear Huddleston Date: Mon, 19 Oct 2026 23:26:58 -0400 Subject: [PATCH] feat(webapp): carry the browser-hosted Desktop's server pieces from #93508 Carried from BearHuddleston's #93508 onto main. The Webapp relay and the profile-incarnation fence stay out: carried code follows main's named-profile rule and refuses a missing or deleted profile home. - Webapp routes: one-use launch tickets, window handoff, and gateway WebSocket tickets minted for the signed-in browser - Browser surface policy and the SPA document policy - A host terminal for the browser that survives reconnects - Staged browser uploads and file tickets; file.attach takes a staged upload - hermes sessions adopt - Webapp bundle build helpers and the renderer freshness check - Regenerated gateway contracts Co-authored-by: emozilla --- .gitignore | 2 + apps/shared/src/gateway-contract.generated.ts | 7 + apps/shared/src/gateway-contract.openrpc.json | 36 + gateway/session_ancillary.py | 46 +- gateway/session_images.py | 8 +- hermes_cli/dashboard_auth/request_utils.py | 18 + hermes_cli/profiles.py | 2 +- hermes_cli/pty_bridge.py | 73 ++- hermes_cli/pty_session.py | 353 +++++++--- hermes_cli/sessions_adopt.py | 135 ++++ hermes_cli/sessions_cmd.py | 6 + hermes_cli/source_build.py | 16 +- hermes_cli/staged_uploads.py | 214 ++++++ hermes_cli/subcommands/sessions.py | 2 + hermes_cli/update_cmd_zip.py | 2 +- hermes_cli/web_host_terminal.py | 161 +++++ hermes_cli/web_host_terminal_sessions.py | 384 +++++++++++ hermes_cli/web_models.py | 7 + hermes_cli/web_routers/chat_ws.py | 99 +-- hermes_cli/web_routers/chat_ws_errors.py | 18 +- hermes_cli/web_routers/files.py | 185 +++++- hermes_cli/web_routers/uploads.py | 55 ++ hermes_cli/web_routers/webapp.py | 324 +++++++++ hermes_cli/web_server.py | 24 +- hermes_cli/web_server_app.py | 5 +- hermes_cli/web_server_chat.py | 147 ++++- hermes_cli/web_server_dashboard.py | 55 +- hermes_cli/web_server_file_response.py | 75 +++ hermes_cli/web_server_file_tickets.py | 105 +++ hermes_cli/web_server_files.py | 4 +- hermes_cli/web_server_surface.py | 47 ++ hermes_cli/web_upload_limit.py | 65 ++ hermes_cli/webapp.py | 88 +++ hermes_cli/win_pty_bridge.py | 96 ++- hermes_constants.py | 6 + locales/_keys.desktop.json | 3 + pm/filesystem.py | 13 + scripts/build/freshness.mjs | 77 ++- scripts/build/frontend-common.mjs | 2 +- scripts/build/webapp.mjs | 37 ++ scripts/ci/classify_changes.py | 1 + tests/gateway/fixtures/owner_process.py | 67 ++ .../fixtures/staged_file_attach_peer.py | 73 +++ .../test_canonical_file_attach_staged.py | 36 + ...st_canonical_file_attach_staged_sources.py | 160 +++++ .../hermes_cli/test_dashboard_auth_ws_auth.py | 63 ++ .../test_host_terminal_viewer_ownership.py | 207 ++++++ .../test_legacy_pump_close_reason.py | 37 ++ tests/hermes_cli/test_pty_bridge.py | 6 + tests/hermes_cli/test_pty_keepalive_ws.py | 3 + tests/hermes_cli/test_pty_session.py | 67 ++ .../hermes_cli/test_pty_session_continuity.py | 373 +++++++++++ .../test_serve_mcp_discovery_after_bind.py | 3 +- tests/hermes_cli/test_sessions_adopt.py | 115 ++++ tests/hermes_cli/test_spa_document_policy.py | 33 + .../test_update_zip_webapp_preserve.py | 89 +++ tests/hermes_cli/test_web_host_terminal.py | 151 +++++ .../test_web_host_terminal_drain_pool.py | 58 ++ .../test_web_host_terminal_lifespan.py | 127 ++++ .../test_web_host_terminal_locale.py | 59 ++ .../test_web_host_terminal_persistence.py | 618 ++++++++++++++++++ ...test_web_server_file_ticket_generations.py | 191 ++++++ tests/hermes_cli/test_web_server_files.py | 178 ++++- tests/hermes_cli/test_web_server_fs.py | 37 ++ .../hermes_cli/test_webapp_gateway_ticket.py | 70 ++ tests/hermes_cli/test_webapp_launch.py | 70 ++ .../test_webapp_private_bootstrap.py | 77 +++ .../hermes_cli/test_webapp_renderer_build.py | 145 ++++ tests/hermes_cli/test_webapp_upload_limit.py | 207 ++++++ tests/hermes_cli/test_webapp_uploads.py | 480 ++++++++++++++ .../hermes_cli/test_webapp_window_handoff.py | 92 +++ tests/hermes_cli/test_win_pty_bridge.py | 205 +++++- .../test_chat_ws_keepalive_takeover.py | 65 ++ .../scripts/test_webapp_renderer_freshness.py | 57 ++ tests/test_managed_runtime_resolution.py | 1 + tests/tui_gateway/test_tui_gateway_server.py | 8 +- tui_gateway/contracts/prompt_voice.py | 9 + tui_gateway/prompt_attachments.py | 65 +- web/AGENTS.md | 2 + 79 files changed, 6946 insertions(+), 361 deletions(-) create mode 100644 hermes_cli/sessions_adopt.py create mode 100644 hermes_cli/staged_uploads.py create mode 100644 hermes_cli/web_host_terminal.py create mode 100644 hermes_cli/web_host_terminal_sessions.py create mode 100644 hermes_cli/web_routers/uploads.py create mode 100644 hermes_cli/web_routers/webapp.py create mode 100644 hermes_cli/web_server_file_response.py create mode 100644 hermes_cli/web_server_file_tickets.py create mode 100644 hermes_cli/web_server_surface.py create mode 100644 hermes_cli/web_upload_limit.py create mode 100644 hermes_cli/webapp.py create mode 100644 scripts/build/webapp.mjs create mode 100644 tests/gateway/fixtures/owner_process.py create mode 100644 tests/gateway/fixtures/staged_file_attach_peer.py create mode 100644 tests/gateway/test_canonical_file_attach_staged.py create mode 100644 tests/gateway/test_canonical_file_attach_staged_sources.py create mode 100644 tests/hermes_cli/test_host_terminal_viewer_ownership.py create mode 100644 tests/hermes_cli/test_legacy_pump_close_reason.py create mode 100644 tests/hermes_cli/test_pty_session_continuity.py create mode 100644 tests/hermes_cli/test_sessions_adopt.py create mode 100644 tests/hermes_cli/test_spa_document_policy.py create mode 100644 tests/hermes_cli/test_update_zip_webapp_preserve.py create mode 100644 tests/hermes_cli/test_web_host_terminal.py create mode 100644 tests/hermes_cli/test_web_host_terminal_drain_pool.py create mode 100644 tests/hermes_cli/test_web_host_terminal_lifespan.py create mode 100644 tests/hermes_cli/test_web_host_terminal_locale.py create mode 100644 tests/hermes_cli/test_web_host_terminal_persistence.py create mode 100644 tests/hermes_cli/test_web_server_file_ticket_generations.py create mode 100644 tests/hermes_cli/test_webapp_gateway_ticket.py create mode 100644 tests/hermes_cli/test_webapp_launch.py create mode 100644 tests/hermes_cli/test_webapp_private_bootstrap.py create mode 100644 tests/hermes_cli/test_webapp_renderer_build.py create mode 100644 tests/hermes_cli/test_webapp_upload_limit.py create mode 100644 tests/hermes_cli/test_webapp_uploads.py create mode 100644 tests/hermes_cli/test_webapp_window_handoff.py create mode 100644 tests/hermes_cli/web_routers/test_chat_ws_keepalive_takeover.py create mode 100644 tests/scripts/test_webapp_renderer_freshness.py diff --git a/.gitignore b/.gitignore index cc99d65e24..47aac96cd9 100644 --- a/.gitignore +++ b/.gitignore @@ -94,6 +94,8 @@ hermes_cli/web_dist/ .web_ui_build.lock apps/desktop/build/ apps/desktop/dist/ +apps/desktop/dist-webapp/ +.dist-webapp-build-* # tsc-emitted artifacts (a stray `tsc -b` compiles into src/, and vite then # resolves the stale .js OVER the .tsx — never track these) diff --git a/apps/shared/src/gateway-contract.generated.ts b/apps/shared/src/gateway-contract.generated.ts index 796e8766a9..6e3d8d9f92 100644 --- a/apps/shared/src/gateway-contract.generated.ts +++ b/apps/shared/src/gateway-contract.generated.ts @@ -2826,6 +2826,13 @@ export interface FileAttachParams { path?: string | null data_url?: string | null name?: string | null + staged_upload?: StagedFileUpload | null +} +/** Source identity returned by the browser upload route, checked against the live profile. */ +export interface StagedFileUpload { + install_id: string + path: string + profile_home: string } export interface FileAttachResult { attached: boolean diff --git a/apps/shared/src/gateway-contract.openrpc.json b/apps/shared/src/gateway-contract.openrpc.json index ac6116ab1b..a5ebc6d5a9 100644 --- a/apps/shared/src/gateway-contract.openrpc.json +++ b/apps/shared/src/gateway-contract.openrpc.json @@ -18008,6 +18008,17 @@ ], "default": null, "title": "Name" + }, + "staged_upload": { + "anyOf": [ + { + "$ref": "#/components/schemas/StagedFileUpload" + }, + { + "type": "null" + } + ], + "default": null } }, "required": [ @@ -41749,6 +41760,31 @@ "title": "SpawnTreeSaveResult", "type": "object" }, + "StagedFileUpload": { + "additionalProperties": false, + "description": "Source identity returned by the browser upload route, checked against the live profile.", + "properties": { + "install_id": { + "title": "Install Id", + "type": "string" + }, + "path": { + "title": "Path", + "type": "string" + }, + "profile_home": { + "title": "Profile Home", + "type": "string" + } + }, + "required": [ + "install_id", + "path", + "profile_home" + ], + "title": "StagedFileUpload", + "type": "object" + }, "StatusUpdatePayload": { "additionalProperties": false, "description": "``server._status_update`` and the direct emitters (goal / loop / heartbeat / process).", diff --git a/gateway/session_ancillary.py b/gateway/session_ancillary.py index 6d9861cbf1..4cc3475b16 100644 --- a/gateway/session_ancillary.py +++ b/gateway/session_ancillary.py @@ -138,13 +138,21 @@ async def attach_file(connection, ref, params): """Desktop's non-image attachment: stage the file for this session (its frozen launch cwd and the owner profile's ``attachments/``) and hand back the ``@file:`` ref its next submit carries. The sidecar handler looked the session up in its own table, which never holds an authority - session, so every file drop failed with ``session not found``.""" - if (set(params) - {'session_id', 'profile', 'path', 'data_url', 'name'} or not ref.session_id - or any(not isinstance(value, str) for value in params.values())): + session, so every file drop failed with ``session not found``. + + A browser pick arrives as ``staged_upload`` (the Webapp upload route's receipt, + ``hermes_cli/staged_uploads.py``): it is copied out of the profile it was uploaded to, which must + be one this connection may reach.""" + staged_upload = params.get('staged_upload') + if (set(params) - {'session_id', 'profile', 'path', 'data_url', 'name', 'staged_upload'} or not ref.session_id + or any(not isinstance(value, str) for key, value in params.items() if key != 'staged_upload') + or (staged_upload is not None and not isinstance(staged_upload, dict))): raise RuntimeStoreError('invalid_params') if ref.session_id not in connection.subscriptions: raise RuntimeStoreError('permission_denied') authorize(connection, ref, params, 'session:submit') + if staged_upload is not None and not _reachable_upload_home(connection, staged_upload.get('profile_home')): + raise RuntimeStoreError('permission_denied') from gateway.session_policy import policy_for_source authority = connection.authority policy = policy_for_source(authority.runner, authority.sessions[ref.session_id].source) @@ -153,11 +161,17 @@ async def attach_file(connection, ref, params): # The staging helpers are bound onto the sidecar module's globals (method_ctx.bind_module). from tui_gateway import server session = {'cwd': policy.cwd, 'profile_home': authority.profile_id} + + def stage(): + if staged_upload is not None: + # Only the receipt is read: a supplied path/data_url never stands in for a missing upload. + return server._stage_browser_file_attachment(session, staged_upload, params.get('name', '')) + return server._stage_session_file_attachment( + session, raw_path=params.get('path', ''), data_url=params.get('data_url', ''), + name=params.get('name', '')) try: - stored, uploaded = await asyncio.to_thread( - server._stage_session_file_attachment, session, raw_path=params.get('path', ''), - data_url=params.get('data_url', ''), name=params.get('name', '')) - except ValueError as exc: + stored, uploaded = await asyncio.to_thread(stage) + except (ValueError, OverflowError) as exc: raise RuntimeStoreError('invalid_params') from exc except OSError as exc: raise RuntimeStoreError('storage_unavailable') from exc @@ -166,6 +180,24 @@ async def attach_file(connection, ref, params): 'ref_text': f'@file:{server._format_ref_value(ref_path)}', 'uploaded': uploaded} +def _reachable_upload_home(connection, raw_home) -> bool: + """A staged upload's source profile: this authority's own home or, for a host-scoped grant, a + sibling this process serves (an owner-routed tile can receive a foreground browser pick). Never + any other profile on disk, whatever its descriptor names.""" + from pathlib import Path + if not isinstance(raw_home, str) or not raw_home: + return False + homes = {connection.authority.profile_id} + if connection.host_scoped: + registry = getattr(getattr(connection.authority, 'runner', None), 'session_authorities', None) + homes |= {sibling.profile_id for sibling in (registry or ())} + try: + target = Path(raw_home).resolve() + return any(Path(home).resolve() == target for home in homes) + except (OSError, ValueError): + return False + + async def approvals(connection, ref, params, *, ack=False): """The Desktop's approval replay (``approval.pending``) and card ack (``approval.received``) over the owner's generation-bound prompt projection. Left to the legacy sidecar they answered diff --git a/gateway/session_images.py b/gateway/session_images.py index 72eee3d9e7..78a120af0e 100644 --- a/gateway/session_images.py +++ b/gateway/session_images.py @@ -24,8 +24,12 @@ def _stage(directory, params): from tui_gateway.prompt_attachments import ( _ATTACH_BYTES_MAX_BYTES, _decode_attach_base64, _sniff_image_ext, stage_image_bytes, ) - data = _decode_attach_base64(params.get('content_base64') or params.get('data') or '', mime_prefix='image/') - if not data or len(data) > _ATTACH_BYTES_MAX_BYTES: + try: + data = _decode_attach_base64(params.get('content_base64') or params.get('data') or '', + mime_prefix='image/', max_bytes=_ATTACH_BYTES_MAX_BYTES) + except OverflowError as exc: + raise RuntimeStoreError('invalid_params') from exc + if not data: raise RuntimeStoreError('invalid_params') hint = params.get('ext', '').strip().lower().lstrip('.') ext = _sniff_image_ext(data, params.get('filename') or (f'x.{hint}' if hint else '')) diff --git a/hermes_cli/dashboard_auth/request_utils.py b/hermes_cli/dashboard_auth/request_utils.py index 2eca3d145c..102615c623 100644 --- a/hermes_cli/dashboard_auth/request_utils.py +++ b/hermes_cli/dashboard_auth/request_utils.py @@ -4,6 +4,7 @@ from __future__ import annotations import logging import time from typing import Callable, Optional +from urllib.parse import urlsplit from fastapi import Request from fastapi.responses import JSONResponse @@ -34,6 +35,23 @@ def extract_bearer(request: Request) -> str: return "" +def _http_origin(value: str) -> tuple[str, str, int] | None: + """Parse one serialized HTTP origin, never an opaque or URL-like origin.""" + if not value or any(ord(c) <= 32 or ord(c) >= 127 or c in "\\,?#%" for c in value): + return None + try: + parsed = urlsplit(value) + if (parsed.scheme not in {"http", "https"} or not parsed.hostname + or parsed.username is not None or parsed.password is not None or parsed.path + or parsed.netloc.endswith(":")): + return None + port = parsed.port + effective_port = port if port is not None else (443 if parsed.scheme == "https" else 80) + return parsed.scheme, parsed.hostname, effective_port + except ValueError: + return None + + def is_safe_next_path(path: str) -> bool: """Same-origin post-login target: rejects non-relative and protocol-relative (``//evil``) values, the auth routes themselves, and every ``/api`` path.""" diff --git a/hermes_cli/profiles.py b/hermes_cli/profiles.py index 08b7eb1057..2a8d40da1a 100644 --- a/hermes_cli/profiles.py +++ b/hermes_cli/profiles.py @@ -273,7 +273,7 @@ _RESERVED_NAMES = frozenset({"hermes", "default", "test", "tmp", "root", "sudo"} # Hermes subcommands that cannot be used as profile names/aliases _HERMES_SUBCOMMANDS = frozenset({ "chat", "model", "gateway", "setup", "whatsapp", "login", "logout", - "status", "cron", "doctor", "dump", "config", "pairing", "skills", "tools", + "status", "cron", "doctor", "dump", "config", "pairing", "skills", "tools", "serve", "dashboard", "webapp", "mcp", "connectors", "sessions", "insights", "version", "update", "uninstall", "profile", "plugins", "honcho", "acp", }) diff --git a/hermes_cli/pty_bridge.py b/hermes_cli/pty_bridge.py index 375dcab600..80bd8c826f 100644 --- a/hermes_cli/pty_bridge.py +++ b/hermes_cli/pty_bridge.py @@ -17,8 +17,9 @@ import signal import struct import sys import termios # windows-footgun: ok — POSIX-only module by design (see docstring) +import threading import time -from typing import Optional, Sequence +from typing import Callable, Optional, Sequence try: import ptyprocess # type: ignore @@ -103,8 +104,8 @@ def _psutil_alive(proc) -> bool: class PtyBridge: """Thin wrapper around ``ptyprocess.PtyProcess`` for byte streaming. Not thread-safe: owned by the WebSocket handler that spawned it; reads run in an executor thread, writes are awaited on - the loop. The master fd is non-blocking so input backpressure suspends only the owning - WebSocket task, never the dashboard event loop. + the loop (or run in a worker under a caller's ``fence``). The master fd is non-blocking so + input backpressure suspends only the owning WebSocket task, never the dashboard event loop. """ _waiter = None # FdWaiter for the master; None until __init__ registers it @@ -113,6 +114,9 @@ class PtyBridge: self._proc = proc self._fd: int = proc.fd self._closed = False + # A fenced write runs in a worker thread; it must never reach a descriptor + # number that close() has already released for reuse. + self._fd_lock = threading.Lock() # Recorded at spawn: once the leader is reaped getpgid() can no longer find its group, # but the helpers it started still belong to it. try: @@ -197,9 +201,9 @@ class PtyBridge: raise return data or None - async def _wait_writable(self, timeout: float) -> bool: - """Wait without blocking the event loop until the master accepts input.""" - if self._closed or timeout <= 0: + async def _wait_writable(self, timeout: Optional[float]) -> bool: + """Wait without blocking the event loop until the master accepts input (``None``: no limit).""" + if self._closed or (timeout is not None and timeout <= 0): return False loop = asyncio.get_running_loop() ready = loop.create_future() @@ -210,7 +214,11 @@ class PtyBridge: try: loop.add_writer(self._fd, _mark_ready) - await asyncio.wait_for(ready, timeout=timeout) + # wait_for can swallow cancellation on Python 3.11 if readiness + # completes in the same tick as viewer takeover. Never retry input + # from that cancelled writer after the new viewer owns the PTY. + async with asyncio.timeout(timeout): + await ready return not self._closed except (TimeoutError, OSError, ValueError): return False @@ -220,12 +228,35 @@ class PtyBridge: except (OSError, ValueError): pass - async def write(self, data: bytes, *, timeout: float = 10.0) -> bool: + def _write_some(self, view: memoryview) -> Optional[int]: + """One non-blocking write: bytes accepted (0 while the child's buffer is full), None once closed.""" + with self._fd_lock: + if self._closed: + return None + try: + return os.write(self._fd, view) + except OSError as exc: + if exc.errno in {errno.EIO, errno.EBADF, errno.EPIPE}: + return None + if exc.errno in {errno.EAGAIN, errno.EWOULDBLOCK}: + return 0 + raise + + async def write( + self, data: bytes, *, timeout: Optional[float] = 10.0, + fence: Optional[Callable[[Callable[[], Optional[int]]], Optional[int]]] = None, + ) -> bool: """Write all raw bytes without ever blocking the dashboard event loop. Returns ``False`` when the bridge closes or the child leaves its input buffer full for ``timeout`` seconds. Callers can then recycle only the affected terminal session while the rest of the dashboard stays live. + ``timeout=None`` waits for as long as the child leaves its input unread + (cancel the call to stop waiting), so ``False`` then means the bridge closed. + + ``fence`` wraps every non-blocking write attempt in a worker thread, so a + caller can hold a lock around exactly the bytes that reach the child; + whatever it raises propagates. Backpressure waits happen outside it. """ if self._closed: return False @@ -233,20 +264,15 @@ class PtyBridge: return True loop = asyncio.get_running_loop() - deadline = loop.time() + max(0.0, timeout) + deadline = None if timeout is None else loop.time() + max(0.0, timeout) view = memoryview(data) while view: - if self._closed: + if fence is None: + n = self._write_some(view) + else: + n = await asyncio.to_thread(fence, lambda chunk=view: self._write_some(chunk)) + if n is None: return False - try: - n = os.write(self._fd, view) - except OSError as exc: - if exc.errno in {errno.EIO, errno.EBADF, errno.EPIPE}: - return False - if exc.errno in {errno.EAGAIN, errno.EWOULDBLOCK}: - n = 0 - else: - raise if n > 0: view = view[n:] # A very large paste can otherwise monopolize the loop while @@ -255,7 +281,7 @@ class PtyBridge: await asyncio.sleep(0) continue - remaining = deadline - loop.time() + remaining = None if deadline is None else deadline - loop.time() if not await self._wait_writable(remaining): return False return True @@ -298,9 +324,10 @@ class PtyBridge: the dashboard never leaks zombies, and close fds. Idempotent; blocks, so call it off the event loop. """ - if self._closed: - return - self._closed = True + with self._fd_lock: + if self._closed: + return + self._closed = True pgid = self._pgid leader_was_alive = self._proc.isalive() diff --git a/hermes_cli/pty_session.py b/hermes_cli/pty_session.py index a697602399..b79754b57b 100644 --- a/hermes_cli/pty_session.py +++ b/hermes_cli/pty_session.py @@ -7,12 +7,34 @@ opaque token replays the buffer and resumes live. from __future__ import annotations import asyncio +import logging +import threading import time +from concurrent.futures import Executor from pathlib import Path from typing import Callable, Dict, Optional, Tuple +from fastapi import WebSocketDisconnect + +logger = logging.getLogger(__name__) + WS_CLOSE_PROCESS_EXITED = 4410 WS_CLOSE_SUPERSEDED = 4409 +# Going Away: the viewer stopped taking output, but its session lives on and the same +# token reattaches. The dashboard chat tab and the Webapp terminal both redial on 1001; +# the chat tab reads 1011 as a failed start and offers a new session instead. +WS_CLOSE_VIEWER_STALLED = 1001 +WS_CLOSE_VIEWER_STALLED_REASON = "Terminal viewer stopped reading; reconnect" +# How long one frame may wait on the viewer before it counts as stalled. +_VIEWER_SEND_TIMEOUT_SECONDS = 5.0 +# A reattach replay is a single frame of up to the 1 MiB buffer (host-terminal clients take exactly +# the next binary frame as the replay, so it is not chunked). It gets the stall limit plus the time +# to cross a link this slow (about 256 kbit/s; 37 s for 1 MiB), else a weak connection redials forever. +_REPLAY_MIN_BYTES_PER_SECOND = 32 * 1024 +# A viewer send that fails because the browser is gone: Starlette raises WebSocketDisconnect for a +# dead transport and RuntimeError for a socket already closed; OSError covers the TimeoutError of +# a viewer that stopped reading. +_VIEWER_SEND_ERRORS = (WebSocketDisconnect, RuntimeError, OSError) TUI_FORCE_REDRAW = b"\x0c" @@ -35,10 +57,10 @@ class RingBuffer: return bytes(self._buf) -async def _close_ws(ws, code: int) -> None: +async def _close_ws(ws, code: int, reason: Optional[str] = None) -> None: try: if ws is not None: - await ws.close(code=code) + await asyncio.wait_for(ws.close(code=code, reason=reason), timeout=5.0) except Exception: pass @@ -66,7 +88,10 @@ def _key_segments(key: str) -> tuple[str, str]: class PtySession: - def __init__(self, key: str, bridge, *, buffer_cap: int, read_timeout: float, active_session_file: Optional[Path] = None) -> None: + def __init__( + self, key: str, bridge, *, buffer_cap: int, read_timeout: float, + active_session_file: Optional[Path] = None, drain_executor: Optional[Executor] = None, + ) -> None: self.key = key self.bridge = bridge self.active_session_file = active_session_file @@ -74,45 +99,90 @@ class PtySession: self.buffer = RingBuffer(buffer_cap) self.alive = True self.attached = False - self.last_detached_at: Optional[float] = None + # Admission can fail before the first attach (e.g. lost metadata). + self.last_detached_at: Optional[float] = time.monotonic() self._read_timeout = read_timeout + # Runs the blocking reads; None is the loop's default executor. + self._drain_executor = drain_executor self._ws = None self._attach_generation = 0 + # Only final sink writes and viewer claims hold this lock. Profile + # admission/backpressure waits happen outside it. + self._viewer_lock = threading.Lock() self._drain_task: Optional[asyncio.Task] = None self._write_lock = asyncio.Lock() + self._output_lock = asyncio.Lock() + self._input_task: Optional[asyncio.Task] = None + self._close_task: Optional[asyncio.Task] = None async def start(self) -> None: self._drain_task = asyncio.create_task(self._drain()) async def _drain(self) -> None: loop = asyncio.get_running_loop() - while True: - chunk = await loop.run_in_executor(None, self.bridge.read, self._read_timeout) + while self.alive: + try: + chunk = await loop.run_in_executor(self._drain_executor, self.bridge.read, self._read_timeout) + except OSError: + chunk = None if chunk is None: # EOF — the agent process exited self.alive = False - await _close_ws(self._ws, WS_CLOSE_PROCESS_EXITED) + ws = self._ws + async with self._output_lock: + if self._ws is ws: + await _close_ws(ws, WS_CLOSE_PROCESS_EXITED) return if not chunk: # idle tick await asyncio.sleep(0) continue self.buffer.append(chunk) ws = self._ws + output_lock = self._output_lock try: if ws is not None: - await ws.send_bytes(chunk) + async with output_lock: + if self._ws is ws: + await asyncio.wait_for(ws.send_bytes(chunk), timeout=_VIEWER_SEND_TIMEOUT_SECONDS) except Exception: # The viewer is gone; nothing else observes this failure (the handler's finally # only runs once ws.receive() sees the disconnect). detach() is a no-op when a # replacement socket attached during the send, so the new viewer keeps its session. - self.detach(ws) + if self._ws is ws: + self.detach(ws) + await _close_ws(ws, WS_CLOSE_VIEWER_STALLED, WS_CLOSE_VIEWER_STALLED_REASON) - async def write(self, ws, data: bytes) -> bool: - """Serialize input and discard bytes from a superseded socket.""" + async def write(self, ws, data: bytes, *, fence=None, **bridge_options) -> bool: + """Serialize input and discard bytes from a superseded socket. + + ``fence`` is handed to the bridge (see ``PtyBridge.write``); what it raises propagates. + ``bridge_options`` (``timeout``) go to the bridge's ``write`` unchanged. + """ async with self._write_lock: - if self._ws is not ws: + if self._ws is not ws or not self.alive: return True generation = self._attach_generation - delivered = await self.bridge.write(data) + + def viewer_fence(write): + def admitted_write(): + with self._viewer_lock: + if self._ws is not ws or self._attach_generation != generation or not self.alive: + return None + return write() + + return fence(admitted_write) + + if fence is not None: + bridge_options["fence"] = viewer_fence + task = self._input_task = asyncio.create_task(self.bridge.write(data, **bridge_options)) + try: + delivered = await task + except asyncio.CancelledError: + if self._attach_generation != generation: + return False + raise + finally: + if self._input_task is task: + self._input_task = None # A replacement socket can attach while the bridge write is # suspended on backpressure. A late failure from the superseded # socket must not poison the replacement's shared PTY session. @@ -124,26 +194,67 @@ class PtySession: self.alive = False return delivered - async def attach(self, ws, *, force_redraw: bool = False) -> bool: + def resize(self, ws, *, cols: int, rows: int) -> None: + if self._ws is ws and self.alive: + self.bridge.resize(cols=cols, rows=rows) + + async def attach(self, ws, *, force_redraw: bool = False, initial_text: Optional[str] = None) -> bool: """Attach a browser terminal and replay buffered PTY output. The TUI renders differentially on an alternate screen, so a bounded ANSI tail is not a self-contained frame; ``force_redraw`` asks the live TUI for one full redraw after replay. """ - if self._ws is not ws: - await _close_ws(self._ws, WS_CLOSE_SUPERSEDED) - self._ws = ws self._attach_generation += 1 - self.attached = True - self.last_detached_at = None - if snap := self.buffer.snapshot(): - try: - await ws.send_bytes(snap) - except Exception: - # Client dropped mid-replay; the caller never reaches its writer loop, so undo the - # attach here or reap_idle() can never reclaim this PTY (#110849). - self.detach(ws) + generation = self._attach_generation + if self._input_task is not None: + self._input_task.cancel() + # Revoke queued workers first, then serialize the claim after any + # already-admitted sink write. ConPTY writes can block: keep the loop + # free to run their timeout/cancellation cleanup while we wait. + while not self._viewer_lock.acquire(blocking=False): + await asyncio.sleep(0.001) + if self._attach_generation != generation: return False + try: + if self._attach_generation != generation: + return False + old_ws = self._ws + self._ws = ws + self.attached = True + self.last_detached_at = None + # Per-viewer lock: a stalled old send must not block its replacement. + output_lock = self._output_lock = asyncio.Lock() + finally: + self._viewer_lock.release() + try: + async with output_lock: + snap = self.buffer.snapshot() + if old_ws is not ws: + await _close_ws(old_ws, WS_CLOSE_SUPERSEDED) + if self._ws is not ws: + return False + if initial_text is not None: + await asyncio.wait_for(ws.send_text(initial_text), timeout=_VIEWER_SEND_TIMEOUT_SECONDS) + if self._ws is not ws: + return False + # Metadata-bearing host clients treat exactly the next binary + # frame as replay. Send an empty snapshot too, so the first live + # frame never loses terminal-query replies on an empty history. + if snap or initial_text is not None: + replay_timeout = _VIEWER_SEND_TIMEOUT_SECONDS + len(snap) / _REPLAY_MIN_BYTES_PER_SECOND + await asyncio.wait_for(ws.send_bytes(snap), timeout=replay_timeout) + except _VIEWER_SEND_ERRORS: + self.detach(ws) + return False + except BaseException: # cancellation or a bug: never leave this socket attached + self.detach(ws) + raise + if self._ws is not ws: + return False + if not self.alive: + self.detach(ws) + await _close_ws(ws, WS_CLOSE_PROCESS_EXITED) + return False if force_redraw: return await self.write(ws, TUI_FORCE_REDRAW) return True @@ -175,13 +286,30 @@ class PtySession: return leader == pid or leader in _process_ancestors(pid) async def close(self) -> None: - self.alive = False + # Disconnect cancellation must not strand a child after its registry + # entry was removed. Cleanup has independent, idempotent ownership. + if self._close_task is None: + self.alive = False + self._attach_generation += 1 + if self._input_task is not None: + self._input_task.cancel() + ws = self._ws + self.detach(ws) + self._close_task = asyncio.create_task(self._close(ws)) + await asyncio.shield(self._close_task) + + async def _close(self, ws) -> None: + await _close_ws(ws, WS_CLOSE_PROCESS_EXITED) if self._drain_task is not None: self._drain_task.cancel() try: await self._drain_task - except (asyncio.CancelledError, Exception): + except asyncio.CancelledError: pass + except Exception: + # The drain handles its own read and viewer errors, so this is a bug; the + # child below must still be closed. + logger.warning("PTY output drain failed", exc_info=True) try: # bridge.close() joins the child — blocking; keep it off the event loop. # See #53227. @@ -219,52 +347,96 @@ async def run_reaper(registry: PtySessionRegistry, *, interval: float = 60.0) -> class PtySessionRegistry: - def __init__(self, *, ttl: float, max_sessions: int, buffer_cap: int, read_timeout: float) -> None: + def __init__( + self, *, ttl: float, max_sessions: int, buffer_cap: int, read_timeout: float, + drain_executor: Optional[Executor] = None, + ) -> None: self._ttl = ttl self._max = max_sessions self._buffer_cap = buffer_cap self._read_timeout = read_timeout + self._drain_executor = drain_executor self._sessions: dict[str, PtySession] = {} - # The get-or-spawn decision spans awaits (reap_idle, the spawn thread, - # session.start), so two connections racing one attach token both saw - # "no session" and forked a PTY each: the token then mapped to whichever - # registered last while the other tab's live session fell out of the - # registry — never reaped, and a reattach landed on the wrong terminal - # (#115304). Serialize the decision so a token maps to one PTY. - # ponytail: one registry-wide lock, not per key — argv resolution is - # already serialized globally for the same reason, and a spawn only - # delays NEW chats. Per-key locks if spawn throughput ever matters. - self._attach_lock = asyncio.Lock() - # Sessions popped from the registry but still closing in the background; close_all() - # awaits them too, and holding the tasks keeps them from being garbage-collected. + # One registry-wide reservation spans lookup, spawn, and registration: + # racing connections with one attach token must share one tracked PTY. + self._spawn_lock = asyncio.Lock() + self._spawn_cleanup: set[asyncio.Task] = set() + self._closed = False + # Retain ownership of removed sessions until helper-process shutdown completes. self._background_closes: set[asyncio.Task] = set() - async def attach_or_spawn(self, key: str, *, spawn: Callable[[], object], active_session_file: Optional[Path] = None) -> tuple[PtySession, bool]: + async def attach_or_spawn( + self, key: str, *, spawn: Callable[[], object], active_session_file: Optional[Path] = None, + ) -> tuple[PtySession, bool]: + # Reserve capacity and the key across blocking fork/exec. On cancellation + # the registry, not the request's cancellation scope, owns that reservation + # until the admission finishes and any unclaimed child has been closed. + await self._spawn_lock.acquire() + admission = asyncio.create_task( + self._attach_or_spawn(key, spawn=spawn, active_session_file=active_session_file)) + try: + # wait() leaves admission running on caller cancellation without + # shield() reporting its late exception (Python 3.14). The registry's + # discard task owns that result once the caller disconnects. + await asyncio.wait({admission}) + result = admission.result() + except asyncio.CancelledError: + cleanup = asyncio.create_task(self._discard_admission(admission)) + self._spawn_cleanup.add(cleanup) + cleanup.add_done_callback(self._spawn_cleanup.discard) + raise + except BaseException: + self._spawn_lock.release() + raise + self._spawn_lock.release() + return result + + async def _discard_admission(self, admission: asyncio.Task) -> None: + try: + try: + session, created = await admission + except Exception: + # The disconnected caller cannot observe a late fork/exec + # failure: retrieve it here and keep the traceback. + logger.debug("PTY spawn failed after its caller disconnected", exc_info=True) + return + if created: + if self._sessions.get(session.key) is session: + self._sessions.pop(session.key) + await session.close() + finally: + self._spawn_lock.release() + + async def _attach_or_spawn( + self, key: str, *, spawn: Callable[[], object], active_session_file: Optional[Path] = None, + ) -> tuple[PtySession, bool]: + if self._closed: + raise RegistryFull("Terminal service is shutting down.") await self.reap_idle() - async with self._attach_lock: - existing = self._sessions.get(key) - if existing is not None and existing.alive: - return existing, False - if existing is not None: # dead remnant - # Close in the background: ending a dead leader's helpers can take the helper - # grace, and this lock serializes every new chat. - self._sessions.pop(key, None) - self._close_in_background(existing) - if len(self._sessions) >= self._max: - self._reap_one_idle_or_raise() - # PTY spawn does blocking fork/exec work — keep it off the event loop. - # See #53227. - bridge = await asyncio.to_thread(spawn) - session = PtySession( - key, - bridge, - buffer_cap=self._buffer_cap, - read_timeout=self._read_timeout, - active_session_file=active_session_file, - ) - await session.start() - self._sessions[key] = session - return session, True + existing = self._sessions.get(key) + if existing is not None and existing.alive: + return existing, False + if existing is not None: # dead remnant + # Close in the background: ending a dead leader's helpers can take the helper + # grace, and this reservation serializes every new chat. + self._sessions.pop(key, None) + self._close_in_background(existing) + if len(self._sessions) >= self._max: + self._reap_one_idle_or_raise() + # PTY spawn does blocking fork/exec work — keep it off the event loop. + # See #53227. + bridge = await asyncio.to_thread(spawn) + session = PtySession( + key, + bridge, + buffer_cap=self._buffer_cap, + read_timeout=self._read_timeout, + active_session_file=active_session_file, + drain_executor=self._drain_executor, + ) + await session.start() + self._sessions[key] = session + return session, True async def close_other_sessions(self, prefix: str, *, keep_key: str) -> None: """Close sessions belonging to the same logical client except ``keep_key``. @@ -274,7 +446,7 @@ class PtySessionRegistry: remain alive long enough to hold the TUI session lease and reject a later return to that chat. """ - async with self._attach_lock: + async with self._spawn_lock: keys = [ key for key in self._sessions if key != keep_key and (key == prefix or key.startswith(prefix + "\0")) @@ -284,8 +456,11 @@ class PtySessionRegistry: if session is not None: # A sibling tab sharing the attach token may still be viewing this # PTY: supersede it explicitly (4409) instead of leaving it silent - # until its next keystroke fails with 1013. - await _close_ws(session._ws, WS_CLOSE_SUPERSEDED) + # until its next keystroke fails with 1013. Detach first so close() + # does not overwrite that code with WS_CLOSE_PROCESS_EXITED. + ws = session._ws + session.detach(ws) + await _close_ws(ws, WS_CLOSE_SUPERSEDED) await session.close() async def close_orphaned_sessions( @@ -308,7 +483,7 @@ class PtySessionRegistry: # The requested chat is (profile, session): the same session id in # another profile's store is a different chat whose terminal stays. target = (_key_segments(keep_key)[0], resume) - async with self._attach_lock: + async with self._spawn_lock: doomed = [ key for key, session in self._sessions.items() if key != keep_key and not session.attached @@ -326,20 +501,26 @@ class PtySessionRegistry: if s is not None: s.detach(ws) + def expired(self, s: PtySession, now: Optional[float] = None) -> bool: + """Whether ``s`` has had no viewer for longer than the retention TTL.""" + return (not s.attached and s.last_detached_at is not None + and (time.monotonic() if now is None else now) - s.last_detached_at > self._ttl) + + def _reapable(self, s: PtySession, now: Optional[float] = None) -> bool: + # EOF never arrives if a helper still holds the PTY slave after the child died (#76759); + # ask the process itself (a WNOHANG waitpid). + return not s.alive or not s.bridge.is_alive() or self.expired(s, now) + async def reap_idle(self, now: Optional[float] = None) -> None: now = time.monotonic() if now is None else now - doomed = [ - key for key, s in self._sessions.items() - if not s.alive - or (not s.attached and s.last_detached_at is not None and (now - s.last_detached_at) > self._ttl) - # EOF never arrives if a helper still holds the PTY slave after the child died (#76759); - # ask the process itself (a WNOHANG waitpid). - or not s.bridge.is_alive() - ] - for key in doomed: + doomed = [(key, s) for key, s in self._sessions.items() if self._reapable(s, now)] + for key, expected in doomed: # Reaps overlap (attach_or_spawn and the background reaper) and close() # awaits, so a concurrent reap can have popped this key already — skip - # it instead of raising KeyError into the websocket handler. + # it instead of raising KeyError into the websocket handler. A viewer can + # also re-attach during an earlier close, so re-check before closing. + if self._sessions.get(key) is not expected or not self._reapable(expected, now): + continue session = self._sessions.pop(key, None) if session is not None: await session.close() @@ -352,13 +533,19 @@ class PtySessionRegistry: self._sessions.pop(oldest.key, None) self._close_in_background(oldest) - def _close_in_background(self, session: PtySession) -> None: + def _close_in_background(self, session: PtySession) -> asyncio.Task: task = asyncio.create_task(session.close()) self._background_closes.add(task) task.add_done_callback(self._background_closes.discard) + return task async def close_all(self) -> None: - # Close concurrently: each close() may wait out its helpers' SIGHUP grace, and shutdown runs - # under the backend's SIGTERM -> SIGKILL budget (dashboard_procs._POSIX_TERM_GRACE_SECONDS). + self._closed = True + # Wait for any admitted spawn before snapshotting shutdown ownership. + async with self._spawn_lock: + pass + if self._spawn_cleanup: + await asyncio.shield(asyncio.gather(*self._spawn_cleanup)) + # Helper-process grace must run concurrently inside the backend shutdown budget. sessions = [self._sessions.pop(key) for key in list(self._sessions)] await asyncio.gather(*(s.close() for s in sessions), *self._background_closes, return_exceptions=True) diff --git a/hermes_cli/sessions_adopt.py b/hermes_cli/sessions_adopt.py new file mode 100644 index 0000000000..325e0f385d --- /dev/null +++ b/hermes_cli/sessions_adopt.py @@ -0,0 +1,135 @@ +"""``hermes sessions adopt``: bring this profile's pre-gateway history under its gateway, as the host. + +Since #106742 the profile's gateway owns its sessions. Chats written before that (main's CLI, TUI and +Desktop rows) carry no gateway receipt until one native, same-OS-user client resumes them, which +adopts them (``gateway/session_local_migration.py::adopt_legacy_session``). A Webapp browser reaches +the gateway through its dashboard mount as the login it authenticated, never natively, so it may not +make that first claim. Run this once on the host instead: it is what opening each old chat in the +native TUI or Desktop would do, and afterwards the browser continues every adopted chat through +shared operator access. +""" +from __future__ import annotations + +from collections import Counter +import sys + +_PROGRESS_EVERY = 25 + + +def adoptable_sessions(path) -> list[dict]: + """This store's unadopted pre-gateway local chats: one compression tip each, oldest first. + + The same eligibility the owner applies on resume (``validate_legacy_row``), read-only, so the + count a dry run reports is what an adoption run attempts. A row the owner already bound is not + offered again, which makes a second run a no-op.""" + from hermes_state import SessionDB + from hermes_state_local import POLICY_PREFIX + from hermes_state_local_migration import LEGACY_LOCAL_SOURCES, LEGACY_PREFIX, validate_legacy_row + from hermes_state_runtime import RuntimeStoreError + + db = SessionDB(db_path=path, read_only=True) + try: + sources = tuple(sorted(LEGACY_LOCAL_SOURCES)) + with db._read_ctx() as conn: + ids = [row[0] for row in conn.execute( + f"SELECT id FROM sessions WHERE source IN ({','.join('?' * len(sources))}) ORDER BY started_at, id", + sources)] + bound = {row[0] for row in conn.execute( + 'SELECT key FROM state_meta WHERE key LIKE ? OR key LIKE ?', (POLICY_PREFIX + '%', LEGACY_PREFIX + '%'))} + found, seen = [], set() + for sid in ids: + tip_id = db.get_compression_tip(sid) or sid + if tip_id in seen: + continue + seen.add(tip_id) + tip = db.get_session(tip_id) + key = tip and (tip.get('chat_id') or tip_id) + if tip is None or POLICY_PREFIX + key in bound or LEGACY_PREFIX + key in bound: + continue + try: + validate_legacy_row(db, tip) + except RuntimeStoreError: + continue + found.append(tip) + return found + finally: + db.close() + + +async def adopt_sessions(rows, *, report=print) -> tuple[int, Counter]: + """Resume each row once over this host's native connection (the owner adopts it), then detach + without ending it. ``(adopted, skipped-by-refusal)``; a row the owner refuses is skipped, + never retried as another principal.""" + from hermes_cli.gateway_client import GatewayRPCError, connect_gateway + from hermes_cli.web_server_boundary import _DESKTOP_ATTACHMENT_WS_MAX_BYTES + + adopted, skipped = 0, Counter() + # The owner answers a resume with the session's whole history; a long chat outgrows the default frame. + async with connect_gateway(max_size=_DESKTOP_ATTACHMENT_WS_MAX_BYTES) as client: + for done, row in enumerate(rows, 1): + try: + snapshot = await client.rpc('session.resume', session_id=row['id']) + except GatewayRPCError as exc: + skipped[_refusal(exc)] += 1 + continue + await client.rpc('session.detach', session_id=snapshot.get('session_id') or row['id'], + subscription_id=snapshot['subscription_id']) + adopted += 1 + if done % _PROGRESS_EVERY == 0: + report(f' {done}/{len(rows)} …') + return adopted, skipped + + +_SKIP_HINTS = { + 'launch_credentials_unavailable': 'its saved model route is gone; open it once and pick a model', + 'not_found': 'not adoptable here (another identity or profile owns it)', +} + + +def _refusal(exc) -> str: + """One owner refusal as the report prints it: the reason code, then why. A field-level refusal + (a launch directory that no longer exists) names the field and the owner's bounded detail.""" + reason, field, detail = str(exc), exc.data.get('field'), exc.data.get('detail') + why = _SKIP_HINTS.get(reason) or (f'{field}: {detail}' if field and detail else 'refused by the gateway') + return f'{reason} ({why})' + + +def cmd_adopt(args) -> int: + import asyncio + from websockets.exceptions import WebSocketException + from hermes_cli.gateway_client import GatewayClientError + from hermes_constants import display_hermes_home, get_hermes_home + + path = get_hermes_home() / 'state.db' + if not path.exists(): + print(f'No session database at {display_hermes_home()}/state.db (nothing to adopt).') + return 0 + rows = adoptable_sessions(path) + if not rows: + print('Nothing to adopt: every local chat in this profile already belongs to its gateway.') + return 0 + counts = Counter(row['source'] for row in rows) + print(f"{len(rows)} pre-gateway chat(s) to adopt ({', '.join(f'{n} {s}' for s, n in sorted(counts.items()))}).") + if args.dry_run: + print('Dry run: nothing was adopted.') + return 0 + try: + adopted, skipped = asyncio.run(adopt_sessions(rows)) + except (GatewayClientError, OSError, TimeoutError, WebSocketException) as exc: + message = str(exc) if isinstance(exc, GatewayClientError) else 'Gateway connection/read failed' + print('Error: ' + message, file=sys.stderr) + return 1 + print(f'Adopted {adopted} chat(s).') + for refusal, count in sorted(skipped.items()): + print(f'Skipped {count}: {refusal}.') + return 0 + + +def add_adopt_parser(sessions_subparsers) -> None: + parser = sessions_subparsers.add_parser( + 'adopt', help="Bring this profile's pre-gateway chats under its gateway (run on the host)", + description="Chats written before the gateway owned sessions become gateway sessions the first " + "time a native client on this host opens them. This adopts them all at once, so remote " + "clients such as the Webapp can continue them. Run it as the gateway's OS user; use " + "`hermes -p sessions adopt` for another profile.") + parser.add_argument('--dry-run', action='store_true', help='Only count what would be adopted') diff --git a/hermes_cli/sessions_cmd.py b/hermes_cli/sessions_cmd.py index ef29f42915..adbf6dd77e 100644 --- a/hermes_cli/sessions_cmd.py +++ b/hermes_cli/sessions_cmd.py @@ -1254,11 +1254,17 @@ def _cmd_discard(args): return cmd_discard(args) +def _cmd_adopt(args): + from hermes_cli.sessions_adopt import cmd_adopt + return cmd_adopt(args) + + _PRE_DB_HANDLERS = { "repair": _cmd_repair, "recover": _cmd_recover, "import": _cmd_import, "repair-profiles": _cmd_repair_profiles, # opens every profile's store itself "set-journal-mode": _cmd_set_journal_mode, # offline: must not open the store it converts "discard": _cmd_discard, # a gateway control (prompt.resolve_unknown), never a direct store write + "adopt": _cmd_adopt, # native resumes on the owner; reads its own store read-only } _OBSERVATIONAL_DB_ACTIONS = frozenset({"list", "stats", "pinned"}) _DB_HANDLERS = { diff --git a/hermes_cli/source_build.py b/hermes_cli/source_build.py index c5c2ee3aa0..901dcebdbd 100644 --- a/hermes_cli/source_build.py +++ b/hermes_cli/source_build.py @@ -73,13 +73,17 @@ def run_source_script(project_root: Path, script: str, *args: str, env: dict, la indent=" ", cwd=project_root, env=env) -def prepare_source_dependencies(project_root: Path, workspaces: tuple[str, ...], *, env: dict, - explicit: bool = False) -> None: +def _install_args(explicit: bool) -> tuple[str, ...]: from pm import lazy_installs_allowed + return () if explicit or lazy_installs_allowed() else ("--no-install",) + + +def prepare_source_dependencies(project_root: Path, workspaces: tuple[str, ...], *, env: dict, + explicit: bool = False) -> None: run_source_script( project_root, "scripts/build/node-deps.mjs", "--source", str(project_root), "--reuse", - *(() if explicit or lazy_installs_allowed() else ("--no-install",)), + *_install_args(explicit), *(arg for workspace in workspaces for arg in ("--workspace", workspace)), env=env, label="Preparing Node dependencies", ) @@ -112,6 +116,12 @@ def build_source_web(project_root: Path, *, env: dict, icons: Path | None = None label="Building the web UI") +def build_source_webapp(project_root: Path, *, env: dict, explicit: bool = False) -> None: + """Browser renderer; its dependencies live in a private workspace, never the checkout's.""" + run_source_script(project_root, "apps/desktop/scripts/build-webapp.mjs", *_install_args(explicit), + env=env, label="Building the Webapp renderer") + + def source_frontends(project_root: Path) -> tuple[str, ...]: """The frontend workspaces this checkout carries. A source slice without them (python-only installs, the installer's acceptance fixture) has no products diff --git a/hermes_cli/staged_uploads.py b/hermes_cli/staged_uploads.py new file mode 100644 index 0000000000..dd3e3f7ad9 --- /dev/null +++ b/hermes_cli/staged_uploads.py @@ -0,0 +1,214 @@ +"""Browser uploads staged in a profile home, and the receipt ``file.attach`` redeems. + +``POST /api/chat/file-upload`` publishes a picked file at ``/uploads/web--`` +and returns a ``staged_upload`` descriptor: that path, the profile home and this installation's id. +The gateway's ``file.attach`` trusts the descriptor only after resolving it back to exactly that +layout, so the writer and the reader of the layout live here together. + +A named profile that is missing or deleted is refused on both sides by the rule every named-home +write follows (``hermes_constants.assert_named_profile_home_live`` / ``mkdir_under_hermes_home``): +an upload never recreates a deleted home, and a receipt from one is never redeemed. +""" + +from __future__ import annotations + +from contextlib import contextmanager, suppress +import os +from pathlib import Path +import re +import secrets +import stat +import time + +from fastapi import HTTPException + +from hermes_constants import WEBAPP_ATTACHMENT_MAX_BYTES +from hermes_cli.install_identity import get_install_id +from hermes_cli.web_deps import late + + +_is_current_profile = late("_is_current_profile", "hermes_cli.web_server_profiles") +_resolve_profile_dir = late("_resolve_profile_dir", "hermes_cli.web_server_profiles") +_UPLOADS_DIR = "uploads" +_STAGED_PREFIX = "web-" +_CHUNK_BYTES = 1024 * 1024 +_UPLOAD_RETENTION_SECONDS = 7 * 24 * 60 * 60 +_SAFE_FILENAME = re.compile(r"[^A-Za-z0-9._-]+") + + +def _safe_filename(value: str | None) -> str: + name = Path(str(value or "attachment")) + # Sanitizing the whole name would turn an all-non-ASCII stem into a + # leading dot, then strip the extension separator needed by read_file. + stem = _SAFE_FILENAME.sub("-", name.stem).strip(".-") or "attachment" + suffix = _SAFE_FILENAME.sub("-", name.suffix).rstrip(".-")[:119] + return stem[:120 - len(suffix)] + suffix + + +def _prune_stale_uploads(root: Path, *, now: float | None = None) -> None: + """Bound abandoned browser-picker staging without following symlinks.""" + cutoff = (time.time() if now is None else now) - _UPLOAD_RETENTION_SECONDS + try: + entries = list(root.iterdir()) + except OSError: + return + for entry in entries: + if not entry.name.startswith(_STAGED_PREFIX): + continue + try: + metadata = entry.stat(follow_symlinks=False) + if stat.S_ISREG(metadata.st_mode) and metadata.st_mtime < cutoff: + entry.unlink() + except OSError: + continue + + +@contextmanager +def removed_on_failure(target: Path): + """Unlink a partially published file; a cleanup error never masks the cause.""" + try: + yield + except BaseException: + with suppress(OSError): + target.unlink(missing_ok=True) + raise + + +def resolve_upload_home(profile: str | None) -> Path: + """Resolve the profile home an upload is published into, with ``_profile_scope``'s 400/404s. + + Without entering that scope: nothing here reads config or secrets, so an upload must not queue + on its process-wide skills lock or wait on its secret-source hydration. A named home that is + missing or deleted is a 404. + """ + from hermes_constants import assert_named_profile_home_live, get_hermes_home + + if _is_current_profile(profile): + home = get_hermes_home() + else: + home = _resolve_profile_dir(profile.strip()) + try: + assert_named_profile_home_live(home) + except FileNotFoundError as exc: + raise HTTPException(status_code=404, detail="Profile home is unavailable") from exc + return home + + +def ensure_profile_subdir( + home: Path, + name: str, + *, + denied: str, + failed: str, + mode: int | None = None, +) -> Path: + """``/``, created without ever recreating a missing or deleted named home. + + That case is a 404 (``mkdir_under_hermes_home``). Other directory errors are 403 *denied* and + 500 ``": "``; *mode* is applied inside that mapping. + """ + from hermes_constants import mkdir_under_hermes_home + + directory = home / name + try: + mkdir_under_hermes_home(directory) + if mode is not None: + directory.chmod(mode) + except FileNotFoundError as exc: + raise HTTPException(status_code=404, detail="Profile was deleted during upload") from exc + except PermissionError as exc: + raise HTTPException(status_code=403, detail=denied) from exc + except OSError as exc: + raise HTTPException(status_code=500, detail=f"{failed}: {exc}") from exc + return directory + + +def publish_staged_upload(staged, profile_home: Path, filename: str) -> Path: + """Publish staged bytes as a new owner-only file under ``/uploads``.""" + from hermes_constants import assert_named_profile_home_live + + upload_root = ensure_profile_subdir( + profile_home, + _UPLOADS_DIR, + denied="Upload directory is not writable", + failed="Could not create upload directory", + mode=0o700, + ) + _prune_stale_uploads(upload_root) + target = upload_root / f"{_STAGED_PREFIX}{secrets.token_hex(8)}-{_safe_filename(filename)}" + try: + with removed_on_failure(target): + fd = os.open( + target, + os.O_WRONLY | os.O_CREAT | os.O_EXCL, + 0o600, + ) + with os.fdopen(fd, "wb") as handle: + staged.seek(0) + while chunk := staged.read(_CHUNK_BYTES): + handle.write(chunk) + handle.flush() + os.fsync(handle.fileno()) + # A deletion recorded while the bytes were written wins: never report a path in it. + assert_named_profile_home_live(profile_home) + return target + except FileNotFoundError as exc: + raise HTTPException(status_code=404, detail="Profile was deleted during upload") from exc + except OSError as exc: + raise HTTPException(status_code=500, detail=f"Could not stage file: {exc}") from exc + + +def staged_upload_descriptor(path: Path, profile_home: Path) -> dict | None: + """The ``staged_upload`` receipt for a published file, or None without an install identity.""" + install_id = get_install_id() + if not install_id: + # Identity is already persisted by Hermes. If it is unavailable, keep + # the legacy path/byte-upload flow rather than invent an ephemeral id. + return None + return { + "install_id": install_id, + "path": str(path), + "profile_home": str(profile_home), + } + + +def parse_staged_upload(descriptor: object) -> tuple[str, str]: + """``(profile_home, path)`` of a descriptor this installation issued. + + Checked before the source is touched. A replacement backend at the same URL has a different + install id, and the source bytes alone never authorize reuse. + """ + if not isinstance(descriptor, dict): + raise ValueError("invalid staged upload") + install_id = get_install_id() + if not install_id or descriptor.get("install_id") != install_id: + raise ValueError("Staged attachment belongs to another Hermes backend; select the file again") + raw_home = descriptor.get("profile_home") + raw_path = descriptor.get("path") + if not isinstance(raw_home, str) or not raw_home or not isinstance(raw_path, str) or not raw_path: + raise ValueError("invalid staged upload source") + return raw_home, raw_path + + +def resolve_staged_upload(home: Path, raw_path: str) -> Path: + """The staged regular file *raw_path* names under *home*, within the browser size cap. + + The check is anchored on *home*, never on where the descriptor's path claims to point. A named + *home* that is missing or deleted raises ``FileNotFoundError``: its receipts are void. + """ + from hermes_constants import assert_named_profile_home_live + + assert_named_profile_home_live(home) + source = Path(raw_path) + if source.is_symlink(): + raise ValueError("staged upload is no longer a regular file") + source = source.resolve(strict=True) + if ( + source.parent != (home / _UPLOADS_DIR).resolve(strict=True) + or not source.name.startswith(_STAGED_PREFIX) + or not source.is_file() + ): + raise ValueError("staged upload is outside its source profile") + if source.stat().st_size > WEBAPP_ATTACHMENT_MAX_BYTES: + raise ValueError("staged upload exceeds the browser attachment size limit") + return source diff --git a/hermes_cli/subcommands/sessions.py b/hermes_cli/subcommands/sessions.py index e47a3bbf7b..a838e1127a 100644 --- a/hermes_cli/subcommands/sessions.py +++ b/hermes_cli/subcommands/sessions.py @@ -26,6 +26,8 @@ def _add_rename_and_discard(sessions_subparsers) -> None: sessions_rename.add_argument("title", nargs="+", help="New title for the session") from hermes_cli.sessions_discard import add_discard_parser add_discard_parser(sessions_subparsers) + from hermes_cli.sessions_adopt import add_adopt_parser + add_adopt_parser(sessions_subparsers) def build_sessions_parser(subparsers, *, cmd_sessions: Callable) -> None: diff --git a/hermes_cli/update_cmd_zip.py b/hermes_cli/update_cmd_zip.py index 6c8a8c451d..47c4079233 100644 --- a/hermes_cli/update_cmd_zip.py +++ b/hermes_cli/update_cmd_zip.py @@ -40,7 +40,7 @@ _ZIP_PRESERVED_TOP_LEVEL = {"venv", ".venv", "node_modules", ".git", ".env"} # its own node_modules (electron itself), and the dashboard assets. The dirty-tree guard admits them and # `_stage_entries` grafts the live copies into the staged tree so the swap keeps them (#90495). _ZIP_PRESERVED_NESTED = { - "apps": ("desktop/release", "desktop/dist", "desktop/node_modules", "desktop/build"), + "apps": ("desktop/release", "desktop/dist", "desktop/dist-webapp", "desktop/node_modules", "desktop/build"), "hermes_cli": ("web_dist",), "scripts": ("whatsapp-bridge/node_modules",), "ui-tui": ("dist", "node_modules", "packages/hermes-ink/dist"), diff --git a/hermes_cli/web_host_terminal.py b/hermes_cli/web_host_terminal.py new file mode 100644 index 0000000000..2750945384 --- /dev/null +++ b/hermes_cli/web_host_terminal.py @@ -0,0 +1,161 @@ +"""Host-terminal policy and process setup for the browser-hosted Desktop UI.""" + +from __future__ import annotations + +import os +from collections.abc import Mapping +from pathlib import Path +from typing import Optional + +META_PREFIX = "\0HERMES_TERMINAL_META:" + + +def request_allowed() -> bool: + """Allow host shells only for Webapp on loopback or behind authentication.""" + from hermes_cli.web_server import app + from hermes_cli.web_server_chat import _LOOPBACK_HOSTS + from hermes_cli.web_server_surface import policy + + if not policy(app.state).host_terminal: + return False + if getattr(app.state, "auth_required", False): + return True + return (getattr(app.state, "bound_host", "") or "").strip().lower() in _LOOPBACK_HOSTS + + +def shell_command(candidate: str) -> Optional[str]: + """Return an executable shell path for ``candidate``, or None. + + A relative path is refused: the PTY chdirs into the workspace before exec, + so it would name a different file than the one checked here. + """ + from hermes_platform.resolver import locate_command + + raw = (candidate or "").strip() + found = locate_command(raw).command if raw else () + return found[0] if found else None + + +def shell_spec() -> tuple[list[str], str]: + """Resolve the interactive-shell ladder the native Desktop uses (electron/terminal-ipc.ts). + + POSIX honours ``$SHELL``, then /bin/zsh, /bin/bash, /bin/sh. Windows ignores ``$SHELL`` + (usually a stray MSYS/Git path) and takes PowerShell 7, Windows PowerShell, ``COMSPEC``, + then cmd.exe. Electron's ``HERMES_DESKTOP_SHELL`` override is not read here: settings do + not ride on ``HERMES_*`` environment variables. + """ + if os.name == "nt": + system_root = os.environ.get("SystemRoot") or os.environ.get("windir") or r"C:\Windows" + windows_powershell = Path(system_root) / "System32" / "WindowsPowerShell" / "v1.0" / "powershell.exe" + ladder = ("pwsh.exe", "pwsh", str(windows_powershell), "powershell.exe", os.environ.get("COMSPEC", "")) + fallback = "cmd.exe" + else: + ladder = (os.environ.get("SHELL", ""), "/bin/zsh", "/bin/bash", "/bin/sh") + fallback = "/bin/sh" + command = next(filter(None, map(shell_command, ladder)), fallback) + + name = Path(command).name.lower() + if name.startswith(("pwsh", "powershell")): + args = ["-NoLogo"] + elif name.startswith("cmd"): + args = [] + elif "zsh" in name or "bash" in name: + args = ["-il"] + else: + args = ["-i"] + return [command, *args], name + + +def safe_cwd(requested: Optional[str]) -> str: + fallback = Path.home() + try: + candidate = Path((requested or "").strip() or fallback).expanduser().resolve() + if candidate.is_dir(): + return str(candidate) + if candidate.is_file(): + return str(candidate.parent) + except (OSError, RuntimeError, ValueError): + pass + return str(fallback) + + +def terminal_lc_ctype(env: Mapping[str, str], platform: str) -> Optional[str]: + """Return the LC_CTYPE a host shell's ``env`` needs added, or None to leave it alone. + + macOS accepts the bare charset name ``UTF-8`` as a locale; glibc rejects it, + and LC_CTYPE outranks LANG, so forcing it on Linux breaks even a valid LANG. + There a set LANG is left in charge (copying it would pin a value a login rc + may still change) and only a missing one falls back to glibc's C.UTF-8. + Windows shells take their code page from the console, not LC_CTYPE. Desktop's + terminalLcCtype (electron/terminal-ipc.ts) shares the darwin and C.UTF-8 choices + but copies LANG into LC_CTYPE and does not defer to LC_ALL. + """ + if env.get("LC_ALL") or env.get("LC_CTYPE") or platform == "win32": + return None + if platform == "darwin": + return "UTF-8" + return None if env.get("LANG") else "C.UTF-8" + + +def resolve_argv( + *, home: Path, requested_cwd: Optional[str] = None, +) -> tuple[list[str], str, dict[str, str], str]: + """Return argv/cwd/env/name for Webapp's authenticated host terminal in ``home``.""" + from gateway.run import _profile_runtime_scope + from hermes_cli import __version__ + from hermes_cli.config import TERMINAL_CONFIG_ENV_MAP + from hermes_constants import ( + get_process_hermes_home, reset_hermes_home_override, set_hermes_home_override, + ) + from hermes_platform.host.facts import os_family + from tools.environments.local import build_subprocess_env, served_profile_child_env + from tools.terminal_scope import enforce_no_refusal, get_terminal_scope + from tui_gateway.launch_profile_policy import ( + activate_multi_profile_hosting, launch_profile_runtime_scope, + ) + + if home.resolve() != get_process_hermes_home().resolve(): + activate_multi_profile_hosting() + scope = _profile_runtime_scope(home) + else: + scope = launch_profile_runtime_scope(home) + # The startup eager-multiplex guard applies even before a secondary's first + # request. Home alone cannot authorize passthrough reads; both shell paths + # need the same complete scope, including the frozen launch environment. + override_token = set_hermes_home_override(str(home)) + try: + with scope: + enforce_no_refusal() + base_env = served_profile_child_env(target_home=home) + for env_var in TERMINAL_CONFIG_ENV_MAP.values(): + base_env.pop(env_var, None) + terminal_scope = get_terminal_scope() + assert terminal_scope is not None # both runtime scopes bind the complete policy + base_env.update(terminal_scope) + env = build_subprocess_env(base=base_env, scrub_secrets=True) + finally: + reset_hermes_home_override(override_token) + + for key in list(env): + if key == "npm_config_prefix" or key.startswith(("npm_config_", "npm_package_")): + env.pop(key, None) + for key in ("NO_COLOR", "FORCE_COLOR", "COLORFGBG"): + env.pop(key, None) + env["COLORTERM"] = "truecolor" + env["TERM"] = "xterm-256color" + env["TERM_PROGRAM"] = "Hermes" + env["TERM_PROGRAM_VERSION"] = __version__ + env["HERMES_DESKTOP_TERMINAL"] = "1" + if lc_ctype := terminal_lc_ctype(env, os_family()): + env["LC_CTYPE"] = lc_ctype + + argv, shell_name = shell_spec() + return argv, safe_cwd(requested_cwd), env, shell_name + + +def query_dimension(raw: Optional[str], default: int, maximum: int) -> int: + try: + value = int(raw or default) + except (TypeError, ValueError, OverflowError): + return default + return max(2, min(maximum, value)) diff --git a/hermes_cli/web_host_terminal_sessions.py b/hermes_cli/web_host_terminal_sessions.py new file mode 100644 index 0000000000..24a3f5c405 --- /dev/null +++ b/hermes_cli/web_host_terminal_sessions.py @@ -0,0 +1,384 @@ +"""Process-lifetime host shells, owned by Webapp identity and profile home. + +This is a separate capability namespace from Dashboard's client-selected chat IDs. +The existing PTY registry owns processes/drains; this module owns host admission. +A shell dies with its profile: once a named home is missing or deleted (main's +tombstone rule, ``hermes_constants.assert_named_profile_home_live``) its input is +refused and the reaper closes it. +""" +from __future__ import annotations + +import asyncio +from concurrent.futures import ThreadPoolExecutor +from contextlib import asynccontextmanager, suppress +from dataclasses import dataclass +import json +from pathlib import Path +import re +import secrets + +from fastapi import HTTPException, WebSocket, WebSocketDisconnect + +from hermes_cli.pty_session import ( + WS_CLOSE_VIEWER_STALLED, WS_CLOSE_VIEWER_STALLED_REASON, PtySession, PtySessionRegistry, + RegistryFull, _close_ws, run_reaper, +) +from hermes_cli.web_host_terminal import META_PREFIX +from hermes_constants import assert_named_profile_home_live, get_hermes_home + +# No survival across backend restart. Disconnects retain a bounded ANSI tail. +RETENTION_SECONDS = 15 * 60 +_MAX_SHELLS = 16 +# Inbound frames buffered while one batch is written (backpressure bound). +_INPUT_BATCH_FRAMES = 64 +TOKEN_RE = re.compile(r"^[A-Za-z0-9_-]{43}$") + + +class TerminalDenied(Exception): + pass + + +class TerminalExpired(Exception): + pass + + +def _home_live(home: Path) -> bool: + """A named home must exist and carry no deletion tombstone; any other home only has to exist.""" + try: + assert_named_profile_home_live(home) + except FileNotFoundError: + return False + return home.is_dir() + + +@dataclass(frozen=True) +class HostOwner: + identity: tuple[str, str] + home: Path + shell: str + cwd: str + + def current(self) -> bool: + """Blocking filesystem reads (home, tombstone): call off-loop.""" + return _home_live(self.home) + + def admit(self, write): + """Run ``write`` while this owner's profile home is live, or raise ``TerminalExpired``. + + Input sent once the profile is deleted never reaches the shell. Blocking: bridges call + it off-loop. + """ + if not self.current(): + raise TerminalExpired() + return write() + + +def _request_identity(ws: WebSocket) -> tuple[str, str]: + identity = getattr(ws, "_hermes_auth_identity", None) + if identity: + user, provider = identity.get("user_id"), identity.get("provider") + if not isinstance(user, str) or not user or not isinstance(provider, str) or not provider: + raise TerminalDenied() + return provider, user + from hermes_cli.web_server import app + if getattr(app.state, "auth_required", False): + raise TerminalDenied() + # _ws_gate already validated the process's loopback credential. + return "loopback", "session-token" + + +def _request_home(profile: str | None) -> Path: + from hermes_cli.web_server_profiles import _resolve_profile_dir + requested = (profile or "").strip() + return ( + _resolve_profile_dir(requested) + if requested and requested.lower() != "current" else get_hermes_home() + ).resolve() + + +class HostTerminalRegistry(PtySessionRegistry): + def __init__(self): + # Every live shell keeps a read in flight. On the loop's default executor, retained + # shells would queue the keystroke fences, ownership checks, uploads and SessionDB + # hops that share it; a pool sized to capacity gives each shell its own reader. + super().__init__( + ttl=RETENTION_SECONDS, max_sessions=_MAX_SHELLS, buffer_cap=1024 * 1024, read_timeout=0.2, + drain_executor=ThreadPoolExecutor(_MAX_SHELLS, thread_name_prefix="host-pty-drain")) + self.owners: dict[str, HostOwner] = {} + + def _reap_one_idle_or_raise(self) -> None: + # Retention is a promise, not an LRU hint; never evict a live shell to + # make room for a different pane, even if its viewer just disconnected. + raise RegistryFull() + + async def create(self, ws: WebSocket, identity: tuple[str, str]): + from hermes_cli.web_host_terminal import query_dimension, resolve_argv + from hermes_cli.web_server_chat import PtyBridge + token = secrets.token_urlsafe(32) + owner = None + + def spawn(): + nonlocal owner + home = _request_home(ws.query_params.get("profile")) + if not _home_live(home): + raise TerminalExpired() + argv, cwd, env, shell = resolve_argv(home=home, requested_cwd=ws.query_params.get("cwd")) + bridge = PtyBridge.spawn( + argv, cwd=cwd, env=env, + cols=query_dimension(ws.query_params.get("cols"), 80, 2000), + rows=query_dimension(ws.query_params.get("rows"), 24, 1000)) + owner = HostOwner(identity, home, shell, cwd) + return bridge + + session, _ = await self.attach_or_spawn(token, spawn=spawn) + assert owner is not None # spawn completed + self.owners[token] = owner + if not await asyncio.to_thread(owner.current): + await self.remove(token) + raise TerminalExpired() + return session, owner + + async def resolve(self, token: str, ws: WebSocket, identity: tuple[str, str], *, closing=False): + session = self._sessions.get(token) + owner = self.owners.get(token) + if session is None or owner is None: + raise TerminalExpired() + if owner.identity != identity: + raise TerminalDenied() + + def validate(): + if _request_home(ws.query_params.get("profile")) != owner.home: + raise TerminalDenied() + owner.admit(lambda: None) + + await asyncio.to_thread(validate) + # No await between the final check and the caller's attach claim. The + # home was just checked off-loop; every input write checks it again. + if (self._sessions.get(token) is not session + or (not closing and (not session.alive or self.expired(session)))): + raise TerminalExpired() + return session, owner + + async def remove(self, token: str): + session = self._sessions.pop(token, None) + self.owners.pop(token, None) + if session is not None: + await asyncio.shield(self._close_in_background(session)) + + async def reap_idle(self, now=None): + # Durable tombstones also detect deletion by a different process; attached + # viewers do not exempt a deleted profile. + # One thread hop per tick keeps every shell's marker reads off the loop. + owners = list(self.owners.items()) + retired = await asyncio.to_thread(lambda: {token for token, owner in owners if not owner.current()}) + for token, owner in owners: + session = self._sessions.get(token) + if session is not None and (token in retired or self._reapable(session, now)): + await self.remove(token) + for token in list(self.owners): + if token not in self._sessions: + self.owners.pop(token, None) + + async def close_all(self): + await super().close_all() + self.owners.clear() + # Every drain is cancelled by now; a read still in flight returns within its + # timeout, so let it finish behind us. A closed registry never spawns again. + self._drain_executor.shutdown(wait=False) + + +def get_host_terminals(app) -> HostTerminalRegistry: + registry = getattr(app.state, "host_terminals", None) + if registry is None or registry._closed: + raise RuntimeError("Host terminal service is not running; start the application lifespan first.") + return registry + + +@asynccontextmanager +async def host_terminal_lifespan(app): + # Overlapping lifespans must not share loop-bound sessions or teardown. + registry = HostTerminalRegistry() + registries = getattr(app.state, "_host_terminal_registries", None) + if registries is None: + registries = app.state._host_terminal_registries = [] + registries.append(registry) + app.state.host_terminals = registry + reaper = asyncio.create_task(run_reaper(registry, interval=1.0)) + try: + yield + finally: + # Withdraw ownership before awaiting cleanup so another teardown cannot + # restore this registry while its reaper/sessions are shutting down. + registries.remove(registry) + if getattr(app.state, "host_terminals", None) is registry: + if registries: + app.state.host_terminals = registries[-1] + else: + del app.state.host_terminals + if not registries: + del app.state._host_terminal_registries + reaper.cancel() + with suppress(asyncio.CancelledError): + await reaper + await registry.close_all() + + +def _metadata(token, owner, session, registry, *, reconnected): + return META_PREFIX + json.dumps({ + "shell": owner.shell, "terminalId": token, "cwd": owner.cwd, + "reconnected": reconnected, "retentionSeconds": registry._ttl, + "truncated": session.buffer.truncated, + }, separators=(",", ":")) + + +async def _send_closed(ws: WebSocket, token: str) -> None: + await ws.send_text(META_PREFIX + json.dumps({"terminalId": token, "closed": True})) + await ws.close(code=1000) + + +async def _admit(ws: WebSocket, registry: HostTerminalRegistry): + """Authorize the request and return ``(session, owner)`` to attach. + + Returns None once a ``close`` request has been served. Raises the ``Terminal*`` + errors ``host_terminal`` maps to close codes. + """ + token = ws.query_params.get("attach") + action = ws.query_params.get("action") + identity = _request_identity(ws) + if action not in (None, "close") or (action == "close" and token is None): + raise TerminalDenied() + if token is not None and not TOKEN_RE.fullmatch(token): + raise TerminalExpired() + if action == "close" and token not in registry._sessions: + # Missing is idempotent, but never bypass auth/host policy. + await _send_closed(ws, token) + return None + if token is None: + return await registry.create(ws, identity) + session, owner = await registry.resolve(token, ws, identity, closing=action == "close") + if action == "close": + await registry.remove(token) + await _send_closed(ws, token) + return None + return session, owner + + +async def _read_frames(ws: WebSocket, inbox: asyncio.Queue) -> None: + """Queue inbound frames. The last item is None, or the error that ended reading + (re-raised by ``_pump_input`` so the route's handlers still see it).""" + end = None + try: + while (msg := await ws.receive()).get("type") != "websocket.disconnect": + await inbox.put(msg) + except RuntimeError: # a superseding/drain task already closed us + pass + # health: allow BLE001 -- forwarded, not swallowed: _pump_input re-raises it, and a reader that died unqueued would hang the pump + except Exception as exc: + end = exc + await inbox.put(end) + + +def _input_runs(frames: list[dict], resize_re): + """``frames`` in order: consecutive keystrokes joined into one run, resizes as ``(cols, rows)``.""" + run = b"" + for msg in frames: + raw = msg.get("bytes") + if raw is None: + raw = (msg.get("text") or "").encode("utf-8") + match = resize_re.fullmatch(raw) if raw else None + if match is None: + run += raw + continue + if run: + yield run + run = b"" + yield int(match[1]), int(match[2]) + if run: + yield run + + +async def _pump_input(ws: WebSocket, registry: HostTerminalRegistry, token: str, + session: PtySession, owner: HostOwner) -> None: + """Forward keystrokes and resizes while ``ws`` is the attached viewer. + + Keystrokes reach the shell only inside ``owner.admit``, so input sent after the + profile is deleted never does, even when it queued behind input that was admitted + first. The fence reads files, so it runs off the loop, once per run of keystrokes in + a batch (the frames that queued while the previous batch was written): a paste of + many frames costs a few fences, not one each. + + Writes have no deadline. While the foreground program is not reading (``sleep``, + ``npm install``) the PTY takes about 1 KB of input on macOS, so a paste waits for + that program, in order, with any Ctrl-C behind it, as in native Desktop's terminal. + A re-attach cancels the wait and keeps the shell; only a closed bridge (the shell + exited or the terminal was closed) or a deleted profile drops the session. ConPTY + buffers input itself, so Windows writes land at once and keep the bridge's deadline. + """ + from hermes_cli.web_server_chat import _RESIZE_RE + inbox: asyncio.Queue = asyncio.Queue(maxsize=_INPUT_BATCH_FRAMES) + reader = asyncio.create_task(_read_frames(ws, inbox)) + try: + # No Ctrl-L/redraw input: arbitrary foreground shell programs own stdin. + while session._ws is ws and session.alive: + batch = [await inbox.get()] + while not inbox.empty(): + batch.append(inbox.get_nowait()) + frames = [msg for msg in batch if isinstance(msg, dict)] + for item in _input_runs(frames, _RESIZE_RE): + if isinstance(item, tuple): + session.resize(ws, cols=item[0], rows=item[1]) + continue + try: + delivered = await session.write(ws, item, fence=owner.admit, timeout=None) + except (TerminalExpired, FileNotFoundError): + await registry.remove(token) + return + if not delivered: + # A closed bridge marked the session dead. A takeover only cancelled this + # write: drop the rest of the batch and leave this socket for the new viewer + # to supersede, so this tab does not reconnect and take the shell back. + if not session.alive: + await registry.remove(token) + return + break + if len(frames) < len(batch): + if isinstance(batch[-1], Exception): + raise batch[-1] + return + finally: + reader.cancel() + with suppress(asyncio.CancelledError): + await reader + + +async def host_terminal(ws: WebSocket) -> None: + """Called only after both the existing WS gate and host policy have passed.""" + from hermes_cli.web_server_chat import _pty_fail + session = None + try: + registry = get_host_terminals(ws.app) + admitted = await _admit(ws, registry) + if admitted is None: + return + session, owner = admitted + token = session.key + initial_text = _metadata( + token, owner, session, registry, reconnected=ws.query_params.get("attach") is not None) + if not await session.attach(ws, initial_text=initial_text): + if session._ws is None: + await _close_ws(ws, WS_CLOSE_VIEWER_STALLED, WS_CLOSE_VIEWER_STALLED_REASON) + return + await _pump_input(ws, registry, token, session, owner) + except TerminalDenied: + await ws.close(code=4403, reason="Terminal belongs to another identity or profile") + except (TerminalExpired, FileNotFoundError, HTTPException): + await ws.close(code=4410, reason="Terminal or profile expired; open a new terminal") + except RegistryFull: + await ws.close(code=1013, reason="Terminal capacity reached; close a terminal and retry") + except WebSocketDisconnect: + pass + except (OSError, RuntimeError) as exc: + await _pty_fail(ws, exc, surface="Terminal") + finally: + if session is not None: + session.detach(ws) diff --git a/hermes_cli/web_models.py b/hermes_cli/web_models.py index 7134d8962c..412c4b58d3 100644 --- a/hermes_cli/web_models.py +++ b/hermes_cli/web_models.py @@ -103,6 +103,13 @@ class ManagedFileDelete(BaseModel): path: str recursive: bool = False +class FileTicketRequest(BaseModel): + """POST /api/files/ticket — the exact query the ticketed download/stream URL carries.""" + route: Literal["download", "stream"] + path: str + profile: Optional[str] = None + session_id: Optional[str] = None + class ModelAssignment(BaseModel): """POST /api/model/set — assign a provider/model to a slot. diff --git a/hermes_cli/web_routers/chat_ws.py b/hermes_cli/web_routers/chat_ws.py index 4125f00efe..13553ddc43 100644 --- a/hermes_cli/web_routers/chat_ws.py +++ b/hermes_cli/web_routers/chat_ws.py @@ -19,10 +19,9 @@ from agent.interrupt_scope import InterruptScope, bind_interrupt_scope from hermes_cli.active_sessions import ActiveSessionRegistryError, active_session_registry_snapshot from hermes_cli.pty_session import RegistryFull from hermes_cli.web_deps import LateState, late -from hermes_cli.web_routers.chat_ws_errors import chat_start_failure_message from hermes_cli.web_server_chat import ( - _build_sidecar_url, _close_stalled_pty_input, _get_console_executor, _legacy_pump, _ws_auth_ok, - _ws_request_is_allowed, + _build_sidecar_url, _close_stalled_pty_input, _get_console_executor, _legacy_pump, _pty_fail, _ws_auth_ok, + _ws_close_reason, _ws_gate, _ws_request_is_allowed, ) _log = logging.getLogger("hermes_cli.web_server") @@ -33,9 +32,6 @@ _active_session_file_for_channel = late("_active_session_file_for_channel", "her _profile_scope = late("_profile_scope", "hermes_cli.web_server_profiles") _resolve_chat_argv_async = late("_resolve_chat_argv_async", "hermes_cli.web_server_chat") _resolve_profile_dir = late("_resolve_profile_dir", "hermes_cli.web_server_profiles") -_ws_auth_reason = late("_ws_auth_reason", "hermes_cli.web_server_chat") -_ws_client_reason = late("_ws_client_reason", "hermes_cli.web_server_chat") -_ws_host_origin_reason = late("_ws_host_origin_reason", "hermes_cli.web_server_chat") _DASHBOARD_EMBEDDED_CHAT_ENABLED = LateState("_DASHBOARD_EMBEDDED_CHAT_ENABLED") @@ -75,18 +71,6 @@ def _discard_active_session_file(app: FastAPI, channel: Optional[str], path: Opt pass -def _ws_auth_mode() -> str: - """Short label for the active WS auth mode — logged on every connection.""" - from hermes_cli.web_server_chat import _LOOPBACK_HOSTS - from hermes_cli.web_server import app - if getattr(app.state, "auth_required", False): - return "gated" - bound_host = (getattr(app.state, "bound_host", "") or "").strip().lower() - if bound_host and bound_host not in _LOOPBACK_HOSTS: - return "insecure" - return "loopback" - - async def _broadcast_event(app: Any, channel: str, payload: str) -> None: """Fan out one publisher frame to every subscriber on `channel`.""" event_channels, event_lock = _get_event_state(app) @@ -114,57 +98,13 @@ def _read_active_session_file(path: Path) -> Optional[str]: return str(data.get("session_id") or "").strip() or None -def _ws_close_reason(text: str) -> str: - """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") + "..." - - -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: +async def _close_unless_sidecar_allowed(ws: WebSocket, *, allow_internal: bool = False) -> 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): + if not _ws_auth_ok(ws, allow_internal=allow_internal): await ws.close(code=4401) return False if not _ws_request_is_allowed(ws): @@ -449,14 +389,6 @@ async def console_ws(ws: WebSocket) -> None: pass -async def _pty_fail(ws: WebSocket, exc: BaseException) -> None: - """Tell the user why chat could not start, then close 1011 so the SPA renders - "Start new session". The raw exception goes to the server log only.""" - _log.warning("pty start failed: %s: %s", type(exc).__name__, exc) - await ws.send_text(f"\r\n\x1b[31m{chat_start_failure_message(exc)}\x1b[0m\r\n") - await ws.close(code=1011) - - def _lease_holder_pid(session_id: Optional[str], *, registry_home: Optional[str] = None) -> Optional[int]: """The live process holding ``session_id``'s single-writer lease, if any. @@ -500,10 +432,13 @@ async def _pump_keepalive_input(ws: WebSocket, session: Any) -> None: # 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))) + session.resize(ws, cols=int(match.group(1)), rows=int(match.group(2))) continue if not await session.write(ws, raw): - await _close_stalled_pty_input(ws, path="keepalive") + # Only a stalled or exited child marks the session dead; a takeover just cancelled + # this write, and the new viewer closes this socket as superseded. + if not session.alive: + await _close_stalled_pty_input(ws, path="keepalive") break @@ -548,13 +483,7 @@ async def pty_ws(ws: WebSocket) -> None: # 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 " - "POSIX PTY, which native Windows Python doesn't provide.\x1b[0m\r\n" - "\x1b[33mInstall Hermes inside WSL2 to use the dashboard's /chat " - "tab — the rest of the dashboard works here.\x1b[0m\r\n" - ) - await ws.close(code=1011) + await _pty_fail(ws, PtyUnavailableError("Pseudo-terminal support is not installed on this host.")) return raw_resume = ws.query_params.get("resume") or None @@ -701,7 +630,7 @@ async def pty_ws(ws: WebSocket) -> None: @router.websocket("/api/ws") async def gateway_ws(ws: WebSocket) -> None: - if not await _close_unless_sidecar_allowed(ws): + if not await _close_unless_sidecar_allowed(ws, allow_internal=True): return from hermes_cli.mcp_startup import start_deferred_mcp_discovery_now from tui_gateway.ws import handle_ws @@ -741,8 +670,8 @@ async def gateway_ws(ws: WebSocket) -> None: # child's stdio handshake with Ink. -async def _accept_channel_ws(ws: WebSocket) -> Optional[str]: - if not await _close_unless_sidecar_allowed(ws): +async def _accept_channel_ws(ws: WebSocket, *, allow_internal: bool = False) -> Optional[str]: + if not await _close_unless_sidecar_allowed(ws, allow_internal=allow_internal): return None channel = _channel_or_close_code(ws) if not channel: @@ -754,7 +683,7 @@ async def _accept_channel_ws(ws: WebSocket) -> Optional[str]: @router.websocket("/api/pub") async def pub_ws(ws: WebSocket) -> None: - channel = await _accept_channel_ws(ws) + channel = await _accept_channel_ws(ws, allow_internal=True) if channel is None: return try: diff --git a/hermes_cli/web_routers/chat_ws_errors.py b/hermes_cli/web_routers/chat_ws_errors.py index 94c5fbcdf9..264ed71d9f 100644 --- a/hermes_cli/web_routers/chat_ws_errors.py +++ b/hermes_cli/web_routers/chat_ws_errors.py @@ -14,15 +14,15 @@ from fastapi import HTTPException from hermes_cli.pty_session import RegistryFull CHAT_NEEDS_NODE = ( - "Chat could not start: Hermes needs Node.js to run the terminal chat. " + "{surface} could not start: Hermes needs Node.js to run the terminal chat. " "Install Node 18+ (for example from nodejs.org) and reopen this tab." ) CHAT_TOO_MANY_TERMINALS = ( - "Chat could not start: too many chat terminals are open in other tabs. " + "{surface} could not start: too many chat terminals are open in other tabs. " "Close one and click Start new session." ) -CHAT_PROFILE_UNKNOWN = "Chat could not start: {detail} Pick another profile from the switcher and reopen this tab." -CHAT_START_FAILED = "Chat could not start: {detail} Check the server log (`hermes dashboard` terminal) and click Start new session." +CHAT_PROFILE_UNKNOWN = "{surface} could not start: {detail} Pick another profile from the switcher and reopen this tab." +CHAT_START_FAILED = "{surface} could not start: {detail} Check the server log (`hermes dashboard` terminal) and click Start new session." def _node_missing(exc: BaseException) -> bool: @@ -30,15 +30,15 @@ def _node_missing(exc: BaseException) -> bool: return isinstance(exc, SystemExit) or "'node'" in text or "'npm'" in text or "Node.js" in text -def chat_start_failure_message(exc: BaseException) -> str: +def chat_start_failure_message(exc: BaseException, *, surface: str = "Chat") -> str: """One plain sentence for the terminal pane; never the bare exception.""" if isinstance(exc, RegistryFull): - return CHAT_TOO_MANY_TERMINALS + return CHAT_TOO_MANY_TERMINALS.format(surface=surface) if isinstance(exc, HTTPException): - return CHAT_PROFILE_UNKNOWN.format(detail=_sentence(str(exc.detail))) + return CHAT_PROFILE_UNKNOWN.format(surface=surface, detail=_sentence(str(exc.detail))) if _node_missing(exc): - return CHAT_NEEDS_NODE - return CHAT_START_FAILED.format(detail=_sentence(str(exc) or type(exc).__name__)) + return CHAT_NEEDS_NODE.format(surface=surface) + return CHAT_START_FAILED.format(surface=surface, detail=_sentence(str(exc) or type(exc).__name__)) def _sentence(text: str) -> str: diff --git a/hermes_cli/web_routers/files.py b/hermes_cli/web_routers/files.py index 5605dd1b93..144b6f6e11 100644 --- a/hermes_cli/web_routers/files.py +++ b/hermes_cli/web_routers/files.py @@ -23,17 +23,20 @@ from pathlib import Path from typing import Any, Dict, Optional from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile -from fastapi.responses import FileResponse, StreamingResponse +from fastapi.responses import FileResponse, JSONResponse, StreamingResponse +from hermes_cli import web_server_file_tickets as file_tickets from hermes_cli._subprocess_compat import windows_hide_flags from hermes_cli.sqlite_safe_read import LiveConnectionError, offline_file_access +from hermes_cli.staged_uploads import ensure_profile_subdir, removed_on_failure, resolve_upload_home from hermes_cli.web_deps import late from hermes_cli.web_server_files import ( SANDBOX_NOT_FOUND, _agent_fs_path, _fs_path, _hosted_fs_read_guard, _is_unmapped_container_path, _managed_file_entry, _managed_response_meta, _resolve_managed_path, ) from hermes_cli.web_models import ( - ChatImageUpload, FsWriteText, ManagedDirectoryCreate, ManagedFileDelete, ManagedFileUpload, + ChatImageUpload, FileTicketRequest, FsWriteText, ManagedDirectoryCreate, ManagedFileDelete, + ManagedFileUpload, ) router = APIRouter() @@ -285,6 +288,35 @@ def _fs_backend(profile: Optional[str] = None): return get_ssh_workspace_fs(profile_key, terminal) +def _chat_image_dir(home: Path) -> Path: + """Where browser image pastes land: the profile's canonical image staging dir on this host + (``gateway.platforms.base.get_image_cache_dir``), the only one attachment admission accepts.""" + from hermes_constants import get_hermes_dir + + return get_hermes_dir("cache/images", "image_cache", home=home) + + +def _is_host_image(path: str, profile: Optional[str]) -> bool: + """Whether *path* is in the profile's image cache, which lives on this host even when the + workspace is an SSH machine: a browser paste is saved there, then read back to be sent.""" + try: + target = Path(path).resolve(strict=True) + except (OSError, RuntimeError): + return False # not on this host: the workspace's to answer + with _profile_scope(profile): + cache = _chat_image_dir(get_hermes_home()) + try: + return target.parent == cache.resolve(strict=True) + except (OSError, RuntimeError): + return False + + +def _remote_fs_backend(path: str, profile: Optional[str]): + """The SSH workspace reader for *path*, or None when this host owns it.""" + backend = _fs_backend(profile) + return None if backend is None or _is_host_image(path, profile) else backend + + def _raise_fs_backend_error(exc: Exception) -> None: from hermes_cli.ssh_workspace_fs import SshWorkspaceFsError @@ -485,18 +517,29 @@ async def upload_chat_image(payload: ChatImageUpload, profile: Optional[str] = N """ def _run(): data, mime_type, ext = _decode_chat_image_upload(payload) - with _profile_scope(profile): - from gateway.platforms.base import get_image_cache_dir - img_dir = get_image_cache_dir() - with _io_errors("Image directory is not writable", "Could not create image directory"): - img_dir.mkdir(parents=True, exist_ok=True) + home = resolve_upload_home(profile) + from hermes_constants import apply_secure_dir_policy, assert_named_profile_home_live - 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}" + img_dir = ensure_profile_subdir( + home, + str(_chat_image_dir(home).relative_to(home)), + denied="Image directory is not writable", + failed="Could not create image directory", + ) + apply_secure_dir_policy(img_dir, home=home) + 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}" + + with removed_on_failure(target): with _io_errors("Image directory is not writable", "Could not write image"): target.write_bytes(data) + try: + # A deletion recorded while the bytes were written wins: never report a path in it. + assert_named_profile_home_live(home) + except FileNotFoundError as exc: + raise HTTPException(status_code=404, detail="Profile was deleted during upload") from exc return { "ok": True, @@ -506,9 +549,8 @@ async def upload_chat_image(payload: ChatImageUpload, profile: Optional[str] = N "mime_type": mime_type, } - # _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. + # The body does file I/O: keep it off the loop. It resolves the profile home without + # entering _profile_scope, so a paste never queues behind its process-wide skills lock. return await asyncio.to_thread(_run) @@ -538,13 +580,17 @@ async def list_managed_files(request: Request, path: Optional[str] = None): return {"path": display_path, "parent": parent, "entries": entries, **_managed_response_meta(policy)} -def _managed_readable_file(request: Request, path: str) -> tuple[Any, Path, str, int, str]: +def _managed_readable_file( + request: Request, path: str, *, create_root: bool = True, +) -> tuple[Any, Path, str, int, str]: """Resolve + guard a managed file for reading: existence, regular file, sensitive-path denylist. Returns (policy, target, display_path, max_bytes, mime_type). Callers own the live-SQLite 409 (held through the read for /api/files/read, point-in-time for streamed responses).""" from hermes_cli.web_server import _MANAGED_FILE_MAX_BYTES - policy, target, display_path = _resolve_managed_path(path, request, for_agent_read=True) + policy, target, display_path = _resolve_managed_path( + path, request, create_root=create_root, for_agent_read=True, + ) if not target.exists(): raise HTTPException(status_code=404, detail="File not found") if not target.is_file(): @@ -589,6 +635,18 @@ async def _managed_file_response( media_only: bool = False, ) -> FileResponse: """Range-aware response after applying managed-file policy.""" + authority = getattr(request.state, "file_ticket_authority", None) + if authority is not None: + from hermes_cli.web_server_file_response import OpenedFileResponse + + return OpenedFileResponse( + path=str(authority.target), + opener=lambda: _open_ticket_file(request, authority, media_only=media_only), + media_type=_fs_mime_type(authority.target), + filename=authority.target.name, + content_disposition_type=content_disposition_type, + headers={"X-Content-Type-Options": "nosniff"} if media_only else None, + ) _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") @@ -603,16 +661,92 @@ async def _managed_file_response( ) +def _open_ticket_file(request: Request, authority: file_tickets.FileAuthority, *, media_only: bool): + """Check the ticket's profile homes are still live and pin the file the ticket names.""" + try: + authority.check_live() + _policy, target, _display, max_bytes, _mime = _managed_readable_file( + request, str(authority.target), create_root=False, + ) + if target != authority.target: + raise HTTPException(status_code=403, detail="File ticket target changed") + if media_only and target.suffix.lower() not in _STREAMABLE_MEDIA_EXTENSIONS: + raise HTTPException(status_code=415, detail="Unsupported media type") + _refuse_live_database(target) + handle = target.open("rb") + try: + opened = os.fstat(handle.fileno()) + if not stat.S_ISREG(opened.st_mode): + raise HTTPException(status_code=400, detail="Only regular files can be read") + if opened.st_size > max_bytes: + raise HTTPException(status_code=413, detail="File is too large") + return handle + except BaseException: + handle.close() + raise + except FileNotFoundError as exc: + raise HTTPException(status_code=404, detail="File ticket profile or resource is unavailable") from exc + except PermissionError as exc: + raise HTTPException(status_code=403, detail="File is not readable") from exc + + +@router.post("/api/files/ticket") +async def issue_file_ticket(payload: FileTicketRequest, request: Request): + """Ticket a browser download/media URL instead of putting the session token in it.""" + from hermes_cli.web_routers.sessions import _history_profile_home + from hermes_constants import named_profile_home + + def owner_home(): + return Path(_history_profile_home(payload.profile)) + + def capture(): + _policy, target, _display, _max, _mime = _managed_readable_file(request, path, create_root=False) + homes = [owner] + target_home = named_profile_home(target) + if target_home is not None and target_home not in homes: + homes.append(target_home) + authority = file_tickets.FileAuthority(target, tuple(homes)) + authority.check_live() + return authority + + try: + # Capture the session/profile owner BEFORE looking up its cwd. The file + # may be outside that home, but the ticket must still die with its owner. + owner = await asyncio.to_thread(owner_home) + path = payload.path + if payload.session_id is not None or path.lower().startswith("file:"): + path = str(await _fs_download_path(path, payload.profile, payload.session_id)) + authority = await asyncio.to_thread(capture) + except FileNotFoundError as exc: + raise HTTPException(status_code=404, detail="File ticket profile is unavailable") from exc + query = {"path": payload.path, "profile": payload.profile, "session_id": payload.session_id} + ticket = file_tickets.mint( + payload.route, {key: value for key, value in query.items() if value is not None}, authority=authority, + ) + return JSONResponse({"ticket": ticket}, headers={"Cache-Control": "no-store"}) + + @router.get("/api/files/download") -async def download_managed_file(request: Request, path: str): +@router.head("/api/files/download") +async def download_managed_file( + request: Request, path: str, profile: Optional[str] = None, session_id: Optional[str] = None, +): """Stream a managed file as an attachment download. - ``auth_middleware`` also accepts the session token as ``?token=`` here so a - shell/browser-opened download (no session header) still authenticates. + ``auth_middleware`` also accepts a path-bound ``?ticket=`` (browsers) or the + session token as ``?token=`` (Electron's remote external link) here so a + download without the session header still authenticates. Chromium marks ``