feat(serve): add secure Desktop SSH bootstrap contract

Accept descriptor-safe one-shot token files and exact owner nonces, expose an authenticated ownership proof endpoint, and preserve the process-local contract across parser and server startup paths.

hermes_cli/main.py

@@ -272,6 +272,7 @@ if _try_termux_ultrafast_version():
 import argparse
 import hashlib
 import json
+import re
 import shlex
 import shutil
 import stat
@@ -12070,8 +12071,92 @@ def _maybe_setup_dashboard_auth_interactively(args) -> None:
     print()
 
 
+def _read_ssh_session_token_file(path: str) -> str:
+    """Read and unlink a Desktop SSH token from its private runtime directory."""
+    import stat as _stat
+    from pathlib import Path as _Path
+
+    if not os.path.isabs(path):
+        raise SystemExit("--ssh-session-token-file must be absolute")
+
+    token_path = _Path(path)
+    token_root = _Path.home() / ".hermes" / "desktop-ssh"
+    try:
+        relative = token_path.relative_to(token_root)
+    except ValueError as exc:
+        raise SystemExit("--ssh-session-token-file must be under ~/.hermes/desktop-ssh") from exc
+    if len(relative.parts) != 2 or not re.fullmatch(r"[0-9a-f]{32}", relative.parts[0]):
+        raise SystemExit("--ssh-session-token-file has an invalid runtime path")
+    if not re.fullmatch(r"[0-9a-f]{16}\.token", relative.parts[1]):
+        raise SystemExit("--ssh-session-token-file has an invalid filename")
+
+    directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
+    file_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    root_fd = -1
+    directory_fd = -1
+    file_fd = -1
+    try:
+        try:
+            root_fd = os.open(token_root, directory_flags)
+            root_stat = os.fstat(root_fd)
+            if not _stat.S_ISDIR(root_stat.st_mode):
+                raise SystemExit("--ssh-session-token-file has an unsafe runtime root")
+            if hasattr(os, "getuid") and root_stat.st_uid != os.getuid():
+                raise SystemExit("--ssh-session-token-file runtime root has the wrong owner")
+            directory_fd = os.open(relative.parts[0], directory_flags, dir_fd=root_fd)
+            directory_stat = os.fstat(directory_fd)
+            if not _stat.S_ISDIR(directory_stat.st_mode):
+                raise SystemExit("--ssh-session-token-file has an unsafe parent directory")
+            if hasattr(os, "getuid") and directory_stat.st_uid != os.getuid():
+                raise SystemExit("--ssh-session-token-file parent has the wrong owner")
+            if (directory_stat.st_mode & 0o777) != 0o700:
+                raise SystemExit("--ssh-session-token-file parent has unsafe permissions")
+            file_fd = os.open(relative.parts[1], file_flags, dir_fd=directory_fd)
+        except SystemExit:
+            raise
+        except OSError as exc:
+            if exc.errno == getattr(__import__("errno"), "ELOOP", -1):
+                raise SystemExit("--ssh-session-token-file is a symlink") from exc
+            raise SystemExit("--ssh-session-token-file is not accessible") from exc
+
+        file_stat = os.fstat(file_fd)
+        if not _stat.S_ISREG(file_stat.st_mode):
+            raise SystemExit("--ssh-session-token-file is not a regular file")
+        if file_stat.st_size != 64:
+            raise SystemExit("--ssh-session-token-file contains an invalid token")
+        if hasattr(os, "getuid") and file_stat.st_uid != os.getuid():
+            raise SystemExit("--ssh-session-token-file has the wrong owner")
+        if hasattr(os, "getuid") and (file_stat.st_mode & 0o777) & ~0o600:
+            raise SystemExit("--ssh-session-token-file has unsafe permissions")
+
+        with os.fdopen(file_fd, "r") as token_stream:
+            file_fd = -1
+            token = token_stream.read(65)
+
+        if not re.fullmatch(r"[0-9a-f]{64}", token):
+            raise SystemExit("--ssh-session-token-file contains an invalid token")
+        return token
+    finally:
+        if file_fd >= 0:
+            os.close(file_fd)
+        if directory_fd >= 0:
+            try:
+                os.unlink(relative.parts[1], dir_fd=directory_fd)
+            except OSError:
+                pass
+            os.close(directory_fd)
+        if root_fd >= 0:
+            os.close(root_fd)
+
+
 def cmd_dashboard(args):
     """Start the web UI server, or (with --stop/--status) manage running ones."""
+    _token_file = getattr(args, "ssh_session_token_file", None)
+    if _token_file and (
+        getattr(args, "status", False) or getattr(args, "stop", False)
+    ):
+        raise SystemExit("--ssh-session-token-file cannot be used with --status or --stop")
+
     # --status: report running dashboards and exit, no deps needed.
     if getattr(args, "status", False):
         count = _report_dashboard_status()
@@ -12094,6 +12179,12 @@ def cmd_dashboard(args):
     # ready sentinel. Resolved once and threaded through the re-exec, the
     # build gate, and start_server.
     _headless_backend = getattr(args, "headless_backend", False)
+    _ssh_owner_nonce = getattr(args, "ssh_owner_nonce", None)
+    if _ssh_owner_nonce and not re.fullmatch(r"[0-9a-f]{16}", _ssh_owner_nonce):
+        raise SystemExit("--ssh-owner-nonce must be 16 lowercase hex characters")
+    _ssh_session_token = None
+    if _token_file and not _headless_backend:
+        raise SystemExit("--ssh-session-token-file is only valid with hermes serve")
 
     # ── Unified profile launch routing ────────────────────────────────
     # The dashboard is a MACHINE management surface: it can read/write any
@@ -12147,6 +12238,10 @@ def cmd_dashboard(args):
             "--host", args.host,
             "--open-profile", _launch_profile,
         ]
+        if _ssh_owner_nonce:
+            reexec_argv.extend(["--ssh-owner-nonce", _ssh_owner_nonce])
+        if _token_file:
+            reexec_argv.extend(["--ssh-session-token-file", _token_file])
         if args.no_open:
             reexec_argv.append("--no-open")
         if getattr(args, "insecure", False):
@@ -12183,6 +12278,9 @@ def cmd_dashboard(args):
         else:
             os.execvpe(sys.executable, reexec_argv, env)
 
+    if _token_file:
+        _ssh_session_token = _read_ssh_session_token_file(_token_file)
+
     # Attach gui.log early so dashboard startup/build failures are captured in
     # the same logs directory as every other Hermes surface.
     try:
@@ -12306,6 +12404,8 @@ def cmd_dashboard(args):
         allow_public=getattr(args, "insecure", False),
         initial_profile=getattr(args, "open_profile", "") or "",
         headless=_headless_backend,
+        ssh_session_token=_ssh_session_token,
+        ssh_owner_nonce=_ssh_owner_nonce,
     )
 
 

hermes_cli/subcommands/dashboard.py

