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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 75 additions & 0 deletions hermes_cli/web_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -327,11 +327,47 @@ def _get_pty_active_session_files(app: "FastAPI") -> dict[str, Path]:
# ---------------------------------------------------------------------------


def _session_token_source() -> str:
"""Classify where the dashboard session token came from.

Returns one of:

* ``injected`` — ``HERMES_DASHBOARD_SESSION_TOKEN`` is present AND the
desktop-shell marker ``HERMES_DESKTOP=1`` is set (Electron spawns
``hermes serve`` with both; this is the trusted desktop path). NOTE:
the stale-``.env`` clobber state ALSO lands here — the loader replaces
the injected token with the stale value while the marker survives — so
``injected`` alone does not prove the token the desktop minted is the
one the server adopted (see ``_log_token_mismatch_hint``).
* ``env`` — a token is present but the desktop marker is not. A manual
shell launch (or a non-desktop CLI process) with a leftover exported
``HERMES_DASHBOARD_SESSION_TOKEN``.
* ``generated`` — no token in the environment; a fresh random token was
minted for this server process.

The token value itself is never returned or logged — only this label.
"""
token = os.environ.get("HERMES_DASHBOARD_SESSION_TOKEN")
if not token:
return "generated"
if os.environ.get("HERMES_DESKTOP") == "1":
return "injected"
return "env"


def _resolve_session_token() -> str:
return os.environ.get("HERMES_DASHBOARD_SESSION_TOKEN") or secrets.token_urlsafe(32)


_SESSION_TOKEN = _resolve_session_token()
_SESSION_TOKEN_SOURCE = _session_token_source()
# One provenance line at startup so a stale-token lockout is diagnosable from
# the log alone. Never log the token value — it is a credential.
_log.info(
"dashboard session token: source=%s length=%d",
_SESSION_TOKEN_SOURCE,
len(_SESSION_TOKEN),
)
_SESSION_HEADER_NAME = "X-Hermes-Session-Token"
_SSH_OWNER_NONCE: Optional[str] = None

Expand Down Expand Up @@ -14641,6 +14677,44 @@ def _ws_auth_mode() -> str:
return "loopback"


def _log_token_mismatch_hint(mode: str) -> None:
"""Warn with an actionable fix when a loopback token mismatch is likely
caused by a stale ``HERMES_DASHBOARD_SESSION_TOKEN``.

A leftover value in ``<hermes-home>/.env`` loads into the environment at
import time and clobbers the token the desktop shell injects via
``HERMES_DASHBOARD_SESSION_TOKEN`` + ``HERMES_DESKTOP=1``, so the desktop's
credential never matches ``_SESSION_TOKEN``. Diagnostics only — the
caller still rejects the connection; auth semantics are untouched.
"""
if mode != "loopback":
return
source = _SESSION_TOKEN_SOURCE
if source == "generated":
_log.warning(
"server generated a fresh session token (no "
"HERMES_DASHBOARD_SESSION_TOKEN in the environment) — the client "
"is presenting a stale token; restart the desktop app so it "
"re-injects its token; server token source=generated"
)
return
# source is ``injected`` OR ``env`` — both mean the server resolved its
# token from the environment. A loopback mismatch here IS the lockout
# signature: the desktop injected a fresh token, but a stale
# HERMES_DASHBOARD_SESSION_TOKEN in .env (loaded with override=True)
# replaced it before _SESSION_TOKEN resolved — the marker survives the
# clobber, so ``injected`` cannot be distinguished from the stale state
# by source alone. A genuinely-adopted injection would have matched and
# never reached this point, so the mismatch itself is the trigger.
_log.warning(
"stale HERMES_DASHBOARD_SESSION_TOKEN in %s suspected — remove "
"the line (or run `hermes setup`) and restart the server; "
"server token source=%s",
get_hermes_home() / ".env",
source,
)


