fix(dashboard-auth): bypass loopback WS peer check in gated mode

When the OAuth gate is active, start_server runs uvicorn with
proxy_headers=True so the dashboard can honour X-Forwarded-Proto from
Fly's TLS terminator (cookies, redirect URI reconstruction). A side
effect: ws.client.host is rewritten to the X-Forwarded-For value, which
on Fly is the real internet client IP — never loopback. The loopback
peer guard in _ws_client_is_allowed then rejected every WS upgrade in
gated mode (4403 close) even after a successful OAuth round trip and
ticket consumption, silently breaking /api/pty, /api/ws, /api/pub, and
/api/events.

Fix: in gated mode, bypass the peer-IP check. The OAuth gate +
single-use ticket is the auth. The Host/Origin guard in
_ws_host_origin_is_allowed still runs and is what protects against
DNS-rebinding here, not the peer IP.

Loopback mode behaviour is unchanged: the legacy ?token= path is the
only auth there and we don't want LAN hosts guessing tokens.

Regression coverage: TestWsRequestIsAllowedGated pins all four
behaviours — non-loopback peer allowed in gated mode, non-loopback peer
rejected in loopback mode, loopback peer allowed in loopback mode, and
the Host/Origin guard still firing on a rebinding attempt with gated
mode + matching peer.
This commit is contained in:
Ben 2026-05-26 09:20:08 +10:00 committed by Teknium
parent 866cc988b5
commit c3104195b8
2 changed files with 61 additions and 1 deletions

View file

@ -3376,8 +3376,20 @@ _LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1", "localhost", "testclient"})
def _ws_client_is_allowed(ws: "WebSocket") -> bool:
"""Check if the WebSocket client IP is acceptable.
Allows loopback clients only.
Loopback mode: 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.
Gated mode: any peer is allowed uvicorn's ``proxy_headers=True``
(enabled when the OAuth gate is active so cookies can pick up
``X-Forwarded-Proto``) rewrites ``ws.client.host`` to the
X-Forwarded-For value, which is the real internet client IP. The
OAuth gate + single-use ``?ticket=`` is the auth at that point; the
Host/Origin guard in :func:`_ws_host_origin_is_allowed` is what
blocks DNS-rebinding here, not the peer IP.
"""
if getattr(app.state, "auth_required", False):
return True
client_host = ws.client.host if ws.client else ""
if not client_host:
return True

View file

@ -244,6 +244,54 @@ class TestWsAuthOkGated:
# ---------------------------------------------------------------------------
class TestWsRequestIsAllowedGated:
"""Bug fix: in gated mode, the WS peer-IP loopback check must be
bypassed.
When the OAuth gate is active, ``start_server`` runs uvicorn with
``proxy_headers=True`` so the dashboard can honour
``X-Forwarded-Proto`` from Fly's TLS terminator. A side effect is that
``ws.client.host`` is rewritten to the X-Forwarded-For value the
real internet client IP, never loopback. The loopback peer guard
(intended only for unauthenticated loopback dev) must not also reject
those upgrades: the OAuth gate + single-use ticket is the auth.
Regression coverage: every WS endpoint (``/api/pty``, ``/api/ws``,
``/api/pub``, ``/api/events``) calls ``_ws_request_is_allowed`` after
``_ws_auth_ok``. If the peer-IP check rejects gated mode, the chat
tab + sidebar tool feed silently fail to connect even after a
successful OAuth login.
"""
def test_non_loopback_peer_allowed_in_gated_mode(self, gated_app):
ws = _fake_ws(query={}, client_host="203.0.113.7")
# Host header matches the bound host so the DNS-rebinding guard
# passes; only the peer-IP check is under test.
ws.headers = {"host": "fly-app.fly.dev"}
assert web_server._ws_request_is_allowed(ws) is True
def test_non_loopback_peer_rejected_in_loopback_mode(self, loopback_app):
"""Loopback mode still enforces the peer-IP guard — the legacy
token path is the only auth and we don't want random LAN hosts
guessing it."""
ws = _fake_ws(query={}, client_host="192.168.1.42")
ws.headers = {"host": "127.0.0.1:8080"}
assert web_server._ws_request_is_allowed(ws) is False
def test_loopback_peer_allowed_in_loopback_mode(self, loopback_app):
ws = _fake_ws(query={}, client_host="127.0.0.1")
ws.headers = {"host": "127.0.0.1:8080"}
assert web_server._ws_request_is_allowed(ws) is True
def test_host_origin_guard_still_runs_in_gated_mode(self, gated_app):
"""Bypassing the peer-IP check must not bypass the DNS-rebinding
Host header guard that one still protects against attacker
sites resolving DNS to the public IP."""
ws = _fake_ws(query={}, client_host="203.0.113.7")
ws.headers = {"host": "evil.example.com"}
assert web_server._ws_request_is_allowed(ws) is False
class TestSidecarUrl:
def test_loopback_uses_session_token(self, loopback_app):
url = web_server._build_sidecar_url("ch-1")