feat(webui): add browser companion launch

nanobot/channels/websocket/tests/test_websocket_http_routes.py

@@ -253,6 +253,75 @@ async def test_bootstrap_returns_token_for_localhost(
         await server_task
 
 
+@pytest.mark.asyncio
+async def test_browser_companion_launch_uses_private_refreshable_session(
+    bus: MagicMock,
+) -> None:
+    port = _free_port()
+    channel = _ch(bus, port=port, tokenIssueSecret="persistent-secret")
+    server_task = asyncio.create_task(channel.start())
+    try:
+        status = await _http_get(f"http://127.0.0.1:{port}/webui/companion/status")
+        assert status.status_code == 200
+        assert status.json()["ready"] is True
+        assert isinstance(status.json()["version"], str)
+
+        navigation_headers = {
+            "Sec-Fetch-Mode": "navigate",
+            "Sec-Fetch-Dest": "document",
+            "Sec-Fetch-Site": "none",
+        }
+        launch = await _http_get(
+            f"http://127.0.0.1:{port}/webui/companion/open",
+            headers=navigation_headers,
+        )
+        assert launch.status_code == 302
+        assert launch.headers["cache-control"] == "no-store"
+        assert launch.headers["location"] == "/#/"
+        cookie = launch.headers["set-cookie"]
+        assert cookie.startswith(f"nanobot_companion_{port}=nbcs_")
+        assert "HttpOnly" in cookie
+        assert "SameSite=Strict" in cookie
+        companion_cookie = cookie.split(";", 1)[0]
+
+        bootstrap_headers = {"Cookie": companion_cookie}
+        accepted = await _http_get(
+            f"http://127.0.0.1:{port}/webui/bootstrap",
+            headers=bootstrap_headers,
+        )
+        assert accepted.status_code == 200
+
+        refreshed = await _http_get(
+            f"http://127.0.0.1:{port}/webui/bootstrap",
+            headers=bootstrap_headers,
+        )
+        assert refreshed.status_code == 200
+    finally:
+        await channel.stop()
+        await server_task
+
+
+@pytest.mark.asyncio
+async def test_browser_companion_rejects_cross_site_launch(bus: MagicMock) -> None:
+    port = _free_port()
+    channel = _ch(bus, port=port)
+    server_task = asyncio.create_task(channel.start())
+    try:
+        response = await _http_get(
+            f"http://127.0.0.1:{port}/webui/companion/open",
+            headers={
+                "Sec-Fetch-Mode": "navigate",
+                "Sec-Fetch-Dest": "document",
+                "Sec-Fetch-Site": "cross-site",
+            },
+        )
+        assert response.status_code == 403
+        assert channel.gateway.tokens.companion_sessions == {}
+    finally:
+        await channel.stop()
+        await server_task
+
+
 @pytest.mark.asyncio
 async def test_sessions_routes_require_bearer_token(
     bus: MagicMock, tmp_path: Path

nanobot/webui/browser_companion.py

@@ -0,0 +1,25 @@
+"""Security policy for launching the local WebUI from a browser extension."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from nanobot.webui.http_utils import case_insensitive_header
+
+STATUS_PATH = "/webui/companion/status"
+OPEN_PATH = "/webui/companion/open"
+SESSION_COOKIE_PREFIX = "nanobot_companion_"
+SESSION_TTL_SECONDS = 12 * 60 * 60
+
+
+def session_cookie_name(port: int) -> str:
+    """Keep companion sessions isolated when several local WebUIs use different ports."""
+    return f"{SESSION_COOKIE_PREFIX}{port}"
+
+
+def is_top_level_user_navigation(headers: Any) -> bool:
+    """Accept address-bar or extension-created tabs, not cross-site window.open calls."""
+    mode = case_insensitive_header(headers, "Sec-Fetch-Mode").lower()
+    destination = case_insensitive_header(headers, "Sec-Fetch-Dest").lower()
+    site = case_insensitive_header(headers, "Sec-Fetch-Site").lower()
+    return mode == "navigate" and destination == "document" and site == "none"

nanobot/webui/gateway_tokens.py

@@ -17,8 +17,10 @@ class GatewayTokenStore:
     """Own short-lived WebSocket and WebUI API tokens for one gateway process."""
 
     max_tokens: int = 10_000
+    max_companion_sessions: int = 64
     issued_tokens: dict[str, float] = field(default_factory=dict)
     api_tokens: dict[str, float] = field(default_factory=dict)
+    companion_sessions: dict[str, float] = field(default_factory=dict)
 
     def check_api_token(self, request: WsRequest) -> bool:
         self._purge_expired_api_tokens()
@@ -42,6 +44,10 @@ class GatewayTokenStore:
             return False
         return True
 
+    def can_issue_companion_session(self) -> bool:
+        self._purge_expired_companion_sessions()
+        return len(self.companion_sessions) < self.max_companion_sessions
+
     def issue_token(self, ttl_s: int | float) -> str:
         token_value = f"nbwt_{secrets.token_urlsafe(32)}"
         expiry = time.monotonic() + float(ttl_s)
@@ -54,6 +60,11 @@ class GatewayTokenStore:
         self.api_tokens[token_value] = expiry
         return token_value
 
+    def issue_companion_session(self, ttl_s: int | float) -> str:
+        token_value = f"nbcs_{secrets.token_urlsafe(32)}"
+        self.companion_sessions[token_value] = time.monotonic() + float(ttl_s)
+        return token_value
+
     def take_issued_token_if_valid(self, token_value: str | None) -> bool:
         if not token_value:
             return False
@@ -65,9 +76,17 @@ class GatewayTokenStore:
             return False
         return True
 
+    def companion_session_is_valid(self, token_value: str | None) -> bool:
+        if not token_value:
+            return False
+        self._purge_expired_companion_sessions()
+        expiry = self.companion_sessions.get(token_value)
+        return expiry is not None and time.monotonic() <= expiry
+
     def clear(self) -> None:
         self.issued_tokens.clear()
         self.api_tokens.clear()
+        self.companion_sessions.clear()
 
     def _purge_expired_api_tokens(self) -> None:
         now = time.monotonic()
@@ -81,6 +100,12 @@ class GatewayTokenStore:
             if now > expiry:
                 self.issued_tokens.pop(token_key, None)
 
+    def _purge_expired_companion_sessions(self) -> None:
+        now = time.monotonic()
+        for token_key, expiry in list(self.companion_sessions.items()):
+            if now > expiry:
+                self.companion_sessions.pop(token_key, None)
+
 
 def token_response_payload(token: str, expires_in: Any) -> dict[str, Any]:
     return {"token": token, "expires_in": expires_in}

nanobot/webui/http_utils.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import email.utils
 import hmac
 import http
+import http.cookies
 import ipaddress
 import json
 import re
@@ -41,6 +42,20 @@ def case_insensitive_header(headers: Any, key: str) -> str:
     return str(value or "").strip()
 
 
+def request_cookie(headers: Any, name: str) -> str | None:
+    """Read one request cookie without accepting malformed cookie syntax."""
+    raw = case_insensitive_header(headers, "Cookie")
+    if not raw:
+        return None
+    cookies = http.cookies.SimpleCookie()
+    try:
+        cookies.load(raw)
+    except http.cookies.CookieError:
+        return None
+    morsel = cookies.get(name)
+    return morsel.value if morsel else None
+
+
 def safe_host_header(value: str) -> str:
     """Return a safe Host header value, or empty when it should not be echoed."""
     value = value.strip()

nanobot/webui/ws_http.py

@@ -23,12 +23,14 @@ from loguru import logger
 from websockets.http11 import Request as WsRequest
 from websockets.http11 import Response
 
+from nanobot import __version__
 from nanobot.command.builtin import builtin_command_palette
 from nanobot.cron.session_turns import is_bound_cron_job
 from nanobot.cron.types import CronJob, CronSchedule
 from nanobot.runtime_context import public_history_messages
 from nanobot.triggers.local_types import LocalTrigger
 from nanobot.utils.subagent_channel_display import scrub_subagent_messages_for_channel
+from nanobot.webui import browser_companion
 from nanobot.webui.file_preview import (
     WebUIFilePreviewError,
     file_preview_availability_payload,
@@ -71,6 +73,9 @@ from nanobot.webui.http_utils import (
 from nanobot.webui.http_utils import (
     query_first as _query_first,
 )
+from nanobot.webui.http_utils import (
+    request_cookie as _request_cookie,
+)
 from nanobot.webui.http_utils import (
     safe_host_header as _safe_host_header,
 )
@@ -245,6 +250,12 @@ class GatewayHTTPHandler:
         if got == "/webui/bootstrap":
             return self._handle_bootstrap(connection, request)
 
+        if got == browser_companion.STATUS_PATH:
+            return self._handle_companion_status(connection, request)
+
+        if got == browser_companion.OPEN_PATH:
+            return self._handle_companion_open(connection, request)
+
         # Settings routes (delegated)
         response = await self.settings_routes.dispatch(connection, request, got)
         if response is not None:
@@ -319,16 +330,56 @@ class GatewayHTTPHandler:
 
     # -- Bootstrap ----------------------------------------------------------
 
+    def _handle_companion_status(self, connection: Any, request: Any) -> Response:
+        if not _is_local_browser_request(connection, request.headers):
+            return _http_error(403, "companion is localhost-only")
+        return _http_json_response({"ready": True, "version": __version__})
+
+    def _handle_companion_open(self, connection: Any, request: Any) -> Response:
+        if not _is_local_browser_request(connection, request.headers):
+            return _http_error(403, "companion is localhost-only")
+        if not browser_companion.is_top_level_user_navigation(request.headers):
+            return _http_error(403, "companion launch requires a direct browser navigation")
+        cookie_name = browser_companion.session_cookie_name(self.config.port)
+        companion_session = _request_cookie(request.headers, cookie_name)
+        if not self.tokens.companion_session_is_valid(companion_session):
+            if not self.tokens.can_issue_companion_session():
+                return _http_error(429, "too many companion sessions")
+            companion_session = self.tokens.issue_companion_session(
+                browser_companion.SESSION_TTL_SECONDS
+            )
+        cookie = (
+            f"{cookie_name}={companion_session}; "
+            "Path=/webui; HttpOnly; SameSite=Strict"
+        )
+        return _http_response(
+            b"",
+            status=302,
+            extra_headers=[
+                ("Location", "/#/"),
+                ("Set-Cookie", cookie),
+                ("Cache-Control", "no-store"),
+                ("Referrer-Policy", "no-referrer"),
+                ("X-Content-Type-Options", "nosniff"),
+            ],
+        )
+
     def _handle_bootstrap(self, connection: Any, request: Any) -> Response:
         secret = self.config.token_issue_secret.strip() or self.config.token.strip()
         is_local_browser = _is_local_browser_request(connection, request.headers)
-        if secret:
+        cookie_name = browser_companion.session_cookie_name(self.config.port)
+        companion_authenticated = self.tokens.companion_session_is_valid(
+            _request_cookie(request.headers, cookie_name)
+        )
+        if companion_authenticated and not is_local_browser:
+            return _http_error(403, "companion session is localhost-only")
+        if secret and not companion_authenticated:
             if not _issue_route_secret_matches(request.headers, secret):
                 return _http_error(401, "Unauthorized")
-        elif not is_local_browser:
+        elif not secret and not is_local_browser:
             return _http_error(403, "bootstrap is localhost-only")
 
-        api_token_allowed = bool(secret) or is_local_browser
+        api_token_allowed = companion_authenticated or 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"),

tests/webui/test_browser_companion.py

@@ -0,0 +1,30 @@
+from nanobot.webui.browser_companion import is_top_level_user_navigation, session_cookie_name
+from nanobot.webui.gateway_tokens import GatewayTokenStore
+
+
+def test_companion_session_can_refresh_until_expiry() -> None:
+    tokens = GatewayTokenStore()
+    session = tokens.issue_companion_session(30)
+    assert tokens.companion_session_is_valid(session) is True
+    assert tokens.companion_session_is_valid(session) is True
+
+
+def test_companion_session_capacity_is_bounded() -> None:
+    tokens = GatewayTokenStore(max_companion_sessions=1)
+    tokens.issue_companion_session(30)
+    assert tokens.can_issue_companion_session() is False
+
+
+def test_companion_navigation_policy_is_fail_closed() -> None:
+    direct = {
+        "Sec-Fetch-Mode": "navigate",
+        "Sec-Fetch-Dest": "document",
+        "Sec-Fetch-Site": "none",
+    }
+    assert is_top_level_user_navigation(direct) is True
+    assert is_top_level_user_navigation({}) is False
+    assert is_top_level_user_navigation({**direct, "Sec-Fetch-Site": "cross-site"}) is False
+
+
+def test_companion_cookie_is_isolated_by_webui_port() -> None:
+    assert session_cookie_name(8765) != session_cookie_name(8766)