@@ -149,6 +149,20 @@ def build_dashboard_parser(
     serve_parser.add_argument(
         "--no-open", action="store_true", help=argparse.SUPPRESS
     )
+    serve_parser.add_argument(
+        "--ssh-session-token-file",
+        dest="ssh_session_token_file",
+        metavar="PATH",
+        default=None,
+        help="Read a one-shot Desktop SSH session token from PATH",
+    )
+    serve_parser.add_argument(
+        "--ssh-owner-nonce",
+        dest="ssh_owner_nonce",
+        metavar="NONCE",
+        default=None,
+        help="Identify a Desktop-owned SSH backend process",
+    )
     # `headless_backend` marks the lean path: desktop/remote clients speak pure
     # JSON-RPC/WS, so `serve` skips the web UI build AND never serves the SPA
     # (cmd_dashboard exports HERMES_SERVE_HEADLESS=1). `dashboard` leaves it

hermes_cli/web_server.py

@@ -278,6 +278,18 @@ app.include_router(_memory_oauth_router)
 # ---------------------------------------------------------------------------
 _SESSION_TOKEN = os.environ.get("HERMES_DASHBOARD_SESSION_TOKEN") or secrets.token_urlsafe(32)
 _SESSION_HEADER_NAME = "X-Hermes-Session-Token"
+_SSH_OWNER_NONCE: Optional[str] = None
+
+
+def _apply_ssh_session_token(token: str) -> None:
+    global _SESSION_TOKEN
+    if token:
+        _SESSION_TOKEN = token
+
+
+def _apply_ssh_owner_nonce(nonce: Optional[str]) -> None:
+    global _SSH_OWNER_NONCE
+    _SSH_OWNER_NONCE = nonce
 
 # In-browser Chat tab (/chat, /api/pty, /api/ws, …).  Always enabled: the
 # desktop app and the dashboard's own Chat tab both drive the agent over the
@@ -2585,6 +2597,14 @@ def _collect_profile_gateway_topology() -> Dict[str, Any]:
     return {"profiles": profile_names, "gateway_mode": mode, "gateways": gateways}
 
 
+@app.get("/api/ssh/ownership")
+async def get_ssh_ownership(request: Request):
+    _require_token(request)
+    if not _SSH_OWNER_NONCE:
+        raise HTTPException(status_code=404, detail="SSH ownership is not active")
+    return {"ok": True, "sshOwnerNonce": _SSH_OWNER_NONCE, "protocolVersion": 1}
+
+
 @app.get("/api/status")
 async def get_status(profile: Optional[str] = None):
     status_scope = None
@@ -17214,6 +17234,8 @@ def start_server(
     allow_public: bool = False,
     initial_profile: str = "",
     headless: bool = False,
+    ssh_session_token: Optional[str] = None,
+    ssh_owner_nonce: Optional[str] = None,
 ):
     """Start the web UI server.
 
@@ -17225,7 +17247,13 @@ def start_server(
     ``headless`` is the ``serve`` path: the JSON-RPC/WS backend with no UI
     build and no SPA mount (mount_spa() honours ``HERMES_SERVE_HEADLESS``), so
     the banner announces the bind rather than a browser URL.
+
+    ``ssh_session_token`` and ``ssh_owner_nonce`` are process-local Desktop SSH
+    bootstrap state. Neither is persisted or exported to child processes.
     """
+    _apply_ssh_session_token(ssh_session_token or "")
+    _apply_ssh_owner_nonce(ssh_owner_nonce)
+
     import uvicorn
 
     try:

tests/hermes_cli/test_ssh_ownership_endpoint.py

@@ -0,0 +1,38 @@
+from fastapi.testclient import TestClient
+
+from hermes_cli import web_server
+
+
+def test_ssh_ownership_endpoint_requires_token_and_returns_exact_nonce(monkeypatch):
+    token = "t" * 64
+    nonce = "0123456789abcdef"
+    monkeypatch.setattr(web_server, "_SESSION_TOKEN", token)
+    monkeypatch.setattr(web_server, "_SSH_OWNER_NONCE", nonce)
+    web_server.app.state.auth_required = False
+    client = TestClient(web_server.app)
+
+    assert client.get("/api/ssh/ownership").status_code == 401
+    response = client.get(
+        "/api/ssh/ownership",
+        headers={"X-Hermes-Session-Token": token},
+    )
+    assert response.status_code == 200
+    assert response.json() == {
+        "ok": True,
+        "sshOwnerNonce": nonce,
+        "protocolVersion": 1,
+    }
+
+
+def test_ssh_ownership_endpoint_is_absent_without_owner_nonce(monkeypatch):
+    token = "t" * 64
+    monkeypatch.setattr(web_server, "_SESSION_TOKEN", token)
+    monkeypatch.setattr(web_server, "_SSH_OWNER_NONCE", None)
+    web_server.app.state.auth_required = False
+    client = TestClient(web_server.app)
+
+    response = client.get(
+        "/api/ssh/ownership",
+        headers={"X-Hermes-Session-Token": token},
+    )
+    assert response.status_code == 404

tests/hermes_cli/test_ssh_session_token_parser.py

@@ -0,0 +1,86 @@
+import argparse
+import os
+
+import pytest
+
+from hermes_cli.main import _read_ssh_session_token_file, cmd_dashboard
+from hermes_cli.subcommands.dashboard import build_dashboard_parser
+
+
+def dashboard_parser():
+    parser = argparse.ArgumentParser()
+    subparsers = parser.add_subparsers(dest="command")
+    build_dashboard_parser(
+        subparsers,
+        cmd_dashboard=lambda _args: None,
+        cmd_dashboard_register=lambda _args: None,
+    )
+    return parser
+
+
+def test_serve_help_advertises_secure_ssh_bootstrap_flags(capsys):
+    with pytest.raises(SystemExit) as exit_info:
+        dashboard_parser().parse_args(["serve", "--help"])
+    assert exit_info.value.code == 0
+    output = capsys.readouterr().out
+    assert "--ssh-session-token-file PATH" in output
+    assert "--ssh-owner-nonce NONCE" in output
+
+
+def test_serve_accepts_owner_nonce():
+    args = dashboard_parser().parse_args(["serve", "--ssh-owner-nonce", "0123456789abcdef"])
+    assert args.ssh_owner_nonce == "0123456789abcdef"
+
+
+@pytest.mark.parametrize("operation", ["--status", "--stop"])
+def test_one_shot_token_file_rejects_non_starting_operations(operation):
+    args = dashboard_parser().parse_args([
+        "serve", operation, "--ssh-session-token-file", "/tmp/token",
+    ])
+    with pytest.raises(SystemExit, match="cannot be used"):
+        cmd_dashboard(args)
+
+
+def test_token_file_is_read_and_unlinked_through_private_directory(tmp_path, monkeypatch):
+    home = tmp_path / "home"
+    token_dir = home / ".hermes" / "desktop-ssh" / ("a" * 32)
+    token_dir.mkdir(parents=True, mode=0o700)
+    token_path = token_dir / "0123456789abcdef.token"
+    token_path.write_text("b" * 64)
+    token_path.chmod(0o600)
+    monkeypatch.setattr("pathlib.Path.home", lambda: home)
+
+    assert _read_ssh_session_token_file(str(token_path)) == "b" * 64
+    assert not token_path.exists()
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX symlink contract")
+def test_token_file_rejects_symlink(tmp_path, monkeypatch):
+    home = tmp_path / "home"
+    token_dir = home / ".hermes" / "desktop-ssh" / ("a" * 32)
+    token_dir.mkdir(parents=True, mode=0o700)
+    target = tmp_path / "token"
+    target.write_text("b" * 64)
+    target.chmod(0o600)
+    token_path = token_dir / "0123456789abcdef.token"
+    token_path.symlink_to(target)
+    monkeypatch.setattr("pathlib.Path.home", lambda: home)
+
+    with pytest.raises(SystemExit, match="symlink|not accessible"):
+        _read_ssh_session_token_file(str(token_path))
+    assert not token_path.exists()
+    assert target.read_text() == "b" * 64
+
+
+def test_token_file_rejects_parent_escape(tmp_path, monkeypatch):
+    home = tmp_path / "home"
+    token_root = home / ".hermes" / "desktop-ssh"
+    token_root.mkdir(parents=True, mode=0o700)
+    escaped = token_root.parent / "0123456789abcdef.token"
+    escaped.write_text("b" * 64)
+    escaped.chmod(0o600)
+    monkeypatch.setattr("pathlib.Path.home", lambda: home)
+
+    with pytest.raises(SystemExit, match="invalid runtime path"):
+        _read_ssh_session_token_file(str(token_root / ".." / escaped.name))
+    assert escaped.exists()

tests/test_web_server.py

@@ -69,6 +69,22 @@ def _stub_uvicorn(monkeypatch):
     return captured
 
 
+def test_start_server_applies_process_local_ssh_bootstrap_state(monkeypatch):
+    captured = _stub_uvicorn(monkeypatch)
+
+    web_server.start_server(
+        host="127.0.0.1",
+        port=0,
+        open_browser=False,
+        ssh_session_token="s" * 64,
+        ssh_owner_nonce="0123456789abcdef",
+    )
+
+    assert web_server._SESSION_TOKEN == "s" * 64
+    assert web_server._SSH_OWNER_NONCE == "0123456789abcdef"
+    assert captured["port"] == 0
+
+
 def test_start_server_disables_ws_ping_on_loopback(monkeypatch):
     """Loopback binds (the Desktop case) MUST disable uvicorn's protocol-level
     keepalive ping so an event-loop stall can never trigger a false disconnect.