def _ws_auth_reason(ws: "WebSocket") -> tuple[Optional[str], str]:
"""Validate WS-upgrade auth; return ``(reason, credential)``.

Expand Down Expand Up @@ -14722,6 +14796,7 @@ def _ws_auth_reason(ws: "WebSocket") -> tuple[Optional[str], str]:
return "no_credential", "none"
if hmac.compare_digest(token.encode(), _SESSION_TOKEN.encode()):
return None, "token"
_log_token_mismatch_hint(_ws_auth_mode())
return "token_mismatch", "token"


Expand Down
131 changes: 131 additions & 0 deletions tests/hermes_cli/test_web_server.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Tests for hermes_cli.web_server and related config utilities."""

import asyncio
import logging
import os
import json
import shutil
Expand Down Expand Up @@ -233,6 +234,136 @@ def test_session_token_resolution_preserves_loaded_app_auth(self, monkeypatch):
assert ws._SESSION_TOKEN == original_token


class TestSessionTokenSource:
"""Provenance classification for _session_token_source()."""

def test_injected_when_desktop_marker_present(self, monkeypatch):
import hermes_cli.web_server as ws

monkeypatch.setenv("HERMES_DASHBOARD_SESSION_TOKEN", "t")
monkeypatch.setenv("HERMES_DESKTOP", "1")
assert ws._session_token_source() == "injected"

def test_env_when_token_without_desktop_marker(self, monkeypatch):
# The stale-~/.hermes/.env clobber case: a token is in the
# environment but the desktop shell marker is not.
import hermes_cli.web_server as ws

monkeypatch.setenv("HERMES_DASHBOARD_SESSION_TOKEN", "stale-from-dotenv")
monkeypatch.delenv("HERMES_DESKTOP", raising=False)
assert ws._session_token_source() == "env"

def test_generated_when_no_token_in_env(self, monkeypatch):
import hermes_cli.web_server as ws

monkeypatch.delenv("HERMES_DASHBOARD_SESSION_TOKEN", raising=False)
monkeypatch.setenv("HERMES_DESKTOP", "1")
assert ws._session_token_source() == "generated"

def test_env_when_token_and_empty_desktop_marker(self, monkeypatch):
import hermes_cli.web_server as ws

monkeypatch.setenv("HERMES_DASHBOARD_SESSION_TOKEN", "t")
monkeypatch.setenv("HERMES_DESKTOP", "")
assert ws._session_token_source() == "env"


class TestTokenMismatchHint:
"""Actionable diagnostics on loopback token_mismatch rejections."""

def _warning_messages(self, caplog):
return [r.message for r in caplog.records if r.levelno >= logging.WARNING]

def test_hint_emitted_for_env_source_loopback(self, monkeypatch, caplog):
import hermes_cli.web_server as ws

monkeypatch.setattr(ws, "_SESSION_TOKEN_SOURCE", "env")
with caplog.at_level(logging.WARNING, logger="hermes_cli.web_server"):
ws._log_token_mismatch_hint("loopback")
assert any(
"stale HERMES_DASHBOARD_SESSION_TOKEN" in m and ".env" in m
for m in self._warning_messages(caplog)
)

def test_hint_emitted_for_generated_source_loopback(self, monkeypatch, caplog):
import hermes_cli.web_server as ws

monkeypatch.setattr(ws, "_SESSION_TOKEN_SOURCE", "generated")
with caplog.at_level(logging.WARNING, logger="hermes_cli.web_server"):
ws._log_token_mismatch_hint("loopback")
assert any(
"fresh session token" in m for m in self._warning_messages(caplog)
)

def test_hint_emitted_for_injected_source_loopback(self, monkeypatch, caplog):
import hermes_cli.web_server as ws

monkeypatch.setattr(ws, "_SESSION_TOKEN_SOURCE", "injected")
with caplog.at_level(logging.WARNING, logger="hermes_cli.web_server"):
ws._log_token_mismatch_hint("loopback")
# A mismatch under the desktop marker IS the lockout signature: the
# stale-.env clobber leaves HERMES_DESKTOP=1 intact while replacing
# the token, so source reads "injected" even though the desktop's
# credential was clobbered. The hint must NOT be suppressed here.
assert any(
"stale HERMES_DASHBOARD_SESSION_TOKEN" in m and ".env" in m
for m in self._warning_messages(caplog)
)

