From d257e5f36cfa9328600c8bde6f0de09a6ad9b6f4 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Tue, 30 Jun 2026 06:21:04 +0000 Subject: [PATCH] fix(security): gate embedded-terminal endpoints to local origins when auth disabled MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The embedded workspace terminal spawns a PTY shell that runs arbitrary commands as the server-process user. check_auth() returns True unconditionally when no password/passkey is configured (the default out-of-the-box state), so without a network-scope gate the terminal endpoints were reachable by any unauthenticated caller able to hit the port — which on a passwordless public bind is remote code execution. Apply the same local-origin gate the onboarding/bootstrap endpoints use (_onboarding_gate_allows) to /api/terminal/{start,input,resize,close} and /api/terminal/output: with auth disabled, accept only loopback/private origins, ignore spoofable X-Forwarded-For/X-Real-IP unless HERMES_WEBUI_TRUST_FORWARDED_FOR=1, and honor HERMES_WEBUI_ONBOARDING_OPEN=1 as the explicit opt-out for a deliberately-exposed server. Auth-enabled servers (cookie already verified upstream) and genuine same-host clients are unaffected. Also fixes a latent test-isolation leak in test_extension_route_remains_behind_webui_auth: it set HERMES_WEBUI_PASSWORD but never invalidated the process-wide password-hash cache, so its result depended on suite execution order (exposed when the new test file shifted ordering). Invalidate before+after so it reads the env var deterministically. 12 new gate tests in tests/test_cvd3_terminal_local_origin_gate.py. --- api/routes.py | 35 +++ tests/test_cvd3_terminal_local_origin_gate.py | 202 ++++++++++++++++++ tests/test_embedded_workspace_terminal.py | 5 + tests/test_extension_hooks.py | 33 +-- 4 files changed, 263 insertions(+), 12 deletions(-) create mode 100644 tests/test_cvd3_terminal_local_origin_gate.py diff --git a/api/routes.py b/api/routes.py index 92466731..01579c6d 100644 --- a/api/routes.py +++ b/api/routes.py @@ -5079,6 +5079,31 @@ def _onboarding_gate_allows(handler, auth_enabled: bool | None = None) -> bool: return _onboarding_request_is_local(handler) +# Operator-facing copy reused by every embedded-terminal endpoint refusal. +_EMBEDDED_TERMINAL_GATE_DENIED_MESSAGE = ( + "Embedded terminal is only available from local networks when authentication " + "is not configured. Configure a password/passkey, or set " + "HERMES_WEBUI_ONBOARDING_OPEN=1 to allow it on a deliberately-exposed server." +) + + +def _embedded_terminal_gate_allows(handler) -> bool: + """Local-origin gate for the embedded-terminal endpoints. + + The embedded terminal spawns a PTY shell that runs arbitrary commands as the + server-process user, so admitting an unauthenticated remote caller is remote + code execution. When auth is enabled, ``check_auth()`` has already verified + the session cookie before the request reaches these handlers, so this returns + True. When auth is DISABLED (the default out-of-the-box state) ``check_auth()`` + admits every caller unconditionally, so restrict the terminal to local/private + origins — the same trust model the onboarding/bootstrap endpoints use, ignoring + spoofable forwarded headers unless an operator has opted into trusting them. + A deliberately-exposed passwordless server (access secured at another layer) + opts out with ``HERMES_WEBUI_ONBOARDING_OPEN=1``. + """ + return _onboarding_gate_allows(handler) + + def _csp_report_rate_limited(handler, *, now: float | None = None) -> bool: now = time.time() if now is None else now key = _client_ip_for_rate_limit(handler) @@ -14952,6 +14977,8 @@ def _terminal_remote_backend_enabled() -> bool: def _handle_terminal_start(handler, body): try: + if not _embedded_terminal_gate_allows(handler): + return bad(handler, _EMBEDDED_TERMINAL_GATE_DENIED_MESSAGE, 403) sid, session = _terminal_session_lookup(body) if _terminal_remote_backend_enabled(): return j( @@ -14990,6 +15017,8 @@ def _handle_terminal_start(handler, body): def _handle_terminal_input(handler, body): try: + if not _embedded_terminal_gate_allows(handler): + return bad(handler, _EMBEDDED_TERMINAL_GATE_DENIED_MESSAGE, 403) require(body, "session_id") data = str(body.get("data", "")) if len(data) > 8192: @@ -15007,6 +15036,8 @@ def _handle_terminal_input(handler, body): def _handle_terminal_resize(handler, body): try: + if not _embedded_terminal_gate_allows(handler): + return bad(handler, _EMBEDDED_TERMINAL_GATE_DENIED_MESSAGE, 403) require(body, "session_id") from api.terminal import resize_terminal resize_terminal( @@ -15025,6 +15056,8 @@ def _handle_terminal_resize(handler, body): def _handle_terminal_close(handler, body): try: + if not _embedded_terminal_gate_allows(handler): + return bad(handler, _EMBEDDED_TERMINAL_GATE_DENIED_MESSAGE, 403) require(body, "session_id") from api.terminal import close_terminal closed = close_terminal(body["session_id"]) @@ -15034,6 +15067,8 @@ def _handle_terminal_close(handler, body): def _handle_terminal_output(handler, parsed): + if not _embedded_terminal_gate_allows(handler): + return bad(handler, _EMBEDDED_TERMINAL_GATE_DENIED_MESSAGE, 403) qs = parse_qs(parsed.query) sid = qs.get("session_id", [""])[0] if not sid: diff --git a/tests/test_cvd3_terminal_local_origin_gate.py b/tests/test_cvd3_terminal_local_origin_gate.py new file mode 100644 index 00000000..7570cd44 --- /dev/null +++ b/tests/test_cvd3_terminal_local_origin_gate.py @@ -0,0 +1,202 @@ +"""Tests: embedded-terminal endpoints are gated to local origins when auth is disabled. + +CVD #3 (A2): the embedded terminal endpoints (`/api/terminal/start|input|resize|close` +and `/api/terminal/output`) spawn / drive a PTY shell that runs arbitrary commands as the +server-process user. `check_auth()` returns True unconditionally when authentication is not +configured (the default out-of-the-box state), so without a network-scope gate ANY caller +able to reach the port — including an unauthenticated remote attacker on a passwordless +public bind — could obtain remote code execution. + +These tests pin the local-origin gate (`_embedded_terminal_gate_allows`, mirroring the +onboarding/bootstrap trust model) on every terminal handler: public clients are refused with +403 when auth is disabled; loopback/private clients and auth-enabled sessions are admitted; +spoofed forwarded headers do not establish locality; and the explicit +HERMES_WEBUI_ONBOARDING_OPEN escape hatch is honored. +""" + +import io +from types import SimpleNamespace + + +class _Headers(dict): + def get(self, key, default=None): + for k, v in self.items(): + if k.lower() == key.lower(): + return v + return default + + +class _Handler: + def __init__(self, *, client_ip="8.8.8.8", headers=None, body=b"{}"): + self.client_address = (client_ip, 12345) + self.headers = _Headers(headers or {}) + self.rfile = io.BytesIO(body) + self.wfile = io.BytesIO() + self.request = None + self.status = None + self.sent_headers = [] + + def send_response(self, code): + self.status = code + + def send_header(self, key, value): + self.sent_headers.append((key, value)) + + def end_headers(self): + pass + + +def _no_auth(monkeypatch): + """Default out-of-the-box state: no password, no passkey, no opt-outs.""" + monkeypatch.setattr("api.auth.is_auth_enabled", lambda: False) + monkeypatch.delenv("HERMES_WEBUI_ONBOARDING_OPEN", raising=False) + monkeypatch.delenv("HERMES_WEBUI_TRUST_FORWARDED_FOR", raising=False) + + +# -------------------------------------------------------------------------- +# Gate predicate +# -------------------------------------------------------------------------- + +def test_terminal_gate_blocks_public_client_when_auth_disabled(monkeypatch): + from api import routes + + _no_auth(monkeypatch) + handler = _Handler(client_ip="8.8.8.8", headers={}) + assert routes._embedded_terminal_gate_allows(handler) is False + + +def test_terminal_gate_allows_loopback_client_when_auth_disabled(monkeypatch): + from api import routes + + _no_auth(monkeypatch) + handler = _Handler(client_ip="127.0.0.1", headers={}) + assert routes._embedded_terminal_gate_allows(handler) is True + + +def test_terminal_gate_allows_private_client_when_auth_disabled(monkeypatch): + """Docker bridge / LAN (no forwarded header present) is treated as local.""" + from api import routes + + _no_auth(monkeypatch) + handler = _Handler(client_ip="172.17.0.1", headers={}) + assert routes._embedded_terminal_gate_allows(handler) is True + + +def test_terminal_gate_ignores_spoofed_forwarded_header_from_public_socket(monkeypatch): + """A public socket spoofing X-Forwarded-For: 127.0.0.1 must NOT pass the gate.""" + from api import routes + + _no_auth(monkeypatch) + handler = _Handler(client_ip="8.8.8.8", headers={"X-Forwarded-For": "127.0.0.1"}) + assert routes._embedded_terminal_gate_allows(handler) is False + + +def test_terminal_gate_allows_any_client_when_auth_enabled(monkeypatch): + """With auth enabled, check_auth() already verified the cookie upstream.""" + from api import routes + + monkeypatch.setattr("api.auth.is_auth_enabled", lambda: True) + monkeypatch.delenv("HERMES_WEBUI_ONBOARDING_OPEN", raising=False) + handler = _Handler(client_ip="8.8.8.8", headers={}) + assert routes._embedded_terminal_gate_allows(handler) is True + + +def test_terminal_gate_honors_onboarding_open_escape_hatch(monkeypatch): + """Deliberately-exposed passwordless server (secured elsewhere) opts out.""" + from api import routes + + monkeypatch.setattr("api.auth.is_auth_enabled", lambda: False) + monkeypatch.setenv("HERMES_WEBUI_ONBOARDING_OPEN", "1") + handler = _Handler(client_ip="8.8.8.8", headers={}) + assert routes._embedded_terminal_gate_allows(handler) is True + + +# -------------------------------------------------------------------------- +# Handler dispatch — the RCE-bearing endpoints refuse public clients (403) +# without ever reaching the PTY-spawning code. +# -------------------------------------------------------------------------- + +def test_terminal_start_refuses_public_client_without_spawning(monkeypatch): + from api import routes + + _no_auth(monkeypatch) + + spawned = {"called": False} + # If the gate fails to block, this would be the RCE primitive — assert it + # is never reached. _terminal_session_lookup also must not run. + monkeypatch.setattr( + routes, "_terminal_session_lookup", + lambda body: spawned.__setitem__("called", True) or ("s", SimpleNamespace(workspace="")), + ) + + handler = _Handler(client_ip="8.8.8.8", body=b'{"session_id":"s"}') + routes._handle_terminal_start(handler, {"session_id": "s"}) + + assert handler.status == 403 + assert spawned["called"] is False + + +def test_terminal_input_refuses_public_client_without_writing(monkeypatch): + from api import routes + + _no_auth(monkeypatch) + + wrote = {"called": False} + import api.terminal as term_mod + monkeypatch.setattr( + term_mod, "write_terminal", + lambda sid, data: wrote.__setitem__("called", True), + ) + + handler = _Handler(client_ip="8.8.8.8") + routes._handle_terminal_input(handler, {"session_id": "s", "data": "id\n"}) + + assert handler.status == 403 + assert wrote["called"] is False + + +def test_terminal_output_refuses_public_client(monkeypatch): + from api import routes + + _no_auth(monkeypatch) + handler = _Handler(client_ip="8.8.8.8") + routes._handle_terminal_output(handler, SimpleNamespace(path="/api/terminal/output", query="session_id=s")) + assert handler.status == 403 + + +def test_terminal_close_refuses_public_client(monkeypatch): + from api import routes + + _no_auth(monkeypatch) + handler = _Handler(client_ip="8.8.8.8") + routes._handle_terminal_close(handler, {"session_id": "s"}) + assert handler.status == 403 + + +def test_terminal_resize_refuses_public_client(monkeypatch): + from api import routes + + _no_auth(monkeypatch) + handler = _Handler(client_ip="8.8.8.8") + routes._handle_terminal_resize(handler, {"session_id": "s", "rows": 24, "cols": 80}) + assert handler.status == 403 + + +def test_terminal_start_loopback_client_passes_gate(monkeypatch): + """A genuine same-host client clears the gate and proceeds to normal lookup.""" + from api import routes + + _no_auth(monkeypatch) + + reached = {"called": False} + monkeypatch.setattr( + routes, "_terminal_session_lookup", + lambda body: reached.__setitem__("called", True) or (_ for _ in ()).throw(KeyError("Session not found")), + ) + + handler = _Handler(client_ip="127.0.0.1", body=b'{"session_id":"s"}') + routes._handle_terminal_start(handler, {"session_id": "s"}) + + # Gate passed → lookup ran → 404 (no such session), NOT 403. + assert reached["called"] is True + assert handler.status == 404 diff --git a/tests/test_embedded_workspace_terminal.py b/tests/test_embedded_workspace_terminal.py index 808ec20a..c816008d 100644 --- a/tests/test_embedded_workspace_terminal.py +++ b/tests/test_embedded_workspace_terminal.py @@ -268,6 +268,11 @@ def test_terminal_button_and_start_path_respect_remote_backend_guard(): class _RouteHandler: def __init__(self): self.headers = {} + # Real terminal requests come from a same-host browser; present as a + # genuine loopback client so the embedded-terminal local-origin gate + # (CVD #3 / test_cvd3_terminal_local_origin_gate.py) admits the request + # and these tests exercise the terminal logic they target. + self.client_address = ("127.0.0.1", 12345) self.wfile = io.BytesIO() self.responses = [] diff --git a/tests/test_extension_hooks.py b/tests/test_extension_hooks.py index 515b5442..5ec55ce1 100644 --- a/tests/test_extension_hooks.py +++ b/tests/test_extension_hooks.py @@ -194,20 +194,29 @@ def test_extension_settings_only_manifest_still_injects_runtime_config(tmp_path, def test_extension_route_remains_behind_webui_auth(monkeypatch): monkeypatch.setenv("HERMES_WEBUI_PASSWORD", "test-password") - from api.auth import check_auth + from api.auth import check_auth, _invalidate_password_hash_cache - extension = FakeHandler() - # SimpleNamespace must include `query` because api.auth.check_auth (since - # v0.50.258, the multi-param ?next= encoding fix) accesses `parsed.query` - # when constructing the redirect Location header. - assert check_auth(extension, SimpleNamespace(path="/extensions/app.js", query="")) is False - assert extension.status == 302 - assert extension.header("Location") == "login?next=/extensions/app.js" + # The password hash is cached process-wide (PBKDF2 is ~1s/call). Invalidate + # before so this test reads the just-set env var rather than a stale None + # cached by an earlier auth-disabled test, and again in finally so the hash + # computed here can't leak into a later test that expects auth disabled. + # Without this the test's result depends on suite execution order. + _invalidate_password_hash_cache() + try: + extension = FakeHandler() + # SimpleNamespace must include `query` because api.auth.check_auth (since + # v0.50.258, the multi-param ?next= encoding fix) accesses `parsed.query` + # when constructing the redirect Location header. + assert check_auth(extension, SimpleNamespace(path="/extensions/app.js", query="")) is False + assert extension.status == 302 + assert extension.header("Location") == "login?next=/extensions/app.js" - # Existing core static assets remain public; extension assets intentionally - # do not share that exemption because they are administrator-supplied code. - static = FakeHandler() - assert check_auth(static, SimpleNamespace(path="/static/ui.js", query="")) is True + # Existing core static assets remain public; extension assets intentionally + # do not share that exemption because they are administrator-supplied code. + static = FakeHandler() + assert check_auth(static, SimpleNamespace(path="/static/ui.js", query="")) is True + finally: + _invalidate_password_hash_cache()