feat(webui): bypass tokens for trusted proxy auth

nanobot/channels/websocket/runtime.py

@@ -61,6 +61,9 @@ from nanobot.session.webui_turns import (
 from nanobot.webui.cli_apps_api import normalize_cli_app_mentions
 from nanobot.webui.forking import handle_webui_fork_chat
 from nanobot.webui.gateway_services import GatewayServices
+from nanobot.webui.http_utils import (
+    is_trusted_proxy_authenticated_request as _is_trusted_proxy_authenticated_request,
+)
 from nanobot.webui.http_utils import (
     normalize_config_path as _normalize_config_path,
 )
@@ -220,11 +223,11 @@ class WebSocketConfig(Base):
     def wildcard_host_requires_auth(self) -> Self:
         if self.host not in ("0.0.0.0", "::"):
             return self
-        if self.token.strip() or self.token_issue_secret.strip():
+        if self.token.strip() or self.token_issue_secret.strip() or self.trusted_proxy_auth is not None:
             return self
         raise ValueError(
-            "host is 0.0.0.0 (all interfaces) but neither token nor "
-            "token_issue_secret is set — set one to prevent unauthenticated access"
+            "host is 0.0.0.0 (all interfaces) but neither token, token_issue_secret, "
+            "nor trusted_proxy_auth is set — set one to prevent unauthenticated access"
         )
 
 
@@ -480,16 +483,16 @@ class WebSocketChannel(BaseChannel):
     async def _dispatch_http(self, connection: ServerConnection, request: WsRequest) -> Any:
         """Route an inbound HTTP request to the HTTP handler or WS upgrade."""
         got, query = _parse_request_path(request.path)
+        expected_ws = self._expected_path()
 
         # WebSocket upgrade — channel handles this itself
-        expected_ws = self._expected_path()
         if got == expected_ws and _is_websocket_upgrade(request):
             client_id = _query_first(query, "client_id") or ""
             if len(client_id) > 128:
                 client_id = client_id[:128]
             if not self.is_allowed(client_id):
                 return connection.respond(403, "Forbidden")
-            return self._authorize_websocket_handshake(connection, query)
+            return self._authorize_websocket_handshake(connection, query, request.headers)
 
         # Everything else goes to the HTTP handler
         return await self._http_router.dispatch(connection, request)
@@ -498,7 +501,12 @@ class WebSocketChannel(BaseChannel):
         self,
         connection: ServerConnection,
         query: dict[str, list[str]],
+        headers: Any = None,
     ) -> Any:
+        if _is_trusted_proxy_authenticated_request(connection, headers or {}, self.config):
+            self._webui_connections.add(connection)
+            return None
+
         supplied = _query_first(query, "token")
         static_token = self.config.token.strip()
 

nanobot/channels/websocket/tests/test_websocket_http_routes.py

@@ -3351,7 +3351,7 @@ def test_trusted_proxy_rejects_untrusted_peer_spoof(bus: MagicMock) -> None:
     assert resp.status_code == 403
 
 
