mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
feat(dashboard-auth): delete legacy _SESSION_TOKEN server-side
Removes the ephemeral dashboard session token entirely from the server: - delete _SESSION_TOKEN, _SESSION_HEADER_NAME, _has_valid_session_token - delete the no-op auth_middleware shell (loopback has no identity gate; the bind + CSRF guard + CORS are the boundary) - _serve_index no longer injects window.__HERMES_SESSION_TOKEN__ in either mode (loopback needs no credential; gated reads identity from /api/auth/me) - PTY-child WS URL builders (_build_gateway_ws_url / _build_sidecar_url) emit a bare loopback URL with no ?token= (gated mode unchanged: ?internal=) - redefine the --insecure warning: names the CSRF + Host/Origin guards that still apply, drops the stale 'no robust authentication' wording The pluggable OAuth gate is now the ONLY identity gate. On loopback there is no per-request identity check at all. Tests: every file that pinned the old _SESSION_TOKEN contract is updated to the new reality. Obsolete tests (token-unlocks-route, index-injects-token) are deleted (they tested deleted behavior; the no-identity-gate siblings already pin the new contract). Sensitive endpoints retain gated-mode coverage. Full tests/hermes_cli (7049), tests/plugins (1245), and the docker dashboard suite (8) are green. Co-authored-by: Hermes subagent <noreply@nousresearch.com>
This commit is contained in:
Ben
parent
6cf12eef4e
commit
85d9d27043
16 changed files with 170 additions and 264 deletions
|
|
@ -16,7 +16,6 @@ import base64
|
|||
import binascii
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
import hmac
|
||||
import importlib.util
|
||||
import json
|
||||
import logging
|
||||
|
|
@ -174,23 +173,12 @@ def _get_event_state(app: "FastAPI"):
|
|||
|
||||
app = FastAPI(title="Hermes Agent", version=__version__, lifespan=_lifespan)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Session token for protecting sensitive endpoints (reveal).
|
||||
# The desktop shell mints the token and injects it via
|
||||
# HERMES_DASHBOARD_SESSION_TOKEN so its main process can authenticate the
|
||||
# /api calls it makes on the user's behalf; otherwise we generate one fresh
|
||||
# on every server start. Either way it dies when the process exits and is
|
||||
# injected into the SPA HTML so only the legitimate web UI can use it.
|
||||
# ---------------------------------------------------------------------------
|
||||
_SESSION_TOKEN = os.environ.get("HERMES_DASHBOARD_SESSION_TOKEN") or secrets.token_urlsafe(32)
|
||||
_SESSION_HEADER_NAME = "X-Hermes-Session-Token"
|
||||
|
||||
# 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
|
||||
# `/api/ws` + `/api/pty` WebSockets, so the embedded-chat surface is an
|
||||
# unconditional part of the dashboard. Kept as a module-level constant (rather
|
||||
# than inlining ``True`` at every gate) so the WS endpoints and the SPA token
|
||||
# injection share a single, testable seam.
|
||||
# than inlining ``True`` at every gate) so the WS endpoints and the SPA
|
||||
# bootstrap share a single, testable seam.
|
||||
_DASHBOARD_EMBEDDED_CHAT_ENABLED = True
|
||||
|
||||
# Simple rate limiter for the reveal endpoint
|
||||
|
|
@ -227,39 +215,6 @@ from hermes_cli.dashboard_auth.public_paths import (
|
|||
)
|
||||
|
||||
|
||||
def _has_valid_session_token(request: Request) -> bool:
|
||||
"""True if the request carries a valid dashboard session token.
|
||||
|
||||
The dedicated session header avoids collisions with reverse proxies that
|
||||
already use ``Authorization`` (for example Caddy ``basic_auth``). We still
|
||||
accept the legacy Bearer path for backward compatibility with older
|
||||
dashboard bundles.
|
||||
"""
|
||||
session_header = request.headers.get(_SESSION_HEADER_NAME, "")
|
||||
if session_header and hmac.compare_digest(
|
||||
session_header.encode(),
|
||||
_SESSION_TOKEN.encode(),
|
||||
):
|
||||
return True
|
||||
|
||||
auth = request.headers.get("authorization", "")
|
||||
expected = f"Bearer {_SESSION_TOKEN}"
|
||||
return hmac.compare_digest(auth.encode(), expected.encode())
|
||||
|
||||
|
||||
# Routes that may also authenticate via a ``?token=`` query param, for download
|
||||
# links opened by the OS shell or a new browser tab where the session header
|
||||
# can't be set. Kept narrow — same query-token tradeoff as the /api/pty WS.
|
||||
_QUERY_TOKEN_API_PATHS: frozenset[str] = frozenset({"/api/files/download"})
|
||||
|
||||
|
||||
def _has_valid_query_token(request: Request, path: str) -> bool:
|
||||
if path not in _QUERY_TOKEN_API_PATHS:
|
||||
return False
|
||||
token = request.query_params.get("token", "")
|
||||
return bool(token) and hmac.compare_digest(token.encode(), _SESSION_TOKEN.encode())
|
||||
|
||||
|
||||
def _require_token(request: Request) -> None:
|
||||
"""Authorize a sensitive endpoint, raising 401 if the caller isn't allowed.
|
||||
|
||||
|
|
@ -449,10 +404,11 @@ async def csrf_guard_middleware(request: Request, call_next):
|
|||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dashboard OAuth auth gate — engaged only when start_server flags the
|
||||
# bind as non-loopback-without-insecure. No-op pass-through in loopback
|
||||
# mode so the legacy auth_middleware (below) handles those binds via
|
||||
# the injected ``_SESSION_TOKEN``. Registered between host_header and
|
||||
# auth_middleware so the order is: host check → cookie auth → token auth.
|
||||
# bind as non-loopback-without-insecure (``app.state.auth_required``). It is
|
||||
# a no-op pass-through on a loopback bind, where the dashboard runs no
|
||||
# identity gate at all: the loopback bind is the security boundary, the
|
||||
# csrf_guard_middleware blocks cross-origin mutations, and the localhost-only
|
||||
# CORS policy blocks cross-origin reads.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
|
|
@ -462,25 +418,6 @@ async def _dashboard_auth_gate(request: Request, call_next):
|
|||
return await gated_auth_middleware(request, call_next)
|
||||
|
||||
|
||||
@app.middleware("http")
|
||||
async def auth_middleware(request: Request, call_next):
|
||||
"""Loopback path: NO identity gate.
|
||||
|
||||
The dashboard's identity authentication is the pluggable gate
|
||||
(``gated_auth_middleware``), engaged only on non-loopback binds. On a
|
||||
loopback bind the OS boundary IS the security boundary: nothing off the
|
||||
machine can reach 127.0.0.1. Cross-origin mutations are rejected by
|
||||
``csrf_guard_middleware``; cross-origin reads are neutralised by the
|
||||
localhost-only CORS policy. There is no per-request identity token on
|
||||
loopback anymore (the legacy ``_SESSION_TOKEN`` is being removed).
|
||||
|
||||
This remains a registered middleware (rather than being deleted) so the
|
||||
Phase-2 diff is minimal and reversible; Phase 5 removes it entirely once
|
||||
the token symbol is gone.
|
||||
"""
|
||||
return await call_next(request)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Config schema — auto-generated from DEFAULT_CONFIG
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -10163,10 +10100,12 @@ async def get_models_analytics(days: int = 30, profile: Optional[str] = None):
|
|||
# WebSocket. The browser renders the ANSI through xterm.js (see
|
||||
# web/src/pages/ChatPage.tsx).
|
||||
#
|
||||
# Auth: ``?token=<session_token>`` query param (browsers can't set
|
||||
# Authorization on the WS upgrade). Same ephemeral ``_SESSION_TOKEN`` as
|
||||
# REST. Localhost-only — we defensively reject non-loopback clients even
|
||||
# though uvicorn binds to 127.0.0.1.
|
||||
# Auth: loopback binds require no credential on the WS upgrade — the
|
||||
# peer-IP loopback gate + Host/Origin guard are the boundary. Gated
|
||||
# (non-loopback) binds require a single-use ``?ticket=`` (browser) or the
|
||||
# process-lifetime ``?internal=`` credential (server-spawned PTY child);
|
||||
# browsers can't set Authorization on a WS upgrade. Localhost-only on a
|
||||
# loopback bind — we defensively reject non-loopback clients.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# PTY bridge: POSIX uses pty_bridge (fcntl/termios/ptyprocess); native Windows
|
||||
|
|
@ -10229,9 +10168,10 @@ def _ws_client_reason(ws: "WebSocket") -> Optional[str]:
|
|||
def _ws_client_is_allowed(ws: "WebSocket") -> bool:
|
||||
"""Check if the WebSocket client IP is acceptable.
|
||||
|
||||
Loopback bind: only loopback clients allowed — the legacy
|
||||
``?token=<_SESSION_TOKEN>`` path is the only auth we have, so we
|
||||
don't want LAN hosts guessing tokens.
|
||||
Loopback bind: only loopback clients allowed — there is no identity
|
||||
token on a loopback WS upgrade anymore, so the loopback-only peer gate
|
||||
(plus the Host/Origin guard) IS the boundary; we don't want LAN hosts
|
||||
reaching the credential-free loopback WS.
|
||||
|
||||
Explicit non-loopback bind (``--host 0.0.0.0``, ``--host ::``, or a
|
||||
specific address such as a Tailscale/LAN IP, always with
|
||||
|
|
@ -10339,11 +10279,12 @@ def _ws_auth_reason(ws: "WebSocket") -> tuple[Optional[str], str]:
|
|||
machine-parseable token explaining the rejection (``no_credential``,
|
||||
``token_mismatch``, ``ticket_invalid``, ``internal_invalid``).
|
||||
``credential`` names which credential type was presented (``ticket``,
|
||||
``internal``, ``token``, or ``none``) so the accepted path can log *how*
|
||||
a peer authed, not just that it did.
|
||||
``internal``, or ``none``/``loopback``) so the accepted path can log
|
||||
*how* a peer authed, not just that it did.
|
||||
|
||||
Loopback / ``--insecure``: legacy ``?token=<_SESSION_TOKEN>`` query
|
||||
parameter, constant-time compared.
|
||||
Loopback / ``--insecure``: NO credential is consulted (returns
|
||||
``(None, "loopback")``). The peer-IP loopback gate + Host/Origin guard
|
||||
are the boundary.
|
||||
|
||||
Gated (public bind, no ``--insecure``): one of two credentials —
|
||||
|
||||
|
|
@ -10516,10 +10457,11 @@ def _resolve_chat_argv(
|
|||
def _build_gateway_ws_url() -> Optional[str]:
|
||||
"""ws:// URL the PTY child should attach to for JSON-RPC gateway traffic.
|
||||
|
||||
Loopback / ``--insecure``: ``?token=<_SESSION_TOKEN>``.
|
||||
Loopback / ``--insecure``: a bare ``/api/ws`` URL with no credential —
|
||||
the child connects from loopback, which the WS peer-IP + Host/Origin
|
||||
guard accepts without a token (there is no identity token anymore).
|
||||
|
||||
Gated mode: the legacy token path is rejected by ``_ws_auth_ok``, so the
|
||||
server-spawned PTY child authenticates with the process-lifetime internal
|
||||
Gated mode: the child authenticates with the process-lifetime internal
|
||||
credential (``?internal=``). It must NOT use a single-use browser ticket:
|
||||
the child reads this URL once at startup and reuses it on every reconnect,
|
||||
and a 30s-TTL ticket can expire before a slow cold boot even dials.
|
||||
|
|
@ -10540,16 +10482,17 @@ def _build_gateway_ws_url() -> Optional[str]:
|
|||
from hermes_cli.dashboard_auth.ws_tickets import internal_ws_credential
|
||||
|
||||
qs = urllib.parse.urlencode({"internal": internal_ws_credential()})
|
||||
else:
|
||||
qs = urllib.parse.urlencode({"token": _SESSION_TOKEN})
|
||||
|
||||
return f"ws://{netloc}/api/ws?{qs}"
|
||||
return f"ws://{netloc}/api/ws?{qs}"
|
||||
# Loopback: no credential needed (peer-IP + Host/Origin guard is the gate).
|
||||
return f"ws://{netloc}/api/ws"
|
||||
|
||||
|
||||
def _build_sidecar_url(channel: str) -> Optional[str]:
|
||||
"""ws:// URL the PTY child should publish events to, or None when unbound.
|
||||
|
||||
Loopback / ``--insecure``: uses ``?token=<_SESSION_TOKEN>``.
|
||||
Loopback / ``--insecure``: a bare ``/api/pub`` URL with no credential
|
||||
(the child connects from loopback; the peer-IP + Host/Origin guard is
|
||||
the gate).
|
||||
|
||||
Gated mode: authenticates with the process-lifetime internal credential
|
||||
(``?internal=``), the same one ``_build_gateway_ws_url`` uses. The PTY
|
||||
|
|
@ -10576,9 +10519,9 @@ def _build_sidecar_url(channel: str) -> Optional[str]:
|
|||
qs = urllib.parse.urlencode(
|
||||
{"internal": internal_ws_credential(), "channel": channel}
|
||||
)
|
||||
else:
|
||||
qs = urllib.parse.urlencode({"token": _SESSION_TOKEN, "channel": channel})
|
||||
|
||||
return f"ws://{netloc}/api/pub?{qs}"
|
||||
# Loopback: no credential; only the channel is needed.
|
||||
qs = urllib.parse.urlencode({"channel": channel})
|
||||
return f"ws://{netloc}/api/pub?{qs}"
|
||||
|
||||
|
||||
|
|
@ -10908,37 +10851,30 @@ def mount_spa(application: FastAPI):
|
|||
_index_path = WEB_DIST / "index.html"
|
||||
|
||||
def _serve_index(prefix: str = ""):
|
||||
"""Return index.html with the session token + base-path injected.
|
||||
"""Return index.html with the base-path + auth-mode flag injected.
|
||||
|
||||
``prefix`` is the normalised ``X-Forwarded-Prefix`` (e.g. ``/hermes``)
|
||||
or empty string when served at root.
|
||||
|
||||
When the OAuth auth gate is active (``app.state.auth_required``),
|
||||
the legacy ``_SESSION_TOKEN`` is NOT injected — the SPA reads
|
||||
identity from ``/api/auth/me`` over cookie auth instead. The
|
||||
``__HERMES_AUTH_REQUIRED__`` flag lets the SPA pick the right
|
||||
auth scheme for /api/pty and /api/ws (ticket vs token).
|
||||
No identity token is injected in either mode. On a loopback bind the
|
||||
SPA needs no credential (the bind is the boundary; the CSRF guard
|
||||
covers mutations). When the OAuth gate is active
|
||||
(``app.state.auth_required``) the SPA reads identity from
|
||||
``/api/auth/me`` over cookie auth. The ``__HERMES_AUTH_REQUIRED__``
|
||||
flag lets the SPA pick the right WS-auth scheme for /api/pty and
|
||||
/api/ws (ticket in gated mode, no credential on loopback).
|
||||
"""
|
||||
html = _index_path.read_text(encoding="utf-8")
|
||||
chat_js = "true" if _DASHBOARD_EMBEDDED_CHAT_ENABLED else "false"
|
||||
gated = bool(getattr(app.state, "auth_required", False))
|
||||
gated_js = "true" if gated else "false"
|
||||
if gated:
|
||||
bootstrap_script = (
|
||||
f"<script>"
|
||||
f"window.__HERMES_DASHBOARD_EMBEDDED_CHAT__={chat_js};"
|
||||
f'window.__HERMES_BASE_PATH__="{prefix}";'
|
||||
f"window.__HERMES_AUTH_REQUIRED__={gated_js};"
|
||||
f"</script>"
|
||||
)
|
||||
else:
|
||||
bootstrap_script = (
|
||||
f'<script>window.__HERMES_SESSION_TOKEN__="{_SESSION_TOKEN}";'
|
||||
f"window.__HERMES_DASHBOARD_EMBEDDED_CHAT__={chat_js};"
|
||||
f'window.__HERMES_BASE_PATH__="{prefix}";'
|
||||
f"window.__HERMES_AUTH_REQUIRED__={gated_js};"
|
||||
f"</script>"
|
||||
)
|
||||
bootstrap_script = (
|
||||
f"<script>"
|
||||
f"window.__HERMES_DASHBOARD_EMBEDDED_CHAT__={chat_js};"
|
||||
f'window.__HERMES_BASE_PATH__="{prefix}";'
|
||||
f"window.__HERMES_AUTH_REQUIRED__={gated_js};"
|
||||
f"</script>"
|
||||
)
|
||||
if prefix:
|
||||
# Rewrite absolute asset URLs baked into the Vite build so the
|
||||
# browser fetches them through the same proxy prefix.
|
||||
|
|
@ -12086,10 +12022,13 @@ def start_server(
|
|||
", ".join(p.name for p in list_providers()),
|
||||
)
|
||||
elif host not in _LOOPBACK_HOST_VALUES and allow_public:
|
||||
# --insecure path — no auth, loud warning.
|
||||
# --insecure path — no identity gate, loud warning.
|
||||
_log.warning(
|
||||
"Binding to %s with --insecure — the dashboard has no robust "
|
||||
"authentication. Only use on trusted networks.", host,
|
||||
"Binding to %s with --insecure — no identity authentication. "
|
||||
"The Sec-Fetch-Site CSRF guard and the WebSocket Host/Origin "
|
||||
"guard still apply, but anyone who can reach this address can "
|
||||
"use the dashboard. Rely on network controls; only use on "
|
||||
"trusted networks.", host,
|
||||
)
|
||||
|
||||
# Record the bound host so host_header_middleware can validate incoming
|
||||
|
|
|
|||
|
|
@ -48,7 +48,7 @@ def test_cross_origin_mutation_blocked(loopback_client, sfs):
|
|||
r = loopback_client.post(
|
||||
_MUTATING_ROUTE,
|
||||
headers={
|
||||
"X-Hermes-Session-Token": web_server._SESSION_TOKEN,
|
||||
"X-Hermes-Session-Token": "stale-token-ignored",
|
||||
"Sec-Fetch-Site": sfs,
|
||||
},
|
||||
json={"key": "OPENAI_API_KEY", "value": "x"},
|
||||
|
|
@ -62,7 +62,7 @@ def test_same_origin_mutation_allowed(loopback_client, sfs):
|
|||
r = loopback_client.post(
|
||||
_MUTATING_ROUTE,
|
||||
headers={
|
||||
"X-Hermes-Session-Token": web_server._SESSION_TOKEN,
|
||||
"X-Hermes-Session-Token": "stale-token-ignored",
|
||||
"Sec-Fetch-Site": sfs,
|
||||
},
|
||||
json={"key": "OPENAI_API_KEY", "value": "x"},
|
||||
|
|
@ -76,7 +76,7 @@ def test_absent_header_fails_open(loopback_client):
|
|||
Sec-Fetch-Site and must NOT be blocked."""
|
||||
r = loopback_client.post(
|
||||
_MUTATING_ROUTE,
|
||||
headers={"X-Hermes-Session-Token": web_server._SESSION_TOKEN},
|
||||
headers={"X-Hermes-Session-Token": "stale-token-ignored"},
|
||||
json={"key": "OPENAI_API_KEY", "value": "x"},
|
||||
)
|
||||
assert r.status_code != 403
|
||||
|
|
|
|||
|
|
@ -17,14 +17,16 @@ def _client():
|
|||
pytest.skip("fastapi/starlette not installed")
|
||||
import hermes_state
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
client = TestClient(app)
|
||||
client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
# Loopback bind has no identity gate; no session header needed. A literal
|
||||
# header name is returned so the "bogus token is ignored" tests can still
|
||||
# send an arbitrary header under it.
|
||||
# Keep the state DB under the isolated HERMES_HOME for any handler that
|
||||
# touches it.
|
||||
hermes_state.DEFAULT_DB_PATH = get_hermes_home() / "state.db"
|
||||
return client, _SESSION_HEADER_NAME
|
||||
return client, "X-Hermes-Session-Token"
|
||||
|
||||
|
||||
class TestMcpEndpoints:
|
||||
|
|
|
|||
|
|
@ -52,31 +52,6 @@ def test_loopback_protected_route_no_identity_gate(client_loopback):
|
|||
assert r.status_code != 401
|
||||
|
||||
|
||||
def test_loopback_protected_route_accepts_session_token(client_loopback):
|
||||
"""The injected SPA token unlocks protected /api/ routes."""
|
||||
r = client_loopback.get(
|
||||
"/api/sessions",
|
||||
headers={"X-Hermes-Session-Token": web_server._SESSION_TOKEN},
|
||||
)
|
||||
# 200 or 404 (no sessions yet) both prove the auth layer let it through.
|
||||
# 500 is also acceptable if there's a downstream issue unrelated to auth.
|
||||
assert r.status_code != 401, (
|
||||
f"Expected auth to succeed but got 401; body: {r.text}"
|
||||
)
|
||||
|
||||
|
||||
def test_loopback_index_injects_session_token(client_loopback):
|
||||
"""Loopback mode keeps injecting the SPA token into index.html.
|
||||
|
||||
This is the property that the new auth gate MUST disable once a gated
|
||||
bind is detected. Phase 3 will add an inverse test for the gated path.
|
||||
"""
|
||||
r = client_loopback.get("/")
|
||||
if r.status_code == 404:
|
||||
pytest.skip("WEB_DIST not built in this env")
|
||||
assert "__HERMES_SESSION_TOKEN__" in r.text
|
||||
|
||||
|
||||
def test_loopback_host_header_validation_still_enforced(client_loopback):
|
||||
"""DNS-rebinding protection: a foreign Host header is rejected."""
|
||||
r = client_loopback.get("/api/status", headers={"Host": "evil.test"})
|
||||
|
|
|
|||
|
|
@ -261,7 +261,7 @@ class TestWsAuthOkGated:
|
|||
"""Critical: gated mode must NOT honour the legacy token path
|
||||
even when someone has access to the in-process value of
|
||||
_SESSION_TOKEN (e.g. a leaked log line)."""
|
||||
ws = _fake_ws(query={"token": web_server._SESSION_TOKEN})
|
||||
ws = _fake_ws(query={"token": "stale-token-ignored"})
|
||||
assert web_server._ws_auth_ok(ws) is False
|
||||
|
||||
def test_rejection_audit_logs(self, gated_app, tmp_path, monkeypatch):
|
||||
|
|
@ -502,11 +502,16 @@ class TestWsHostOriginGuardOrigins:
|
|||
|
||||
|
||||
class TestSidecarUrl:
|
||||
def test_loopback_uses_session_token(self, loopback_app):
|
||||
def test_loopback_has_no_credential(self, loopback_app):
|
||||
# Loopback child connects from localhost; the peer-IP + Host/Origin
|
||||
# guard is the gate, so the sidecar URL carries no credential — just
|
||||
# the channel. (The legacy ?token= is gone.)
|
||||
url = web_server._build_sidecar_url("ch-1")
|
||||
assert url is not None
|
||||
assert f"token={web_server._SESSION_TOKEN}" in url
|
||||
assert "token=" not in url
|
||||
assert "ticket=" not in url
|
||||
assert "internal=" not in url
|
||||
assert "channel=ch-1" in url
|
||||
|
||||
def test_gated_uses_internal_credential(self, gated_app):
|
||||
url = web_server._build_sidecar_url("ch-1")
|
||||
|
|
@ -539,11 +544,13 @@ class TestSidecarUrl:
|
|||
|
||||
|
||||
class TestGatewayWsUrl:
|
||||
def test_loopback_uses_session_token(self, loopback_app):
|
||||
def test_loopback_has_no_credential(self, loopback_app):
|
||||
# Loopback: bare /api/ws with no credential (peer-IP + Host/Origin
|
||||
# guard is the gate; the legacy ?token= is gone).
|
||||
url = web_server._build_gateway_ws_url()
|
||||
assert url is not None
|
||||
assert "/api/ws?" in url
|
||||
assert f"token={web_server._SESSION_TOKEN}" in url
|
||||
assert url.endswith("/api/ws")
|
||||
assert "token=" not in url
|
||||
assert "internal=" not in url
|
||||
|
||||
def test_gated_uses_internal_credential(self, gated_app):
|
||||
|
|
|
|||
|
|
@ -85,7 +85,7 @@ def test_gated_ignores_legacy_token_header(gated_client):
|
|||
a *valid* ``X-Hermes-Session-Token`` and no cookie must still 401."""
|
||||
r = gated_client.get(
|
||||
"/api/sessions",
|
||||
headers={"X-Hermes-Session-Token": web_server._SESSION_TOKEN},
|
||||
headers={"X-Hermes-Session-Token": "stale-token-ignored"},
|
||||
)
|
||||
assert r.status_code == 401
|
||||
assert r.json().get("error") in ("unauthenticated", "session_expired")
|
||||
|
|
@ -171,7 +171,7 @@ def test_ws_gated_rejects_legacy_token():
|
|||
web_server.app.state.auth_required = True
|
||||
try:
|
||||
reason, cred = web_server._ws_auth_reason(
|
||||
_fake_ws({"token": web_server._SESSION_TOKEN})
|
||||
_fake_ws({"token": "stale-token-ignored"})
|
||||
)
|
||||
assert reason == "no_credential" # token ignored; no ticket present
|
||||
finally:
|
||||
|
|
|
|||
|
|
@ -187,12 +187,11 @@ def test_migration_disables_existing_dangerous_entry(tmp_path):
|
|||
|
||||
def test_dashboard_mcp_add_rejects_dangerous_entry():
|
||||
from fastapi.testclient import TestClient
|
||||
from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN, app
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
client = TestClient(app)
|
||||
response = client.post(
|
||||
"/api/mcp/servers",
|
||||
headers={_SESSION_HEADER_NAME: _SESSION_TOKEN},
|
||||
json={"name": "evil", **_dangerous_entry()},
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -28,10 +28,12 @@ import httpx
|
|||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from hermes_cli.web_server import _SESSION_TOKEN, app
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
client = TestClient(app)
|
||||
HEADERS = {"X-Hermes-Session-Token": _SESSION_TOKEN}
|
||||
# Loopback bind has no identity gate; no session header needed. Kept as an
|
||||
# empty mapping so the existing call sites can keep passing headers=HEADERS.
|
||||
HEADERS: dict[str, str] = {}
|
||||
|
||||
|
||||
def _make_profile_home(tmp_path, monkeypatch, profile="coder"):
|
||||
|
|
|
|||
|
|
@ -184,35 +184,6 @@ class TestRedactKey:
|
|||
assert "not set" in result.lower() or result == "***" or "\x1b" in result
|
||||
|
||||
|
||||
class TestSessionTokenInjection:
|
||||
"""The desktop shell mints HERMES_DASHBOARD_SESSION_TOKEN and signs its
|
||||
/api + /api/ws calls with it. The backend must adopt that token, else every
|
||||
desktop request 401s ("gateway is offline"). A main-merge once silently
|
||||
dropped this read — this guards the contract, not a literal value.
|
||||
"""
|
||||
|
||||
def test_honors_injected_token(self, monkeypatch):
|
||||
import importlib
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.setenv("HERMES_DASHBOARD_SESSION_TOKEN", "desktop-seeded-token")
|
||||
try:
|
||||
importlib.reload(ws)
|
||||
assert ws._SESSION_TOKEN == "desktop-seeded-token"
|
||||
finally:
|
||||
monkeypatch.delenv("HERMES_DASHBOARD_SESSION_TOKEN", raising=False)
|
||||
importlib.reload(ws)
|
||||
|
||||
def test_falls_back_to_random_token(self, monkeypatch):
|
||||
import importlib
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.delenv("HERMES_DASHBOARD_SESSION_TOKEN", raising=False)
|
||||
importlib.reload(ws)
|
||||
|
||||
assert ws._SESSION_TOKEN and len(ws._SESSION_TOKEN) >= 32
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# web_server tests (FastAPI endpoints)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -231,12 +202,12 @@ class TestWebServerEndpoints:
|
|||
|
||||
import hermes_state
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")
|
||||
|
||||
self.client = TestClient(app)
|
||||
self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
# Loopback bind has no identity gate; no session header needed.
|
||||
|
||||
def test_get_status(self):
|
||||
resp = self.client.get("/api/status")
|
||||
|
|
@ -315,12 +286,10 @@ class TestWebServerEndpoints:
|
|||
outside the media roots, so the handler returns 403 — the point is it's
|
||||
no longer 401. Identity is enforced only in gated mode.
|
||||
"""
|
||||
from hermes_cli.web_server import _SESSION_HEADER_NAME
|
||||
|
||||
resp = self.client.get(
|
||||
"/api/media",
|
||||
params={"path": "/tmp/x.png"},
|
||||
headers={_SESSION_HEADER_NAME: "wrong-token"},
|
||||
headers={"X-Hermes-Session-Token": "wrong-token"},
|
||||
)
|
||||
assert resp.status_code != 401
|
||||
assert resp.status_code == 403
|
||||
|
|
@ -1368,12 +1337,10 @@ class TestWebServerEndpoints:
|
|||
def test_reveal_env_var(self, tmp_path):
|
||||
"""POST /api/env/reveal should return the real unredacted value."""
|
||||
from hermes_cli.config import save_env_value
|
||||
from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
save_env_value("TEST_REVEAL_KEY", "super-secret-value-12345")
|
||||
resp = self.client.post(
|
||||
"/api/env/reveal",
|
||||
json={"key": "TEST_REVEAL_KEY"},
|
||||
headers={_SESSION_HEADER_NAME: _SESSION_TOKEN},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
|
|
@ -1382,11 +1349,9 @@ class TestWebServerEndpoints:
|
|||
|
||||
def test_reveal_env_var_not_found(self):
|
||||
"""POST /api/env/reveal should 404 for unknown keys."""
|
||||
from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
resp = self.client.post(
|
||||
"/api/env/reveal",
|
||||
json={"key": "NONEXISTENT_KEY_XYZ"},
|
||||
headers={_SESSION_HEADER_NAME: _SESSION_TOKEN},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
|
@ -1459,28 +1424,28 @@ class TestWebServerEndpoints:
|
|||
of 401ing. Identity is enforced only in gated mode.
|
||||
"""
|
||||
from hermes_cli.config import save_env_value
|
||||
from hermes_cli.web_server import _SESSION_HEADER_NAME
|
||||
save_env_value("TEST_REVEAL_BADAUTH", "secret-value")
|
||||
resp = self.client.post(
|
||||
"/api/env/reveal",
|
||||
json={"key": "TEST_REVEAL_BADAUTH"},
|
||||
headers={_SESSION_HEADER_NAME: "wrong-token-here"},
|
||||
headers={"X-Hermes-Session-Token": "wrong-token-here"},
|
||||
)
|
||||
assert resp.status_code != 401
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["value"] == "secret-value"
|
||||
|
||||
def test_reveal_env_var_custom_session_header_ignores_proxy_authorization(self, tmp_path):
|
||||
"""A valid dashboard session header should coexist with proxy auth."""
|
||||
"""A stale dashboard session header should be ignored, not break the
|
||||
request: on loopback there's no identity gate, and a proxy
|
||||
``Authorization`` header must not interfere with the reveal."""
|
||||
from hermes_cli.config import save_env_value
|
||||
from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
|
||||
save_env_value("TEST_REVEAL_PROXY_AUTH", "secret-value")
|
||||
resp = self.client.post(
|
||||
"/api/env/reveal",
|
||||
json={"key": "TEST_REVEAL_PROXY_AUTH"},
|
||||
headers={
|
||||
_SESSION_HEADER_NAME: _SESSION_TOKEN,
|
||||
"X-Hermes-Session-Token": "stale-token-ignored",
|
||||
"Authorization": "Basic dXNlcjpwYXNz",
|
||||
},
|
||||
)
|
||||
|
|
@ -1497,13 +1462,12 @@ class TestWebServerEndpoints:
|
|||
that token mechanism is slated for deletion.)
|
||||
"""
|
||||
from hermes_cli.config import save_env_value
|
||||
from hermes_cli.web_server import _SESSION_TOKEN
|
||||
|
||||
save_env_value("TEST_REVEAL_LEGACY_AUTH", "secret-value")
|
||||
resp = self.client.post(
|
||||
"/api/env/reveal",
|
||||
json={"key": "TEST_REVEAL_LEGACY_AUTH"},
|
||||
headers={"Authorization": f"Bearer {_SESSION_TOKEN}"},
|
||||
headers={"Authorization": "Bearer stale-token-ignored"},
|
||||
)
|
||||
|
||||
assert resp.status_code != 401
|
||||
|
|
@ -2547,9 +2511,9 @@ class TestConfigRoundTrip:
|
|||
from starlette.testclient import TestClient
|
||||
except ImportError:
|
||||
pytest.skip("fastapi/starlette not installed")
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
self.client = TestClient(app)
|
||||
self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
# Loopback bind has no identity gate; no session header needed.
|
||||
|
||||
def test_get_config_no_internal_keys(self):
|
||||
"""GET /api/config should not expose _config_version or _model_meta."""
|
||||
|
|
@ -2683,12 +2647,12 @@ class TestNewEndpoints:
|
|||
|
||||
import hermes_state
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")
|
||||
|
||||
self.client = TestClient(app)
|
||||
self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
# Loopback bind has no identity gate; no session header needed.
|
||||
|
||||
def test_get_logs_default(self):
|
||||
resp = self.client.get("/api/logs")
|
||||
|
|
@ -4025,9 +3989,9 @@ class TestStatusRemoteGateway:
|
|||
except ImportError:
|
||||
pytest.skip("fastapi/starlette not installed")
|
||||
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
self.client = TestClient(app)
|
||||
self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
# Loopback bind has no identity gate; no session header needed.
|
||||
|
||||
def test_status_falls_back_to_remote_probe(self, monkeypatch):
|
||||
"""When local PID check fails and remote probe succeeds, gateway shows running."""
|
||||
|
|
@ -4445,7 +4409,7 @@ class TestBulkDeleteSessionsEndpoint:
|
|||
|
||||
import hermes_state
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
monkeypatch.setattr(
|
||||
hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db"
|
||||
|
|
@ -4453,7 +4417,7 @@ class TestBulkDeleteSessionsEndpoint:
|
|||
|
||||
self.client = TestClient(app)
|
||||
self.auth_client = TestClient(app)
|
||||
self.auth_client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
# Loopback bind has no identity gate; no session header needed.
|
||||
|
||||
def _seed(self, ids):
|
||||
from hermes_state import SessionDB
|
||||
|
|
@ -4578,7 +4542,7 @@ class TestDeleteEmptySessionsEndpoint:
|
|||
|
||||
import hermes_state
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
# Pin the SessionDB to the isolated HERMES_HOME so each test
|
||||
# starts with a clean state.db.
|
||||
|
|
@ -4588,7 +4552,7 @@ class TestDeleteEmptySessionsEndpoint:
|
|||
|
||||
self.client = TestClient(app)
|
||||
self.auth_client = TestClient(app)
|
||||
self.auth_client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
# Loopback bind has no identity gate; no session header needed.
|
||||
|
||||
def _seed(self):
|
||||
"""Build the standard test corpus:
|
||||
|
|
@ -4736,13 +4700,13 @@ class TestPluginAPIAuth:
|
|||
|
||||
import hermes_state
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")
|
||||
|
||||
self.client = TestClient(app)
|
||||
self.auth_client = TestClient(app)
|
||||
self.auth_client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
# Loopback bind has no identity gate; no session header needed.
|
||||
|
||||
def test_plugin_route_no_identity_gate_on_loopback(self):
|
||||
"""Plugin API GET routes serve on loopback without a session token."""
|
||||
|
|
@ -4996,7 +4960,9 @@ class TestPtyWebSocket:
|
|||
# its own fake argv via ``ws._resolve_chat_argv``.
|
||||
self.ws_module = ws
|
||||
monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True)
|
||||
self.token = ws._SESSION_TOKEN
|
||||
# Loopback ignores any ?token= on the WS upgrade (no identity gate);
|
||||
# a literal keeps _url() working for the connect tests below.
|
||||
self.token = "ignored"
|
||||
self.client = TestClient(ws.app)
|
||||
|
||||
def _url(self, token: str | None = None, **params: str) -> str:
|
||||
|
|
@ -5284,7 +5250,9 @@ class TestPtyWebSocket:
|
|||
url = captured.get("sidecar_url") or ""
|
||||
assert url.startswith("ws://127.0.0.1:9119/api/pub?")
|
||||
assert "channel=abc-123" in url
|
||||
assert "token=" in url
|
||||
# Loopback sidecar URL carries no credential — the bind + peer-IP guard
|
||||
# are the boundary (the legacy ?token= is gone).
|
||||
assert "token=" not in url
|
||||
|
||||
def test_pub_broadcasts_to_events_subscribers(self):
|
||||
"""A frame handed to _broadcast_event is sent verbatim to every
|
||||
|
|
@ -5370,8 +5338,10 @@ def test_resolve_chat_argv_injects_gateway_ws_url(monkeypatch):
|
|||
|
||||
assert env is not None
|
||||
gateway_url = env.get("HERMES_TUI_GATEWAY_URL", "")
|
||||
assert gateway_url.startswith("ws://127.0.0.1:9119/api/ws?")
|
||||
assert "token=" in gateway_url
|
||||
# Loopback gateway URL is a bare /api/ws with no credential (the legacy
|
||||
# ?token= is gone; the loopback bind + peer-IP guard are the boundary).
|
||||
assert gateway_url == "ws://127.0.0.1:9119/api/ws"
|
||||
assert "token=" not in gateway_url
|
||||
|
||||
|
||||
class TestDashboardPluginStaticAssetAllowlist:
|
||||
|
|
@ -5497,10 +5467,10 @@ class TestValidateProviderCredential:
|
|||
except ImportError:
|
||||
pytest.skip("fastapi/starlette not installed")
|
||||
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
self.client = TestClient(app)
|
||||
self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
# Loopback bind has no identity gate; no session header needed.
|
||||
|
||||
def _post(self, key, value):
|
||||
return self.client.post("/api/providers/validate", json={"key": key, "value": value})
|
||||
|
|
|
|||
|
|
@ -7,6 +7,10 @@ from starlette.testclient import TestClient
|
|||
|
||||
from hermes_cli import web_server
|
||||
|
||||
# These tests mutate web_server.app.state (auth_required / bound_host); share
|
||||
# the dashboard-auth xdist group so they don't race other app.state mutators.
|
||||
pytestmark = pytest.mark.xdist_group("dashboard_auth_app_state")
|
||||
|
||||
|
||||
def _client_with_app_state():
|
||||
prev_auth_required = getattr(web_server.app.state, "auth_required", None)
|
||||
|
|
@ -15,7 +19,7 @@ def _client_with_app_state():
|
|||
web_server.app.state.bound_host = None
|
||||
|
||||
client = TestClient(web_server.app)
|
||||
client.headers[web_server._SESSION_HEADER_NAME] = web_server._SESSION_TOKEN
|
||||
# Loopback bind has no identity gate; no session header needed.
|
||||
return client, prev_auth_required, prev_bound_host
|
||||
|
||||
|
||||
|
|
@ -276,42 +280,56 @@ def test_download_returns_file_as_attachment(forced_files_client):
|
|||
assert "hello.txt" in disposition
|
||||
|
||||
|
||||
def test_download_authenticates_via_query_token(forced_files_client):
|
||||
def test_download_no_identity_gate_on_loopback(forced_files_client):
|
||||
"""Loopback download needs no credential after the legacy-token teardown.
|
||||
|
||||
The browser/shell-opened download (which can't set a session header) just
|
||||
works on a loopback bind — the bind is the security boundary. The old
|
||||
``?token=`` query-param escape hatch is gone with the token. Gated-mode
|
||||
enforcement is pinned by test_download_requires_auth_in_gated_mode below.
|
||||
"""
|
||||
client, root = forced_files_client
|
||||
file_path = _seed_file(client, root)
|
||||
|
||||
# Drop the session header so only the ?token= query param authenticates —
|
||||
# mirrors a browser/shell-opened download that can't set the session header.
|
||||
del client.headers[web_server._SESSION_HEADER_NAME]
|
||||
|
||||
ok = client.get(
|
||||
"/api/files/download",
|
||||
params={"path": str(file_path), "token": web_server._SESSION_TOKEN},
|
||||
)
|
||||
ok = client.get("/api/files/download", params={"path": str(file_path)})
|
||||
assert ok.status_code == 200
|
||||
assert ok.content == b"hello"
|
||||
|
||||
assert client.get(
|
||||
"/api/files/download", params={"path": str(file_path), "token": "nope"}
|
||||
).status_code == 401
|
||||
assert client.get(
|
||||
"/api/files/download", params={"path": str(file_path)}
|
||||
).status_code == 401
|
||||
# A stale/garbage ?token= is simply ignored, not rejected, on loopback.
|
||||
still_ok = client.get(
|
||||
"/api/files/download", params={"path": str(file_path), "token": "anything"}
|
||||
)
|
||||
assert still_ok.status_code == 200
|
||||
|
||||
|
||||
def test_query_token_does_not_authenticate_other_endpoints(forced_files_client):
|
||||
def test_download_requires_auth_in_gated_mode(forced_files_client, monkeypatch):
|
||||
"""In gated (non-loopback) mode the download endpoint requires a verified
|
||||
session cookie — a cookieless request 401s at the gate, and there is no
|
||||
``?token=`` query-param bypass."""
|
||||
from hermes_cli.dashboard_auth import clear_providers, register_provider
|
||||
from tests.hermes_cli.conftest_dashboard_auth import StubAuthProvider
|
||||
|
||||
client, root = forced_files_client
|
||||
file_path = _seed_file(client, root)
|
||||
|
||||
del client.headers[web_server._SESSION_HEADER_NAME]
|
||||
|
||||
# The query-token escape hatch is scoped to /api/files/download only; it must
|
||||
# not unlock the rest of the API surface.
|
||||
leaked = client.get(
|
||||
"/api/files/read",
|
||||
params={"path": str(file_path), "token": web_server._SESSION_TOKEN},
|
||||
)
|
||||
assert leaked.status_code == 401
|
||||
prev_host = getattr(web_server.app.state, "bound_host", None)
|
||||
clear_providers()
|
||||
register_provider(StubAuthProvider())
|
||||
web_server.app.state.bound_host = "fly-app.fly.dev"
|
||||
web_server.app.state.auth_required = True
|
||||
try:
|
||||
gated = TestClient(web_server.app, base_url="https://fly-app.fly.dev")
|
||||
# Cookieless → 401 at the gate, with or without a bogus ?token=.
|
||||
assert gated.get(
|
||||
"/api/files/download", params={"path": str(file_path)}
|
||||
).status_code == 401
|
||||
assert gated.get(
|
||||
"/api/files/download", params={"path": str(file_path), "token": "anything"}
|
||||
).status_code == 401
|
||||
finally:
|
||||
clear_providers()
|
||||
web_server.app.state.bound_host = prev_host
|
||||
web_server.app.state.auth_required = False
|
||||
|
||||
|
||||
def test_hosted_policy_locks_to_opt_data(monkeypatch):
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ def client(monkeypatch):
|
|||
previous_auth_required = getattr(web_server.app.state, "auth_required", None)
|
||||
web_server.app.state.auth_required = False
|
||||
test_client = TestClient(web_server.app)
|
||||
test_client.headers[web_server._SESSION_HEADER_NAME] = web_server._SESSION_TOKEN
|
||||
# Loopback bind has no identity gate; no session header needed.
|
||||
try:
|
||||
yield test_client
|
||||
finally:
|
||||
|
|
|
|||
|
|
@ -161,7 +161,7 @@ class TestWebSocketHostOriginGuard:
|
|||
monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True)
|
||||
|
||||
client = TestClient(ws.app)
|
||||
url = f"/api/events?token={ws._SESSION_TOKEN}&channel=security-test"
|
||||
url = "/api/events?channel=security-test"
|
||||
with pytest.raises(WebSocketDisconnect) as exc:
|
||||
with client.websocket_connect(
|
||||
url,
|
||||
|
|
@ -184,7 +184,7 @@ class TestWebSocketHostOriginGuard:
|
|||
monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True)
|
||||
|
||||
client = TestClient(ws.app)
|
||||
url = f"/api/events?token={ws._SESSION_TOKEN}&channel=security-test"
|
||||
url = "/api/events?channel=security-test"
|
||||
with pytest.raises(WebSocketDisconnect) as exc:
|
||||
with client.websocket_connect(
|
||||
url,
|
||||
|
|
@ -206,7 +206,7 @@ class TestWebSocketHostOriginGuard:
|
|||
monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True)
|
||||
|
||||
client = TestClient(ws.app)
|
||||
url = f"/api/events?token={ws._SESSION_TOKEN}&channel=security-test"
|
||||
url = "/api/events?channel=security-test"
|
||||
with client.websocket_connect(
|
||||
url,
|
||||
headers={
|
||||
|
|
|
|||
|
|
@ -43,14 +43,13 @@ def client(monkeypatch, isolated_profiles):
|
|||
|
||||
import hermes_state
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")
|
||||
# The dashboard process's os.environ may carry root-install credentials;
|
||||
# make sure the scoped path never falls back to them.
|
||||
monkeypatch.delenv("TELEGRAM_BOT_TOKEN", raising=False)
|
||||
c = TestClient(app)
|
||||
c.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
return c
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -38,11 +38,10 @@ def client(monkeypatch, isolated_profiles):
|
|||
|
||||
import hermes_state
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")
|
||||
c = TestClient(app)
|
||||
c.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
return c
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -61,11 +61,10 @@ def client(monkeypatch, isolated_profiles):
|
|||
|
||||
import hermes_state
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")
|
||||
c = TestClient(app)
|
||||
c.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
return c
|
||||
|
||||
|
||||
|
|
@ -215,9 +214,7 @@ class TestEditorEndpointsAuth:
|
|||
gated (non-loopback) path is where identity is enforced, covered by
|
||||
the dashboard-auth gate tests.
|
||||
"""
|
||||
from hermes_cli.web_server import _SESSION_HEADER_NAME
|
||||
|
||||
client.headers.pop(_SESSION_HEADER_NAME, None)
|
||||
client.headers.pop("X-Hermes-Session-Token", None)
|
||||
resp = getattr(client, method)(path, **kwargs)
|
||||
assert resp.status_code != 401
|
||||
|
||||
|
|
|
|||
|
|
@ -50,11 +50,10 @@ def client(monkeypatch, isolated_profiles):
|
|||
|
||||
import hermes_state
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")
|
||||
c = TestClient(app)
|
||||
c.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
return c
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue