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
9 changes: 9 additions & 0 deletions hermes_cli/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -1531,6 +1531,15 @@ def _ensure_hermes_home_managed(home: Path):
# falls through to request reconstruction rather than breaking
# the login flow.
"public_url": "",
# Extra Host headers to trust when the dashboard is bound to loopback.
# This is the safe path for trusted loopback reverse proxies such as
# Tailscale Serve or a local nginx/Caddy shim: keep ``--host`` on
# 127.0.0.1/localhost/::1, then explicitly allow only the proxy-facing
# hostname(s). This does NOT disable the DNS-rebinding guard and does
# NOT create general proxy trust; it is a narrow operator-supplied
# Host allowlist. CLI: ``hermes dashboard --allowed-host <host>``.
# Env override: ``HERMES_DASHBOARD_ALLOWED_HOSTS=host1,host2``.
"allowed_hosts": [],
},

# Privacy settings
Expand Down
48 changes: 48 additions & 0 deletions hermes_cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -11937,6 +11937,40 @@ def _report_dashboard_status() -> int:
return len(pids)


def _split_dashboard_allowed_hosts(values) -> list[str]:
hosts: list[str] = []
for value in values or []:
if value is None:
continue
for chunk in str(value).split(","):
host = chunk.strip()
if host:
hosts.append(host)
return hosts


def _resolve_dashboard_allowed_hosts(args) -> list[str]:
"""Resolve dashboard Host allowlist from CLI, env, or config."""
from hermes_cli.config import load_config

cli_hosts = _split_dashboard_allowed_hosts(getattr(args, "allowed_host", None))
if cli_hosts:
return cli_hosts

env_hosts = _split_dashboard_allowed_hosts(
[os.environ.get("HERMES_DASHBOARD_ALLOWED_HOSTS", "")]
)
if env_hosts:
return env_hosts

cfg = load_config() or {}
dashboard_cfg = cfg.get("dashboard") or {}
configured = dashboard_cfg.get("allowed_hosts") or []
if isinstance(configured, str):
configured = [configured]
return _split_dashboard_allowed_hosts(configured)


def cmd_dashboard(args):
"""Start the web UI server, or (with --stop/--status) manage running ones."""
# --status: report running dashboards and exit, no deps needed.
Expand Down Expand Up @@ -12025,11 +12059,13 @@ def cmd_dashboard(args):
# The in-browser Chat tab (the embedded TUI over PTY/WebSocket) is always
# available — the desktop app and the dashboard's own Chat tab both rely on
# the `/api/ws` + `/api/pty` sockets, so there is no reason to gate them.
allowed_hosts = _resolve_dashboard_allowed_hosts(args)
start_server(
host=args.host,
port=args.port,
open_browser=not args.no_open,
allow_public=getattr(args, "insecure", False),
allowed_hosts=allowed_hosts,
)


Expand Down Expand Up @@ -15307,6 +15343,18 @@ def cmd_acp(args):
dashboard_parser.add_argument(
"--host", default="127.0.0.1", help="Host (default 127.0.0.1)"
)
dashboard_parser.add_argument(
"--allowed-host",
action="append",
default=[],
help=(
"Extra Host header to trust when bound to loopback. Repeat the flag "
"or pass a comma-separated list. Safe for trusted loopback reverse "
"proxies like Tailscale Serve; does not disable DNS-rebinding "
"protection. Also supports HERMES_DASHBOARD_ALLOWED_HOSTS and "
"dashboard.allowed_hosts."
),
)
dashboard_parser.add_argument(
"--no-open", action="store_true", help="Don't open browser automatically"
)
Expand Down
104 changes: 76 additions & 28 deletions hermes_cli/web_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,52 @@ def _require_token(request: Request) -> None:
})


def _normalize_dashboard_host(value: str) -> str:
"""Normalize a host / netloc / allowlist entry for comparison.

Matching is case-insensitive, strips ports, and understands bracketed
IPv6 notation. Bare IPv6 literals are treated as host-only values.
"""
if not value:
return ""
raw = value.strip()
if not raw:
return ""

if "://" in raw:
return (urllib.parse.urlsplit(raw).hostname or "").lower()

if raw.startswith("["):
close = raw.find("]")
if close != -1:
return raw[1:close].strip().lower()
return raw.strip("[]").lower()

if raw.count(":") >= 2:
return raw.lower()