def test_no_hint_outside_loopback_mode(self, monkeypatch, caplog):
import hermes_cli.web_server as ws

monkeypatch.setattr(ws, "_SESSION_TOKEN_SOURCE", "env")
with caplog.at_level(logging.WARNING, logger="hermes_cli.web_server"):
ws._log_token_mismatch_hint("gated")
ws._log_token_mismatch_hint("insecure")
assert self._warning_messages(caplog) == []

def test_ws_auth_reason_emits_hint_on_token_mismatch(self, monkeypatch, caplog):
# End-to-end through the WS auth helper: a bad token in loopback mode
# with an env-sourced server token warns AND still rejects.
import hermes_cli.web_server as ws

fake_ws = SimpleNamespace(
query_params={"token": "wrong-token"},
client=SimpleNamespace(host="127.0.0.1"),
)
monkeypatch.setattr(ws.app.state, "auth_required", False, raising=False)
monkeypatch.setattr(ws.app.state, "bound_host", None, raising=False)
monkeypatch.setattr(ws, "_SESSION_TOKEN_SOURCE", "env")
with caplog.at_level(logging.WARNING, logger="hermes_cli.web_server"):
reason, cred = ws._ws_auth_reason(fake_ws)
assert reason == "token_mismatch"
assert cred == "token"
assert any(
"stale HERMES_DASHBOARD_SESSION_TOKEN" in m
for m in self._warning_messages(caplog)
)

def test_ws_auth_reason_emits_hint_when_injected(self, monkeypatch, caplog):
# End-to-end: a bad token in loopback mode with an injected-sourced
# server token STILL warns — the marker survives the stale-.env
# clobber, so "injected" does not exonerate the mismatch. The
# rejection itself is preserved.
import hermes_cli.web_server as ws

fake_ws = SimpleNamespace(
query_params={"token": "wrong-token"},
client=SimpleNamespace(host="127.0.0.1"),
)
monkeypatch.setattr(ws.app.state, "auth_required", False, raising=False)
monkeypatch.setattr(ws.app.state, "bound_host", None, raising=False)
monkeypatch.setattr(ws, "_SESSION_TOKEN_SOURCE", "injected")
with caplog.at_level(logging.WARNING, logger="hermes_cli.web_server"):
reason, cred = ws._ws_auth_reason(fake_ws)
assert reason == "token_mismatch"
assert cred == "token"
assert any(
"stale HERMES_DASHBOARD_SESSION_TOKEN" in m
for m in self._warning_messages(caplog)
)


# ---------------------------------------------------------------------------
# web_server tests (FastAPI endpoints)
# ---------------------------------------------------------------------------
Expand Down
17 changes: 17 additions & 0 deletions tests/hermes_cli/test_web_server_console_ws.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from __future__ import annotations

import logging
import time
from urllib.parse import urlencode

Expand Down Expand Up @@ -70,6 +71,22 @@ def test_console_ws_rejects_missing_or_bad_token(console_client):
assert exc.value.code == 4401


def test_console_ws_token_mismatch_hints_at_stale_dotenv(console_client, monkeypatch, caplog):
"""A loopback token mismatch with an env-sourced server token logs an
actionable hint naming the stale ~/.hermes/.env value — while still
rejecting the connection (auth semantics unchanged)."""
monkeypatch.setattr(web_server, "_SESSION_TOKEN_SOURCE", "env")
with caplog.at_level(logging.WARNING, logger="hermes_cli.web_server"):
with pytest.raises(WebSocketDisconnect) as exc:
with console_client.websocket_connect(_url(token="wrong")):
pass
assert exc.value.code == 4401
assert any(
"stale HERMES_DASHBOARD_SESSION_TOKEN" in r.message and ".env" in r.message
for r in caplog.records
)


def test_console_ws_cancel_returns_to_prompt(console_client, monkeypatch):
from hermes_cli.console_engine import ConsoleResult, HermesConsoleEngine

Expand Down
Loading