fix(security): gate embedded-terminal endpoints to local origins when auth disabled

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

@@ -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:

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

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 = []
 

tests/test_extension_hooks.py

@@ -194,8 +194,15 @@ 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
 
+    # 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`
@@ -208,6 +215,8 @@ def test_extension_route_remains_behind_webui_auth(monkeypatch):
         # 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()