return (urllib.parse.urlsplit(f"//{raw}").hostname or "").lower()


def _normalize_allowed_dashboard_hosts(
allowed_hosts: list[str] | tuple[str, ...] | set[str] | frozenset[str] | None,
) -> frozenset[str]:
"""Return the operator-supplied dashboard Host allowlist.

This is intentionally a narrow host allowlist, not a broad proxy-trust
toggle. It is consulted only for loopback binds.
"""
if not allowed_hosts:
return frozenset()

normalized = {
host
for host in (_normalize_dashboard_host(value) for value in allowed_hosts)
if host
}
return frozenset(normalized)


def should_require_auth(host: str, allow_public: bool) -> bool:
"""Return True iff the dashboard OAuth auth gate must be active.

Expand All @@ -230,45 +276,37 @@ def should_require_auth(host: str, allow_public: bool) -> bool:
return (host not in _LOOPBACK_HOST_VALUES) and (not allow_public)


def _is_accepted_host(host_header: str, bound_host: str) -> bool:
def _is_accepted_host(
host_header: str,
bound_host: str,
allowed_hosts: list[str] | tuple[str, ...] | set[str] | frozenset[str] | None = None,
) -> bool:
"""True if the Host header targets the interface we bound to.

Accepts:
- Exact bound host (with or without port suffix)
- Loopback aliases when bound to loopback
- Operator-allowed extra hosts when bound to loopback
- Any host when bound to 0.0.0.0 (explicit opt-in to non-loopback,
no protection possible at this layer)
"""
if not host_header:
host_only = _normalize_dashboard_host(host_header)
if not host_only:
return False
# Strip port suffix. IPv6 addresses use bracket notation:
# [::1] — no port
# [::1]:9119 — with port
# Plain hosts/v4:
# localhost:9119
# 127.0.0.1:9119
h = host_header.strip()
if h.startswith("["):
# IPv6 bracketed — port (if any) follows "]:"
close = h.find("]")
if close != -1:
host_only = h[1:close] # strip brackets
else:
host_only = h.strip("[]")
else:
host_only = h.rsplit(":", 1)[0] if ":" in h else h
host_only = host_only.lower()
bound_lc = _normalize_dashboard_host(bound_host)

# 0.0.0.0 bind means operator explicitly opted into all-interfaces
# (requires --insecure per web_server.start_server). No Host-layer
# defence can protect that mode; rely on operator network controls.
if bound_host in {"0.0.0.0", "::"}:
if bound_lc in {"0.0.0.0", "::"}:
return True

# Loopback bind: accept the loopback names
bound_lc = bound_host.lower()
# Loopback bind: accept the loopback names plus the operator-supplied
# allowlist for trusted loopback reverse proxies such as Tailscale Serve.
if bound_lc in _LOOPBACK_HOST_VALUES:
return host_only in _LOOPBACK_HOST_VALUES
accepted_hosts = set(_LOOPBACK_HOST_VALUES)
accepted_hosts.update(_normalize_allowed_dashboard_hosts(allowed_hosts))
return host_only in accepted_hosts

# Explicit non-loopback bind: require exact host match
return host_only == bound_lc
Expand All @@ -291,13 +329,15 @@ async def host_header_middleware(request: Request, call_next):
bound_host = getattr(app.state, "bound_host", None)
if bound_host:
host_header = request.headers.get("host", "")
if not _is_accepted_host(host_header, bound_host):
allowed_hosts = getattr(app.state, "allowed_hosts", frozenset())
if not _is_accepted_host(host_header, bound_host, allowed_hosts):
return JSONResponse(
status_code=400,
content={
"detail": (
"Invalid Host header. Dashboard requests must use "
"the hostname the server was bound to."
"the hostname the server was bound to or an "
"operator-allowed host."
),
},
)
Expand Down Expand Up @@ -7092,7 +7132,8 @@ def _ws_host_origin_reason(ws: "WebSocket") -> Optional[str]:
return None

host_header = ws.headers.get("host", "")
if not _is_accepted_host(host_header, bound_host):
allowed_hosts = getattr(app.state, "allowed_hosts", frozenset())
if not _is_accepted_host(host_header, bound_host, allowed_hosts):
return f"host_mismatch host={host_header or '?'} bound={bound_host}"

origin = ws.headers.get("origin", "")
Expand All @@ -7109,7 +7150,7 @@ def _ws_host_origin_reason(ws: "WebSocket") -> Optional[str]:
if not parsed.netloc:
return f"origin_mismatch origin={origin} bound={bound_host}"

if not _is_accepted_host(parsed.netloc, bound_host):
if not _is_accepted_host(parsed.netloc, bound_host, allowed_hosts):
return f"origin_mismatch origin={origin} bound={bound_host}"
return None

Expand All @@ -7125,7 +7166,6 @@ def _ws_host_origin_is_allowed(ws: "WebSocket") -> bool:
"""
return _ws_host_origin_reason(ws) is None