-def test_trusted_proxy_accepts_assertion_without_forwarded_header_trust(
+def test_trusted_proxy_bootstrap_has_no_tokens(
     bus: MagicMock,
 ) -> None:
     assertion = "opaque-upstream-assertion"
@@ -3376,9 +3376,36 @@ def test_trusted_proxy_accepts_assertion_without_forwarded_header_trust(
     assert assertion not in body
     assert assertion not in repr(log.mock_calls)
     payload = json.loads(body)
-    assert payload["token"].startswith("nbwt_")
-    assert payload["api_token"].startswith("nbwt_")
-    assert payload["api_token"] != payload["token"]
+    assert "token" not in payload
+    assert "api_token" not in payload
+    assert payload["ws_path"] == "/"
+
+
+@pytest.mark.asyncio
+async def test_trusted_proxy_authorizes_rest_without_api_token(bus: MagicMock) -> None:
+    channel = _ch(bus, **_trusted_proxy_config())
+    response = await channel.gateway.http.dispatch(
+        _LOCAL,
+        _FakeReq(
+            {
+                "Host": "nanobot.example",
+                "Cf-Access-Jwt-Assertion": "present",
+            },
+            path="/api/sessions",
+        ),
+    )
+    assert response.status_code == 503
+
+
+def test_trusted_proxy_authorizes_websocket_without_token(bus: MagicMock) -> None:
+    channel = _ch(bus, **_trusted_proxy_config())
+    response = channel._authorize_websocket_handshake(
+        _LOCAL,
+        {},
+        {"Cf-Access-Jwt-Assertion": "present"},
+    )
+    assert response is None
+    assert _LOCAL in channel._webui_connections
 
 
 def test_forwarding_headers_alone_never_authorize_bootstrap(bus: MagicMock) -> None:
@@ -3397,7 +3424,7 @@ def test_forwarding_headers_alone_never_authorize_bootstrap(bus: MagicMock) -> N
     assert resp.status_code == 403
 
 
-def test_trusted_proxy_does_not_override_bootstrap_secret(bus: MagicMock) -> None:
+def test_trusted_proxy_bypasses_bootstrap_secret_and_tokens(bus: MagicMock) -> None:
     channel = _ch(
         bus,
         tokenIssueSecret="route-secret",
@@ -3407,7 +3434,10 @@ def test_trusted_proxy_does_not_override_bootstrap_secret(bus: MagicMock) -> Non
         _LOCAL,
         _FakeReq({"Cf-Access-Jwt-Assertion": "present"}),
     )
-    assert resp.status_code == 401
+    assert resp.status_code == 200
+    payload = json.loads(resp.body)
+    assert "token" not in payload
+    assert "api_token" not in payload
 
 
 @pytest.mark.parametrize(
@@ -3461,6 +3491,11 @@ def test_wildcard_host_with_secret_is_valid(bus: MagicMock) -> None:
     assert channel.config.host == "0.0.0.0"
 
 
+def test_wildcard_host_with_trusted_proxy_auth_is_valid(bus: MagicMock) -> None:
+    channel = _ch(bus, host="0.0.0.0", **_trusted_proxy_config())
+    assert channel.config.host == "0.0.0.0"
+
+
 def test_wildcard_ipv6_without_auth_raises(bus: MagicMock) -> None:
     import pytest
     from pydantic_core import ValidationError

nanobot/webui/ws_http.py

@@ -266,6 +266,8 @@ class GatewayHTTPHandler:
     # -- Token management ---------------------------------------------------
 
     def check_api_token(self, request: WsRequest) -> bool:
+        if getattr(request, "_nanobot_trusted_proxy_authenticated", False):
+            return True
         return self.tokens.check_api_token(request)
 
     # -- Main dispatch ------------------------------------------------------
@@ -275,6 +277,11 @@ class GatewayHTTPHandler:
         got, _ = _parse_request_path(request.path)
         started = time.perf_counter()
         response: Any | None = None
+        setattr(
+            request,
+            "_nanobot_trusted_proxy_authenticated",
+            _is_trusted_proxy_authenticated_request(connection, request.headers, self.config),
+        )
 
         try:
             response = await self._dispatch_resolved(connection, request, got)
@@ -380,13 +387,27 @@ class GatewayHTTPHandler:
             request.headers,
             self.config,
         )
-        if secret:
-            if not _issue_route_secret_matches(request.headers, secret):
-                return _http_error(401, "Unauthorized")
-        elif not (is_local_browser or is_proxy_authenticated):
-            return _http_error(403, "bootstrap is localhost-only")
+        if not is_proxy_authenticated:
+            if secret:
+                if not _issue_route_secret_matches(request.headers, secret):
+                    return _http_error(401, "Unauthorized")
+            elif not is_local_browser:
+                return _http_error(403, "bootstrap is localhost-only")
 
-        api_token_allowed = bool(secret) or is_local_browser or is_proxy_authenticated
+        if is_proxy_authenticated:
+            payload = {
+                "ws_path": _normalize_config_path(self.config.path),
+                "ws_url": self._bootstrap_ws_url(request),
+                "limits": self.ingress.bootstrap_limits(
+                    max_frame_bytes=self.config.max_message_bytes,
+                ),
+                "model_name": _resolve_bootstrap_model_name(self.runtime_model_name),
+                "runtime_surface": self._runtime_surface,
+                "runtime_capabilities": self._capabilities,
+            }
+            return _http_json_response(payload)
+
+        api_token_allowed = bool(secret) or is_local_browser
         if not self.tokens.can_issue(include_api_token=api_token_allowed):
             return _http_response(
                 json.dumps({"error": "too many outstanding tokens"}).encode("utf-8"),