def _ws_request_reason(ws: "WebSocket") -> Optional[str]:
"""First Host/Origin or peer-IP rejection reason, or None when allowed."""
return _ws_host_origin_reason(ws) or _ws_client_reason(ws)
Expand Down Expand Up @@ -8681,6 +8721,8 @@ def start_server(
port: int = 9119,
open_browser: bool = True,
allow_public: bool = False,
*,
allowed_hosts: list[str] | tuple[str, ...] | set[str] | frozenset[str] | None = None,
):
"""Start the web UI server."""
import uvicorn
Expand Down Expand Up @@ -8755,6 +8797,12 @@ def start_server(
# PTY child uses to publish events to the dashboard sidebar.
app.state.bound_host = host
app.state.bound_port = port
app.state.allowed_hosts = _normalize_allowed_dashboard_hosts(allowed_hosts)
if app.state.allowed_hosts and _normalize_dashboard_host(host) in _LOOPBACK_HOST_VALUES:
_log.info(
"Dashboard loopback Host allowlist enabled for: %s",
", ".join(sorted(app.state.allowed_hosts)),
)

if open_browser:
import webbrowser
Expand Down
46 changes: 46 additions & 0 deletions tests/hermes_cli/test_dashboard_auth_gate.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
pytestmark = pytest.mark.xdist_group("dashboard_auth_app_state")
from fastapi.testclient import TestClient

from hermes_cli import main
from hermes_cli import web_server


Expand Down Expand Up @@ -132,6 +133,25 @@ def test_start_server_loopback_sets_auth_required_false(monkeypatch):
assert web_server.app.state.auth_required is False


def test_start_server_loopback_normalizes_allowed_hosts(monkeypatch):
"""Loopback allowlist entries are normalized once at server startup."""
_stub_uvicorn_run(monkeypatch)
web_server.start_server(
host="127.0.0.1",
port=9119,
open_browser=False,
allow_public=False,
allowed_hosts=[
"Hermes-Agent-VPS.Tail88A68B.TS.Net:443",
"[::1]:9119",
],
)
assert web_server.app.state.allowed_hosts == frozenset({
"hermes-agent-vps.tail88a68b.ts.net",
"::1",
})


def test_start_server_insecure_public_sets_auth_required_false(monkeypatch):
"""``--insecure`` (allow_public=True) on a public host: gate stays OFF."""
_stub_uvicorn_run(monkeypatch)
Expand All @@ -143,6 +163,32 @@ def test_start_server_insecure_public_sets_auth_required_false(monkeypatch):
assert web_server.app.state.auth_required is False


def test_resolve_dashboard_allowed_hosts_prefers_cli_then_env_then_config(monkeypatch):
from argparse import Namespace
from hermes_cli import config as hermes_config

monkeypatch.setattr(
hermes_config,
"load_config",
lambda: {"dashboard": {"allowed_hosts": ["cfg.example", "cfg2.example:443"]}},
)
monkeypatch.setenv("HERMES_DASHBOARD_ALLOWED_HOSTS", "env.example,env2.example:443")

assert main._resolve_dashboard_allowed_hosts(
Namespace(allowed_host=["cli.example,cli2.example:443"])
) == ["cli.example", "cli2.example:443"]
assert main._resolve_dashboard_allowed_hosts(Namespace(allowed_host=[])) == [
"env.example",
"env2.example:443",
]

monkeypatch.delenv("HERMES_DASHBOARD_ALLOWED_HOSTS")
assert main._resolve_dashboard_allowed_hosts(Namespace(allowed_host=[])) == [
"cfg.example",
"cfg2.example:443",
]


def test_start_server_public_without_insecure_records_auth_required(monkeypatch):
"""Public bind without --insecure: the gate engages and auth_required=True.

Expand Down
Loading