diff --git a/gateway/browser_control_broker.py b/gateway/browser_control_broker.py new file mode 100644 index 0000000000..98e5dcdb28 --- /dev/null +++ b/gateway/browser_control_broker.py @@ -0,0 +1,619 @@ +"""Transport-neutral browser-control broker core (Phase 4). + +This module is the in-process heart of the browser-control feature: it binds +an *identity-scoped controller* (the party that physically drives a browser) +to *callers* (agents talking to that browser over any transport) without the +broker itself knowing anything about HTTP, WebSocket, or any wire format. The +transport layers built in later phases wrap this core; nothing here routes +traffic. + +Why a broker at all: the browser is a stateful, single-owner resource and the +agent side is multi-tenant (many principals, profiles, sessions) and +multi-transport (local API, remote API, …). A controller must never be +addressable by a caller that merely resembles the right identity, and a +command must never be completable twice, cancellable by a stranger, or +observable after its owner has gone away. Every rule below exists to make one +of those violations structurally impossible rather than merely discouraged. + +Contract (each rule is exercised by tests/gateway/test_browser_control_broker.py): + +- **Registration tickets are short-lived, single-use, identity-bound, and + cryptographically random.** ``mint_ticket`` returns an opaque value + (``secrets``-derived, >= 32 chars) plus an expiry derived from the injected + clock; ``consume_ticket`` exchanges it exactly once for the + :class:`ControllerScope` it was minted for, raising + :class:`TicketInvalid` for unknown, already-consumed, or expired values. + The ticket is the only cross-transport credential minted here; transports + decide how to carry it. + +- **Exact identity and capability selection.** ``attach`` registers a send + callback under a :class:`ControllerScope`; ``select`` returns a controller + only when the caller's scope matches on *every* identity field — + principal, profile, session, controller id, browser profile id, and + transport family — and the requested capability is present in the + controller's capability set. Partial matches return ``None``. + +- **One pending command per command id; single-shot completion.** Each + ``dispatch`` mints a fresh command id, emits one + ``browser.controller.command`` frame, and parks a waiter keyed by that id. + ``complete`` resolves a command exactly once and returns ``False`` for any + later attempt (late completion after cancellation or detach is ignored). + +- **Scoped cancellation.** ``cancel`` aborts only the pending command whose + scope and tool_call_id match, emits a ``browser.controller.cancel`` frame + for it, and returns ``False`` when nothing matched. + +- **Detach fails closed.** ``detach`` removes the controller and cancels every + pending command of that scope; waiting dispatchers observe + :class:`ControllerCancelled` rather than hanging or racing a detached + controller's late ``complete``. + +Thread-safety: all public state transitions happen under a single reentrant +lock; the send callback is invoked *outside* the lock so a controller may +synchronously ``complete`` from inside its own send (the no-op round trip), +and waiters are parked on per-command events, not on the broker lock. +""" + +from __future__ import annotations + +import logging +import secrets +import threading +import time +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, Optional + +logger = logging.getLogger(__name__) +_OWNER_UNSET = object() + +#: Default lifetime of a minted registration ticket, in clock seconds. +DEFAULT_TICKET_TTL = 30.0 +#: Default wall time a dispatch waits for the controller to complete. +DEFAULT_COMMAND_TIMEOUT = 30.0 + +#: Wire method names for controller frames. Transport-neutral by contract: +#: transports carry these envelopes verbatim. +FRAME_COMMAND = "browser.controller.command" +FRAME_CANCEL = "browser.controller.cancel" + + +class BrowserControlError(Exception): + """Base class for broker contract failures.""" + + +class TicketInvalid(BrowserControlError): + """A registration ticket is unknown, already consumed, or expired.""" + + +class ControllerUnavailable(BrowserControlError): + """No attached controller exactly matches the requested scope/capability.""" + + +class ControllerCancelled(BrowserControlError): + """A pending command was cancelled (explicitly or by detach).""" + + +class ControllerTimeout(BrowserControlError): + """The controller did not complete the command before the timeout.""" + + +class ControllerRejected(BrowserControlError): + """The controller completed the command with ``ok=False``.""" + + +@dataclass(frozen=True) +class ControllerScope: + """Exact identity of a browser controller plus its capability set. + + Equality is structural over *all* fields, so two scopes differing in any + single field (including ``transport_family``) never match — this is the + "exact identity" contract. + """ + + principal_id: Optional[str] = None + profile_id: Optional[str] = None + session_id: Optional[str] = None + controller_id: Optional[str] = None + browser_profile_id: Optional[str] = None + transport_family: Optional[str] = None + capabilities: frozenset = frozenset() + + +@dataclass(frozen=True) +class Ticket: + """Opaque, single-use registration credential.""" + + value: str + expires_at: float + + +@dataclass +class _TicketRecord: + scope: ControllerScope + expires_at: float + consumed: bool = False + + +@dataclass +class _Controller: + scope: ControllerScope + send: Callable[[dict], None] + owner: Any = None + # Serialize command/cancel writes with detach or replacement. Broker state + # is never held while waiting for this lock, so a transport callback may + # synchronously call complete() without deadlocking the broker. + send_lock: threading.Lock = field(default_factory=threading.Lock) + + +@dataclass +class _PendingCommand: + scope: ControllerScope + command_id: str + tool_call_id: Optional[str] + event: threading.Event = field(default_factory=threading.Event) + done: bool = False + cancelled: bool = False + ok: bool = False + result: Any = None + + +class BrowserControlBroker: + """Thread-safe broker core binding controllers to callers. + + Parameters + ---------- + ticket_ttl: + Lifetime of minted tickets in clock seconds. + command_timeout: + Seconds a ``dispatch`` waits for completion before raising + :class:`ControllerTimeout`. + clock: + Injectable time source (defaults to ``time.monotonic``); tests pin it + to make expiry deterministic. + """ + + def __init__( + self, + *, + ticket_ttl: float = DEFAULT_TICKET_TTL, + command_timeout: float = DEFAULT_COMMAND_TIMEOUT, + clock: Optional[Callable[[], float]] = None, + ) -> None: + self._ticket_ttl = ticket_ttl + self._command_timeout = command_timeout + self._clock = clock if clock is not None else time.monotonic + self._lock = threading.RLock() + self._tickets: Dict[str, _TicketRecord] = {} + self._controllers: Dict[ControllerScope, _Controller] = {} + self._pending: Dict[str, _PendingCommand] = {} + + # ------------------------------------------------------------------ + # Registration tickets + # ------------------------------------------------------------------ + + def mint_ticket(self, scope: ControllerScope) -> Ticket: + """Mint a short-lived, single-use ticket bound to ``scope``.""" + now = self._clock() + with self._lock: + self._prune_tickets(now) + value = secrets.token_urlsafe(32) + record = _TicketRecord(scope=scope, expires_at=now + self._ticket_ttl) + self._tickets[value] = record + return Ticket(value=value, expires_at=record.expires_at) + + def consume_ticket(self, value: str) -> ControllerScope: + """Exchange a ticket for its scope, exactly once. + + Raises :class:`TicketInvalid` for unknown, already-consumed, or + expired tickets. The expiry check happens against the live clock at + consume time, so a ticket that outlived its TTL can never be used. + """ + now = self._clock() + with self._lock: + record = self._tickets.get(value) + if record is None: + raise TicketInvalid("unknown ticket") + if record.consumed: + raise TicketInvalid("ticket already consumed") + if now > record.expires_at: + raise TicketInvalid("ticket expired") + record.consumed = True + return record.scope + + def _prune_tickets(self, now: float) -> None: + """Drop expired tickets (caller must hold the lock).""" + expired = [value for value, rec in self._tickets.items() if rec.expires_at <= now] + for value in expired: + del self._tickets[value] + + # ------------------------------------------------------------------ + # Controller registration / selection + # ------------------------------------------------------------------ + + def attach( + self, + scope: ControllerScope, + send: Callable[[dict], None], + *, + owner: Any = None, + ) -> None: + """Register the controller owning ``scope`` with frame callback ``send``. + + Re-attaching an already-attached scope replaces the prior controller + (logged); a controller that wants to go away must call ``detach``. + """ + replacement = _Controller(scope=scope, send=send, owner=owner) + with self._lock: + existing = self._controllers.get(scope) + if existing is None: + with self._lock: + # A concurrent attach may have won after the optimistic read; + # retry through the replacement path rather than overwriting it. + existing = self._controllers.get(scope) + if existing is None: + self._controllers[scope] = replacement + return + + assert existing is not None + logger.warning( + "browser controller re-attached for scope %r; replacing prior controller", + scope, + ) + with existing.send_lock: + with self._lock: + if self._controllers.get(scope) is not existing: + # Another replacement won while this caller waited. Re-run + # against the new generation so its pending work is not + # orphaned by an unconditional overwrite. + retry = True + pendings = [] + else: + retry = False + pendings = self._pending_for_scope_locked(scope) + for pending in pendings: + self._resolve_pending(pending, cancelled=True) + self._controllers[scope] = replacement + if not retry: + self._emit_cancel_frames(existing, pendings) + return + self.attach(scope, send, owner=owner) + + def select(self, scope: ControllerScope, capability: str) -> Optional[_Controller]: + """Return the controller exactly matching ``scope`` and ``capability``. + + ``None`` when any identity field differs or the capability is not in + the controller's capability set. The controller's own scope is the + authority on capabilities. + """ + with self._lock: + controller = self._controllers.get(scope) + if controller is None: + return None + if capability not in controller.scope.capabilities: + return None + return controller + + def detach( + self, + scope: ControllerScope, + *, + owner: Any = _OWNER_UNSET, + notify_controller: bool = True, + ) -> None: + """Remove the controller for ``scope`` and fail its pending work closed. + + Every pending command of the scope is marked cancelled and resolved, + so waiting dispatchers raise :class:`ControllerCancelled`; a late + ``complete`` for any of them returns ``False`` (the command id is no + longer pending). + """ + with self._lock: + controller = self._controllers.get(scope) + if controller is None: + return + if owner is not _OWNER_UNSET and controller.owner != owner: + return + with controller.send_lock: + with self._lock: + if self._controllers.get(scope) is not controller: + return + if owner is not _OWNER_UNSET and controller.owner != owner: + return + self._controllers.pop(scope, None) + pendings = self._pending_for_scope_locked(scope) + for pending in pendings: + self._resolve_pending(pending, cancelled=True) + # Keep the old generation's send lock through cancellation so a + # command frame can never overtake its terminal cancel frame. + if notify_controller: + self._emit_cancel_frames(controller, pendings) + + # ------------------------------------------------------------------ + # Command lifecycle + # ------------------------------------------------------------------ + + def dispatch( + self, + scope: ControllerScope, + *, + action: str, + arguments: Optional[dict] = None, + tool_call_id: Optional[str] = None, + ) -> Any: + """Send one controller command and block for its completion. + + Emits a ``browser.controller.command`` frame carrying a fresh command + id, then waits up to ``command_timeout`` seconds. Returns the + controller's completion result, or raises: + + - :class:`ControllerUnavailable` — no exact scope/capability match; + - :class:`ControllerCancelled` — cancelled via ``cancel``/``detach``; + - :class:`ControllerTimeout` — no completion within the timeout; + - :class:`ControllerRejected` — completed with ``ok=False``. + + Exactly one pending command exists per command id; ``complete`` is + single-shot, so a command can never resolve twice. + """ + controller = self.select(scope, action) + if controller is None: + raise ControllerUnavailable( + f"no controller for scope {scope!r} with capability {action!r}" + ) + + command_id = secrets.token_hex(16) + frame = { + "method": FRAME_COMMAND, + "params": { + "command_id": command_id, + "action": action, + "arguments": dict(arguments or {}), + "controller_id": scope.controller_id, + "browser_profile_id": scope.browser_profile_id, + "tool_call_id": tool_call_id, + }, + } + pending = _PendingCommand( + scope=scope, + command_id=command_id, + tool_call_id=tool_call_id, + ) + with controller.send_lock: + with self._lock: + # select() intentionally runs outside the send lock. Revalidate + # the exact controller generation after acquiring it so detach + # or replacement cannot leave a stale command waiting forever. + if self._controllers.get(scope) is not controller: + raise ControllerUnavailable( + f"controller for scope {scope!r} detached before dispatch" + ) + self._pending[command_id] = pending + + try: + controller.send(frame) + except Exception: + # The command never left the building; unreserve the id and + # surface the transport failure to the caller. + with self._lock: + self._pending.pop(command_id, None) + raise + + if not pending.event.wait(timeout=self._command_timeout): + timed_out = False + with self._lock: + # Event.wait() may return False at the exact boundary where a + # completion already won and removed the pending command. + if not pending.done and self._pending.get(command_id) is pending: + pending.done = True + del self._pending[command_id] + timed_out = True + if timed_out: + with controller.send_lock: + with self._lock: + still_attached = self._controllers.get(scope) is controller + if still_attached: + self._emit_cancel_frames(controller, [pending]) + raise ControllerTimeout( + f"controller did not complete command {command_id!r} " + f"within {self._command_timeout}s" + ) + + if pending.cancelled: + raise ControllerCancelled(f"command {command_id!r} was cancelled") + if not pending.ok: + raise ControllerRejected( + f"controller rejected command {command_id!r}: {pending.result!r}" + ) + return pending.result + + def complete( + self, + command_id: str, + *, + scope: Optional[ControllerScope] = None, + ok: bool, + result: Any = None, + ) -> bool: + """Resolve a pending command by id; ``False`` when none is pending. + + Safe to call from inside the controller's own ``send`` callback (the + broker never holds its lock across a send). Late completions — after + ``cancel`` or ``detach`` already resolved the command — are ignored + and report ``False``. + """ + with self._lock: + pending = self._pending.get(command_id) + if pending is None or pending.done: + return False + if scope is not None and pending.scope != scope: + return False + pending.done = True + pending.ok = ok is True + pending.result = result + del self._pending[command_id] + pending.event.set() + return True + + def cancel(self, scope: ControllerScope, *, tool_call_id: Optional[str]) -> bool: + """Cancel exactly the pending command matching ``scope`` + tool_call_id. + + Emits one ``browser.controller.cancel`` frame naming the cancelled + command's id. Returns ``True`` when a command was cancelled and + ``False`` when nothing matched (so transports can answer idempotently + without inventing state). + """ + with self._lock: + controller = self._controllers.get(scope) + if controller is None: + return False + with controller.send_lock: + with self._lock: + if self._controllers.get(scope) is not controller: + return False + target = None + for pending in self._pending.values(): + if ( + pending.scope == scope + and pending.tool_call_id == tool_call_id + and not pending.done + ): + target = pending + break + if target is None: + return False + self._resolve_pending(target, cancelled=True) + self._emit_cancel_frames(controller, [target]) + return True + + # ------------------------------------------------------------------ + # Internals (all callers must hold the lock) + # ------------------------------------------------------------------ + + def _resolve_pending(self, pending: _PendingCommand, *, cancelled: bool) -> None: + """Mark ``pending`` resolved and drop it from the registry.""" + pending.cancelled = cancelled + pending.done = True + del self._pending[pending.command_id] + pending.event.set() + + def _pending_for_scope_locked(self, scope: ControllerScope) -> list[_PendingCommand]: + return [ + pending + for pending in list(self._pending.values()) + if pending.scope == scope + ] + + def _emit_cancel_frames( + self, controller: _Controller, pendings: list[_PendingCommand] + ) -> None: + for pending in pendings: + frame = { + "method": FRAME_CANCEL, + "params": { + "command_id": pending.command_id, + "tool_call_id": pending.tool_call_id, + }, + } + try: + controller.send(frame) + except Exception: + logger.exception( + "failed to emit cancel frame for command %r", pending.command_id + ) + + def scope_for_session( + self, + *, + session_id: Optional[str] = None, + task_id: Optional[str] = None, + principal_id: Optional[str] = None, + transport_family: Optional[str] = None, + ) -> Optional[ControllerScope]: + """Return one unambiguous attached scope for a server-owned session. + + A public session id is only a lookup hint. The caller must also supply + its server-derived principal and transport family; missing identity, + no match, or multiple matches fail closed rather than selecting by + insertion order. + """ + target = str(session_id or task_id or "").strip() + principal = str(principal_id or "").strip() + family = str(transport_family or "").strip() + if not target or not principal or not family: + return None + with self._lock: + matches = [ + scope + for scope in self._controllers + if scope.session_id == target + and scope.principal_id == principal + and scope.transport_family == family + ] + return matches[0] if len(matches) == 1 else None + + def detach_owner(self, owner: Any, *, notify_controller: bool = True) -> int: + """Detach every controller owned by one transport connection.""" + with self._lock: + scopes = [ + scope + for scope, controller in self._controllers.items() + if controller.owner == owner + ] + for scope in scopes: + self.detach( + scope, + owner=owner, + notify_controller=notify_controller, + ) + return len(scopes) + + def reset(self) -> None: + """Fail all live work closed and clear tickets (tests/shutdown).""" + with self._lock: + scopes = list(self._controllers) + for scope in scopes: + self.detach(scope) + with self._lock: + self._tickets.clear() + # Defensive cleanup for any pending entry whose controller was + # concurrently removed by a transport teardown. + for pending in list(self._pending.values()): + self._resolve_pending(pending, cancelled=True) + + @property + def ticket_ttl_seconds(self) -> float: + """Configured lifetime for newly minted one-shot tickets.""" + return self._ticket_ttl + + @property + def pending_count(self) -> int: + """Number of commands awaiting completion (diagnostics/tests).""" + with self._lock: + return len(self._pending) + + +_GLOBAL_BROKER = BrowserControlBroker() + + +def get_browser_control_broker() -> BrowserControlBroker: + """Process-local broker shared by API and dashboard Gateway transports.""" + return _GLOBAL_BROKER + + +def browser_control_enabled(config: Optional[dict] = None) -> bool: + """Return the explicit Phase 4 feature flag (disabled by default).""" + if config is None: + try: + from hermes_cli.config import load_config + + config = load_config() + except Exception: + return False + if not isinstance(config, dict): + return False + browser = config.get("browser") + if not isinstance(browser, dict): + return False + extension_control = browser.get("extension_control") + if not isinstance(extension_control, dict): + return False + return extension_control.get("enabled", False) is True diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 53e9ad7a70..33efd02172 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -70,6 +70,22 @@ _PROFILE_REJECTED = object() _api_request_profile: ContextVar[Optional[str]] = ContextVar( "api_server_request_profile", default=None ) +_api_request_browser_control_principal: ContextVar[str] = ContextVar( + "api_server_browser_control_principal", default="" +) +_api_request_browser_control_transport_family: ContextVar[str] = ContextVar( + "api_server_browser_control_transport_family", default="" +) + +#: Phase 4 browser-extension control protocol version (advertised in +#: /v1/capabilities and echoed in registration responses). +_BROWSER_CONTROL_PROTOCOL_VERSION = 1 +#: Capabilities this phase actually grants a controller. Only the no-op +#: probe is real until the action protocol ships; any requested capability +#: outside this set is filtered out rather than advertised. +_BROWSER_CONTROL_CAPABILITIES = frozenset({"controller.noop"}) +_BROWSER_CONTROL_WS_PROTOCOL = "hermes-browser-control-v1" +_BROWSER_CONTROL_TICKET_PROTOCOL_PREFIX = "hermes-browser-control-ticket." def _approval_event_choices(*, smart_denied: bool, allow_permanent: bool) -> list[str]: if smart_denied: @@ -95,6 +111,11 @@ from gateway.platforms.base import ( from agent.redact import redact_sensitive_text from agent.interrupt_compat import request_hard_interrupt from gateway.readiness import collect_runtime_readiness +from gateway.browser_control_broker import ( + ControllerScope, + TicketInvalid, + get_browser_control_broker, +) from agent.secret_scope import UnscopedSecretError as _UnscopedSecretError from agent.secret_scope import get_secret as _scoped_get_secret @@ -1494,6 +1515,11 @@ class APIServerAdapter(BasePlatformAdapter): # Shutdown counts this reservation so the request cannot slip through # the drain between its first await and _run_agent()/task registration. self._pending_agent_requests: int = 0 + # Phase 4 browser-control broker core: transport-neutral ticket / + # controller / command lifecycle shared with the dashboard Gateway + # transport. This adapter only maps HTTP registration and the + # controller WebSocket onto the broker; it owns no broker state. + self._browser_control_broker = get_browser_control_broker() def active_agent_work_count(self) -> int: """Return all live agent work owned by this API adapter. @@ -2063,7 +2089,18 @@ class APIServerAdapter(BasePlatformAdapter): token = _api_request_profile.set(profile) try: with self._profile_scope(profile): - return await handler(request) + resolved_profile = profile or "default" + principal_token = _api_request_browser_control_principal.set( + self._derive_browser_control_principal(resolved_profile) + ) + family_token = _api_request_browser_control_transport_family.set( + self._browser_control_transport_family(request) + ) + try: + return await handler(request) + finally: + _api_request_browser_control_transport_family.reset(family_token) + _api_request_browser_control_principal.reset(principal_token) finally: _api_request_profile.reset(token) @@ -2082,6 +2119,12 @@ class APIServerAdapter(BasePlatformAdapter): ("GET", "/v1/models", self._handle_models), ("GET", "/api/model/options", self._handle_model_options), ("GET", "/v1/capabilities", self._handle_capabilities), + # Phase 4 authenticated browser-control surface: POST registration + # mints a short-lived ticket; the controller then opens the WS with + # that ticket. Both are gated on browser.extension_control.enabled + # and API-key auth (see the handlers for the exact status ladder). + ("POST", "/v1/browser-control/register", self._handle_browser_control_register), + ("GET", "/v1/browser-control/ws", self._handle_browser_control_ws), ("GET", "/v1/skills", self._handle_skills), ("GET", "/v1/toolsets", self._handle_toolsets), ("GET", "/api/sessions", self._handle_list_sessions), @@ -3213,6 +3256,21 @@ class APIServerAdapter(BasePlatformAdapter): "session_continuity_header": "X-Hermes-Session-Id", "session_key_header": "X-Hermes-Session-Key", "cors": bool(self._cors_origins), + # Phase 4 browser-extension control. Always advertised (so + # clients can feature-detect), but truthful: disabled until + # browser.extension_control.enabled is set, and Phase 4 + # exposes no real browser actions — only the no-op controller + # capability is ever granted. + "browser_extension_control": { + "enabled": self._browser_control_enabled(), + "protocol_version": _BROWSER_CONTROL_PROTOCOL_VERSION, + "capabilities": list(_BROWSER_CONTROL_CAPABILITIES), + "real_browser_actions": False, + "transports": { + "local_vps": "websocket-subprotocol-ticket", + "cloud": "authenticated-gateway-rpc", + }, + }, }, "endpoints": { "health": {"method": "GET", "path": "/health"}, @@ -3239,9 +3297,307 @@ class APIServerAdapter(BasePlatformAdapter): "session_chat": {"method": "POST", "path": "/api/sessions/{session_id}/chat"}, "session_chat_stream": {"method": "POST", "path": "/api/sessions/{session_id}/chat/stream"}, "session_model_lock": {"method": "POST", "path": "/api/sessions/{session_id}/model"}, + "browser_control_register": {"method": "POST", "path": "/v1/browser-control/register"}, + "browser_control_ws": {"method": "GET", "path": "/v1/browser-control/ws"}, }, }) + # ------------------------------------------------------------------ + # Phase 4 browser-extension control (authenticated local/VPS API) + # ------------------------------------------------------------------ + + async def _handle_browser_control_register(self, request: "web.Request") -> "web.Response": + """POST /v1/browser-control/register — mint a controller ticket. + + The extension controller proves itself with the same Bearer API key + every other API-server client uses, then receives a short-lived, + single-use ticket to open the controller WebSocket. Identity is NOT + taken from the request body: the scope principal is derived + server-side from the authenticated key/profile as a non-reversible + digest, and the capability set is filtered to what this phase + actually grants (``controller.noop``), so a spoofed + ``principal_id`` or inflated capability list in the payload is + ignored rather than honored. The named session must already exist in + the active profile's server-owned SessionDB before a ticket is minted. + + Status ladder: 404 when the feature is disabled, 403 when no API key + is configured at all (registration can never be authenticated), 401 + for a missing/invalid Bearer token, 201 on success. + """ + if not self._browser_control_enabled(): + return web.json_response( + _openai_error( + "Browser control is not enabled on this server.", + code="browser_control_disabled", + ), + status=404, + ) + if not self._api_key: + logger.warning( + "browser-control registration rejected: no API key configured; " + "set API_SERVER_KEY to enable authenticated browser control." + ) + return web.json_response( + _openai_error( + "Browser control registration requires a configured API key.", + err_type="gateway_auth_error", + code="browser_control_auth_required", + ), + status=403, + ) + auth_err = self._check_auth(request) + if auth_err: + return auth_err + + try: + payload = await request.json() + except Exception: + return web.json_response( + _openai_error("Request body must be valid JSON."), status=400 + ) + if not isinstance(payload, dict): + return web.json_response( + _openai_error("Request body must be a JSON object."), status=400 + ) + + controller_id = str(payload.get("controller_id") or "").strip() + browser_profile_id = str(payload.get("browser_profile_id") or "").strip() + session_id = str(payload.get("session_id") or "").strip() + if not controller_id or not browser_profile_id or not session_id: + return web.json_response( + _openai_error( + "controller_id, browser_profile_id, and session_id are required.", + code="browser_control_invalid_registration", + ), + status=400, + ) + + db = await self._ensure_session_db_async() + if db is None: + return web.json_response( + _openai_error( + "Session database unavailable.", + code="session_db_unavailable", + ), + status=503, + ) + session = await asyncio.to_thread(db.get_session, session_id) + if not session: + return web.json_response( + _openai_error( + "Browser control may register only for an existing server session.", + err_type="gateway_auth_error", + code="browser_control_session_forbidden", + ), + status=403, + ) + + profile = _api_request_profile.get() or "default" + capabilities = frozenset( + capability + for capability in payload.get("capabilities") or [] + if isinstance(capability, str) + and capability in _BROWSER_CONTROL_CAPABILITIES + ) + scope = ControllerScope( + principal_id=self._derive_browser_control_principal(profile), + profile_id=profile, + session_id=session_id or None, + controller_id=controller_id, + browser_profile_id=browser_profile_id, + transport_family=self._browser_control_transport_family(request), + capabilities=capabilities, + ) + ticket = self._browser_control_broker.mint_ticket(scope) + ticket_ttl = self._browser_control_broker.ticket_ttl_seconds + return web.json_response( + { + "protocol_version": _BROWSER_CONTROL_PROTOCOL_VERSION, + "ticket": ticket.value, + "ticket_expires_at": time.time() + ticket_ttl, + "ticket_expires_in_seconds": ticket_ttl, + "ws_path": "/v1/browser-control/ws", + "scope": { + "principal_id": scope.principal_id, + "profile_id": scope.profile_id, + "session_id": scope.session_id, + "controller_id": scope.controller_id, + "browser_profile_id": scope.browser_profile_id, + "transport_family": scope.transport_family, + "capabilities": sorted(scope.capabilities), + }, + }, + status=201, + ) + + async def _handle_browser_control_ws(self, request: "web.Request") -> "web.WebSocketResponse": + """GET /v1/browser-control/ws — controller WebSocket (one-shot ticket). + + A ticket-bearing ``Sec-WebSocket-Protocol`` token is exchanged exactly + once for the identity scope minted at registration; query-string, + unknown, already-consumed, or expired tickets are rejected with 401 + before upgrade. The socket then attaches to the shared broker under + that scope, forwards broker command/cancel frames onto the aiohttp loop + thread-safely, and accepts controller result/cancel frames. Completion + is exact-scope checked, and owner-aware teardown cannot detach a newer + replacement controller generation. + """ + # Re-check at upgrade time so disabling the feature immediately closes + # the admission gate without consuming still-live one-shot tickets. + if not self._browser_control_enabled(): + raise web.HTTPNotFound() + + # Credentials in the request target are liable to appear in access + # logs. Accept the one-shot ticket only as a WebSocket subprotocol; + # reject the former query-string shape without consuming it. + if request.query.get("ticket"): + raise web.HTTPUnauthorized() + requested_protocols = [ + value.strip() + for value in request.headers.get("Sec-WebSocket-Protocol", "").split(",") + if value.strip() + ] + ticket_protocols = [ + value + for value in requested_protocols + if value.startswith(_BROWSER_CONTROL_TICKET_PROTOCOL_PREFIX) + ] + if ( + _BROWSER_CONTROL_WS_PROTOCOL not in requested_protocols + or len(ticket_protocols) != 1 + ): + raise web.HTTPUnauthorized() + ticket_value = ticket_protocols[0][len(_BROWSER_CONTROL_TICKET_PROTOCOL_PREFIX) :] + if not ticket_value: + raise web.HTTPUnauthorized() + try: + scope = self._browser_control_broker.consume_ticket(ticket_value) + except TicketInvalid: + raise web.HTTPUnauthorized() from None + except Exception: + logger.exception("browser-control WS ticket consumption failed") + raise web.HTTPUnauthorized() from None + + ws = web.WebSocketResponse( + heartbeat=30.0, + protocols=(_BROWSER_CONTROL_WS_PROTOCOL,), + ) + await ws.prepare(request) + loop = asyncio.get_running_loop() + + def _send(frame: dict) -> None: + """Broker send callback: forward a frame onto the aiohttp loop. + + Called from broker dispatch threads; aiohttp socket writes must + happen on the event loop. Waiting for the write here preserves + command ordering and lets a closed socket fail the dispatch + instead of silently dropping the frame. + """ + if ws.closed: + raise ConnectionError("browser-control websocket is closed") + try: + on_loop = asyncio.get_running_loop() is loop + except RuntimeError: + on_loop = False + if on_loop: + loop.create_task(ws.send_json(frame)) + return + future = asyncio.run_coroutine_threadsafe(ws.send_json(frame), loop) + future.result(timeout=10.0) + + self._browser_control_broker.attach(scope, _send, owner=ws) + try: + async for msg in ws: + if msg.type == web.WSMsgType.TEXT: + try: + frame = msg.json() + except Exception: + continue + if isinstance(frame, dict): + self._handle_browser_control_frame(scope, frame) + elif msg.type in (web.WSMsgType.CLOSE, web.WSMsgType.ERROR): + break + finally: + self._browser_control_broker.detach( + scope, + owner=ws, + notify_controller=False, + ) + return ws + + def _handle_browser_control_frame(self, scope: "ControllerScope", frame: dict) -> None: + """Apply one controller→broker frame with exact-scope checks.""" + method = frame.get("method") + params = frame.get("params") + if not isinstance(params, dict): + return + if method == "browser.controller.result": + command_id = params.get("command_id") + if isinstance(command_id, str) and command_id: + # Broker resolves only the pending command whose scope equals + # this socket's scope; a stranger's command id is a no-op. + ok = params.get("ok") is True + self._browser_control_broker.complete( + command_id, + scope=scope, + ok=ok, + result=params.get("result") if ok else params.get("error"), + ) + elif method == "browser.controller.cancel": + tool_call_id = params.get("tool_call_id") + if isinstance(tool_call_id, str) and tool_call_id: + self._browser_control_broker.cancel(scope, tool_call_id=tool_call_id) + + def _browser_control_enabled(self) -> bool: + """Phase 4 feature flag; False unless explicitly enabled. + + Reads ``browser.extension_control.enabled`` from the global config + (defaults to False). Tests monkeypatch this method directly to force + the feature on/off without touching config. + """ + try: + from gateway.browser_control_broker import browser_control_enabled as _flag + + return _flag() + except Exception: + return False + + def _derive_browser_control_principal(self, profile: str) -> str: + """Server-derived controller principal (non-reversible digest). + + The principal is bound to the credential that authenticated the + registration request — the expected API key for the request's + profile — so a client cannot impersonate another controller by + echoing an id in the registration body. + """ + key = self._expected_api_key() or self._api_key or "" + digest = hashlib.sha256(f"{profile}\x00{key}".encode("utf-8")).hexdigest() + return f"principal:{profile}:{digest[:32]}" + + def _browser_control_transport_family(self, request: "web.Request") -> str: + """Local vs remote API family, decided by the loopback peer. + + A controller speaking to a localhost listener is in the same trust + domain as the host and gets the ``local-api`` family; anything else + is ``remote-api``. The broker treats the family as part of exact + identity, so a remote controller can never satisfy a local-only + dispatch (and vice versa). + """ + host = None + try: + transport = request.transport + if transport is not None: + peer = transport.get_extra_info("peername") + if isinstance(peer, tuple) and peer: + host = peer[0] + elif isinstance(peer, str): + host = peer + except Exception: + host = None + if host in ("127.0.0.1", "::1", "localhost"): + return "local-api" + return "remote-api" + async def _handle_skills(self, request: "web.Request") -> "web.Response": """GET /v1/skills — list installed skills visible to the API-server agent. @@ -6308,6 +6664,8 @@ class APIServerAdapter(BasePlatformAdapter): chat_id: str = "", session_key: str = "", session_id: str = "", + browser_control_principal: str = "", + browser_control_transport_family: str = "", ) -> list: """Bind session contextvars for an API-server agent run. @@ -6331,6 +6689,8 @@ class APIServerAdapter(BasePlatformAdapter): chat_id=chat_id, session_key=session_key, session_id=session_id, + browser_control_principal=browser_control_principal, + browser_control_transport_family=browser_control_transport_family, async_delivery=False, cron_session="", ) @@ -6391,6 +6751,12 @@ class APIServerAdapter(BasePlatformAdapter): # run_in_executor threads, so the profile scope must be re-entered # inside _run() from this explicit value. request_profile = _api_request_profile.get() + request_browser_control_principal = ( + _api_request_browser_control_principal.get() + ) + request_browser_control_transport_family = ( + _api_request_browser_control_transport_family.get() + ) def _run(): from gateway.session_context import clear_session_vars @@ -6400,6 +6766,10 @@ class APIServerAdapter(BasePlatformAdapter): chat_id=session_id or "", session_key=gateway_session_key or session_id or "", session_id=session_id or "", + browser_control_principal=request_browser_control_principal, + browser_control_transport_family=( + request_browser_control_transport_family + ), ) agent = None try: @@ -6833,6 +7203,12 @@ class APIServerAdapter(BasePlatformAdapter): # Background task outlives the HTTP response (and thus the middleware # profile scope). Capture now and re-enter inside the task/executor. request_profile = _api_request_profile.get() + request_browser_control_principal = ( + _api_request_browser_control_principal.get() + ) + request_browser_control_transport_family = ( + _api_request_browser_control_transport_family.get() + ) async def _run_and_close(): try: @@ -6922,6 +7298,12 @@ class APIServerAdapter(BasePlatformAdapter): chat_id=session_id or "", session_key=approval_session_key, session_id=session_id or "", + browser_control_principal=( + request_browser_control_principal + ), + browser_control_transport_family=( + request_browser_control_transport_family + ), ) register_gateway_notify(approval_session_key, _approval_notify) # /v1/runs runs its own agent lifecycle (no diff --git a/gateway/session_context.py b/gateway/session_context.py index 7a2c53ab3a..9a6a8c4226 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -102,6 +102,12 @@ _SESSION_UI_SESSION_ID: ContextVar = ContextVar("HERMES_UI_SESSION_ID", default= _SESSION_MESSAGE_ID: ContextVar = ContextVar("HERMES_SESSION_MESSAGE_ID", default=_UNSET) _SESSION_PROFILE: ContextVar = ContextVar("HERMES_SESSION_PROFILE", default=_UNSET) +_BROWSER_CONTROL_PRINCIPAL: ContextVar = ContextVar( + "HERMES_BROWSER_CONTROL_PRINCIPAL", default=_UNSET +) +_BROWSER_CONTROL_TRANSPORT_FAMILY: ContextVar = ContextVar( + "HERMES_BROWSER_CONTROL_TRANSPORT_FAMILY", default=_UNSET +) # Per-session cron marker. Unlike the process-global legacy env var, this is # scoped to one cron job / inbound session. _UNSET preserves the legacy env @@ -151,6 +157,8 @@ _VAR_MAP = { "HERMES_UI_SESSION_ID": _SESSION_UI_SESSION_ID, "HERMES_SESSION_MESSAGE_ID": _SESSION_MESSAGE_ID, "HERMES_SESSION_PROFILE": _SESSION_PROFILE, + "HERMES_BROWSER_CONTROL_PRINCIPAL": _BROWSER_CONTROL_PRINCIPAL, + "HERMES_BROWSER_CONTROL_TRANSPORT_FAMILY": _BROWSER_CONTROL_TRANSPORT_FAMILY, "HERMES_CRON_SESSION": _CRON_SESSION, "HERMES_CRON_AUTO_DELIVER_PLATFORM": _CRON_AUTO_DELIVER_PLATFORM, "HERMES_CRON_AUTO_DELIVER_CHAT_ID": _CRON_AUTO_DELIVER_CHAT_ID, @@ -228,6 +236,8 @@ def set_session_vars( session_id: str = "", message_id: str = "", profile: str = "", + browser_control_principal: str = "", + browser_control_transport_family: str = "", cwd: str = "", async_delivery: bool = True, ui_session_id: str = "", @@ -273,6 +283,8 @@ def set_session_vars( _SESSION_UI_SESSION_ID.set(ui_session_id), _SESSION_MESSAGE_ID.set(message_id), _SESSION_PROFILE.set(profile), + _BROWSER_CONTROL_PRINCIPAL.set(browser_control_principal), + _BROWSER_CONTROL_TRANSPORT_FAMILY.set(browser_control_transport_family), _CRON_SESSION.set(cron_session), _SESSION_ASYNC_DELIVERY.set(bool(async_delivery)), ] @@ -312,6 +324,8 @@ def clear_session_vars(tokens: list) -> None: _SESSION_UI_SESSION_ID, _SESSION_MESSAGE_ID, _SESSION_PROFILE, + _BROWSER_CONTROL_PRINCIPAL, + _BROWSER_CONTROL_TRANSPORT_FAMILY, _CRON_SESSION, ): var.set("") diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index cc24499880..22dd7b56a1 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -16010,7 +16010,16 @@ def _ws_auth_reason(ws: "WebSocket") -> tuple[Optional[str], str]: internal = ws.query_params.get("internal", "") if internal: try: - consume_internal_credential(internal) + info = consume_internal_credential(internal) + # Stamp the server-minted identity onto the WS object so the + # connection (and any transport built from it) can never be + # impersonated by RPC params. Internal peers are marked + # ``server-internal`` and are excluded from privileged + # controller registration downstream. + ws._hermes_auth_identity = { + "user_id": info.get("user_id"), + "provider": info.get("provider"), + } return None, "internal" except TicketInvalid as exc: audit_log( @@ -16026,7 +16035,18 @@ def _ws_auth_reason(ws: "WebSocket") -> tuple[Optional[str], str]: return "no_credential", "none" try: - consume_ticket(ticket) + info = consume_ticket(ticket) + # The ticket binds a server-minted {user_id, provider}; stamp it + # onto the WS object so ``gateway_ws`` can hand it to the gateway + # transport, where it is the sole identity authority for + # browser-controller registration. A client can never supply or + # spoof this value through RPC params. Only the two identity + # fields are carried — bookkeeping (e.g. ``minted_at``) is not + # part of the identity contract. + ws._hermes_auth_identity = { + "user_id": info.get("user_id"), + "provider": info.get("provider"), + } return None, "ticket" except TicketInvalid as exc: audit_log( @@ -17147,7 +17167,11 @@ async def gateway_ws(ws: WebSocket) -> None: from tui_gateway.ws import handle_ws - await handle_ws(ws) + # The authenticated identity (ticket / internal credential) was stamped + # onto the WS object by _ws_auth_reason; carry it into the gateway + # transport where it becomes the identity authority for privileged RPCs + # (browser.controller.register). None on the legacy token path. + await handle_ws(ws, auth_identity=getattr(ws, "_hermes_auth_identity", None)) # --------------------------------------------------------------------------- diff --git a/tests/gateway/test_browser_control_api.py b/tests/gateway/test_browser_control_api.py new file mode 100644 index 0000000000..d47d98e845 --- /dev/null +++ b/tests/gateway/test_browser_control_api.py @@ -0,0 +1,390 @@ +import asyncio +import time + +import pytest +from aiohttp import WSServerHandshakeError, web +from aiohttp.test_utils import TestClient, TestServer + +from gateway.browser_control_broker import ControllerRejected, ControllerScope +from gateway.config import PlatformConfig +from gateway.platforms.api_server import APIServerAdapter + + +API_KEY = "-".join(("fixture", "neutral", "api", "key", "123")) +CONTROL_PROTOCOL = "hermes-browser-control-v1" + + +class _SessionDB: + def __init__(self): + self.sessions = { + "session-fixture": {"id": "session-fixture", "source": "api_server"}, + "remote-session-fixture": { + "id": "remote-session-fixture", + "source": "api_server", + }, + } + + def get_session(self, session_id): + return self.sessions.get(session_id) + + +def _ticket_protocol(ticket): + return f"hermes-browser-control-ticket.{ticket}" + + +def _adapter(*, key=API_KEY): + adapter = APIServerAdapter( + PlatformConfig(enabled=True, extra={"key": key} if key else {}) + ) + adapter._session_db = _SessionDB() + return adapter + + +def _app(adapter): + app = web.Application() + app.router.add_get("/v1/capabilities", adapter._handle_capabilities) + app.router.add_post( + "/v1/browser-control/register", adapter._handle_browser_control_register + ) + app.router.add_get( + "/v1/browser-control/ws", adapter._handle_browser_control_ws + ) + return app + + +def _registration_body(**overrides): + payload = { + "protocol_version": 1, + "controller_id": "controller-fixture", + "browser_profile_id": "browser-profile-fixture", + "session_id": "session-fixture", + "capabilities": ["controller.noop", "browser_navigate"], + "principal_id": "spoofed-client-principal", + "product": { + "id": "chromium", + "engine": "chromium", + "label": "Chromium browser", + }, + } + payload.update(overrides) + return payload + + +def test_route_table_advertises_registration_and_controller_ws_without_replacing_existing_routes(): + adapter = _adapter() + routes = {(method, path) for method, path, _handler in adapter._http_route_table()} + assert ("POST", "/v1/browser-control/register") in routes + assert ("GET", "/v1/browser-control/ws") in routes + assert ("POST", "/v1/chat/completions") in routes + + +def test_api_agent_context_binds_server_principal_and_transport_family(): + from gateway.session_context import clear_session_vars, get_session_env + + adapter = _adapter() + tokens = adapter._bind_api_server_session( + session_id="session-fixture", + browser_control_principal="principal-fixture", + browser_control_transport_family="local-api", + ) + try: + assert ( + get_session_env("HERMES_BROWSER_CONTROL_PRINCIPAL") + == "principal-fixture" + ) + assert ( + get_session_env("HERMES_BROWSER_CONTROL_TRANSPORT_FAMILY") + == "local-api" + ) + finally: + clear_session_vars(tokens) + + +@pytest.mark.asyncio +async def test_capabilities_are_truthful_and_disabled_by_default(monkeypatch): + adapter = _adapter() + monkeypatch.setattr(adapter, "_browser_control_enabled", lambda: False) + async with TestClient(TestServer(_app(adapter))) as client: + response = await client.get( + "/v1/capabilities", headers={"Authorization": f"Bearer {API_KEY}"} + ) + assert response.status == 200 + data = await response.json() + + control = data["features"]["browser_extension_control"] + assert control == { + "enabled": False, + "protocol_version": 1, + "capabilities": ["controller.noop"], + "real_browser_actions": False, + "transports": { + "local_vps": "websocket-subprotocol-ticket", + "cloud": "authenticated-gateway-rpc", + }, + } + assert data["endpoints"]["browser_control_register"] == { + "method": "POST", + "path": "/v1/browser-control/register", + } + assert data["endpoints"]["browser_control_ws"] == { + "method": "GET", + "path": "/v1/browser-control/ws", + } + + +@pytest.mark.asyncio +async def test_api_middleware_stamps_server_control_identity_for_agent_entry(): + from gateway.platforms.api_server import ( + _api_request_browser_control_principal, + _api_request_browser_control_transport_family, + ) + + adapter = _adapter() + + async def inspect(_request): + return web.json_response( + { + "principal": _api_request_browser_control_principal.get(), + "transport_family": ( + _api_request_browser_control_transport_family.get() + ), + } + ) + + app = web.Application(middlewares=[adapter._make_profile_prefix_middleware()]) + app.router.add_get("/inspect", inspect) + async with TestClient(TestServer(app)) as client: + response = await client.get("/inspect") + body = await response.json() + + assert body == { + "principal": adapter._derive_browser_control_principal("default"), + "transport_family": "local-api", + } + + +@pytest.mark.asyncio +async def test_registration_requires_enabled_feature_and_configured_bearer_auth(monkeypatch): + disabled = _adapter() + monkeypatch.setattr(disabled, "_browser_control_enabled", lambda: False) + async with TestClient(TestServer(_app(disabled))) as client: + response = await client.post( + "/v1/browser-control/register", + json=_registration_body(), + headers={"Authorization": f"Bearer {API_KEY}"}, + ) + assert response.status == 404 + + unkeyed = _adapter(key="") + monkeypatch.setattr(unkeyed, "_browser_control_enabled", lambda: True) + async with TestClient(TestServer(_app(unkeyed))) as client: + response = await client.post( + "/v1/browser-control/register", json=_registration_body() + ) + assert response.status == 403 + assert (await response.json())["error"]["code"] == "browser_control_auth_required" + + keyed = _adapter() + monkeypatch.setattr(keyed, "_browser_control_enabled", lambda: True) + async with TestClient(TestServer(_app(keyed))) as client: + response = await client.post( + "/v1/browser-control/register", json=_registration_body() + ) + assert response.status == 401 + response = await client.post( + "/v1/browser-control/register", + json=_registration_body(session_id=""), + headers={"Authorization": f"Bearer {API_KEY}"}, + ) + assert response.status == 400 + + response = await client.post( + "/v1/browser-control/register", + json=_registration_body(session_id="not-a-server-session"), + headers={"Authorization": f"Bearer {API_KEY}"}, + ) + assert response.status == 403 + assert (await response.json())["error"]["code"] == ( + "browser_control_session_forbidden" + ) + + +@pytest.mark.asyncio +async def test_controller_ws_rechecks_feature_flag_before_consuming_ticket(monkeypatch): + adapter = _adapter() + monkeypatch.setattr(adapter, "_browser_control_enabled", lambda: True) + async with TestClient(TestServer(_app(adapter))) as client: + response = await client.post( + "/v1/browser-control/register", + json=_registration_body(), + headers={"Authorization": f"Bearer {API_KEY}"}, + ) + ticket = (await response.json())["ticket"] + monkeypatch.setattr(adapter, "_browser_control_enabled", lambda: False) + with pytest.raises(WSServerHandshakeError) as disabled: + await client.ws_connect( + "/v1/browser-control/ws", + protocols=[CONTROL_PROTOCOL, _ticket_protocol(ticket)], + ) + assert disabled.value.status == 404 + + # Neither a missing protocol nor the legacy query-string shape may + # consume the one-shot credential. + monkeypatch.setattr(adapter, "_browser_control_enabled", lambda: True) + with pytest.raises(WSServerHandshakeError) as query_ticket: + await client.ws_connect(f"/v1/browser-control/ws?ticket={ticket}") + assert query_ticket.value.status == 401 + ws = await client.ws_connect( + "/v1/browser-control/ws", + protocols=[CONTROL_PROTOCOL, _ticket_protocol(ticket)], + ) + await ws.close() + + +@pytest.mark.asyncio +async def test_local_api_ticket_ws_noop_round_trip_filters_spoofed_identity_and_disabled_actions(monkeypatch): + adapter = _adapter() + monkeypatch.setattr(adapter, "_browser_control_enabled", lambda: True) + async with TestClient(TestServer(_app(adapter))) as client: + response = await client.post( + "/v1/browser-control/register", + json=_registration_body(), + headers={"Authorization": f"Bearer {API_KEY}"}, + ) + assert response.status == 201 + registration = await response.json() + assert registration["protocol_version"] == 1 + assert registration["ticket"] + assert registration["ticket_expires_at"] > time.time() + assert 0 < registration["ticket_expires_in_seconds"] <= 30 + assert registration["ws_path"] == "/v1/browser-control/ws" + assert registration["scope"]["principal_id"] != "spoofed-client-principal" + assert registration["scope"]["transport_family"] == "local-api" + assert registration["scope"]["capabilities"] == ["controller.noop"] + + ws = await client.ws_connect( + "/v1/browser-control/ws", + protocols=[CONTROL_PROTOCOL, _ticket_protocol(registration["ticket"])], + ) + scope = ControllerScope( + principal_id=registration["scope"]["principal_id"], + profile_id=registration["scope"]["profile_id"], + session_id=registration["scope"]["session_id"], + controller_id=registration["scope"]["controller_id"], + browser_profile_id=registration["scope"]["browser_profile_id"], + transport_family=registration["scope"]["transport_family"], + capabilities=frozenset(registration["scope"]["capabilities"]), + ) + + pending = asyncio.create_task( + asyncio.to_thread( + adapter._browser_control_broker.dispatch, + scope, + action="controller.noop", + arguments={"echo": "local-api"}, + tool_call_id="tool-call-fixture", + ) + ) + command = await ws.receive_json(timeout=2.0) + assert command["method"] == "browser.controller.command" + assert command["params"]["action"] == "controller.noop" + await ws.send_json( + { + "method": "browser.controller.result", + "params": { + "command_id": command["params"]["command_id"], + "ok": True, + "result": {"echo": "local-api"}, + }, + } + ) + assert await asyncio.wait_for(pending, timeout=2.0) == {"echo": "local-api"} + + rejected = asyncio.create_task( + asyncio.to_thread( + adapter._browser_control_broker.dispatch, + scope, + action="controller.noop", + arguments={"echo": "reject"}, + tool_call_id="tool-call-rejected", + ) + ) + rejected_command = await ws.receive_json(timeout=2.0) + await ws.send_json( + { + "method": "browser.controller.result", + "params": { + "command_id": rejected_command["params"]["command_id"], + "ok": "false", + "error": {"code": "controller_rejected", "message": "fixture rejection"}, + }, + } + ) + with pytest.raises(ControllerRejected, match="controller_rejected"): + await asyncio.wait_for(rejected, timeout=2.0) + await ws.close() + + with pytest.raises(WSServerHandshakeError) as replay: + await client.ws_connect( + "/v1/browser-control/ws", + protocols=[CONTROL_PROTOCOL, _ticket_protocol(registration["ticket"])], + ) + assert replay.value.status == 401 + + +@pytest.mark.asyncio +async def test_remote_api_uses_the_same_authenticated_noop_round_trip(monkeypatch): + adapter = _adapter() + monkeypatch.setattr(adapter, "_browser_control_enabled", lambda: True) + monkeypatch.setattr( + adapter, + "_browser_control_transport_family", + lambda request: "remote-api", + ) + async with TestClient(TestServer(_app(adapter))) as client: + response = await client.post( + "/v1/browser-control/register", + json=_registration_body(session_id="remote-session-fixture"), + headers={"Authorization": f"Bearer {API_KEY}"}, + ) + registration = await response.json() + assert response.status == 201 + assert registration["scope"]["transport_family"] == "remote-api" + + ws = await client.ws_connect( + "/v1/browser-control/ws", + protocols=[CONTROL_PROTOCOL, _ticket_protocol(registration["ticket"])], + ) + scope = ControllerScope( + principal_id=registration["scope"]["principal_id"], + profile_id=registration["scope"]["profile_id"], + session_id=registration["scope"]["session_id"], + controller_id=registration["scope"]["controller_id"], + browser_profile_id=registration["scope"]["browser_profile_id"], + transport_family="remote-api", + capabilities=frozenset(registration["scope"]["capabilities"]), + ) + pending = asyncio.create_task( + asyncio.to_thread( + adapter._browser_control_broker.dispatch, + scope, + action="controller.noop", + arguments={"family": "remote-api"}, + tool_call_id="tool-call-remote", + ) + ) + command = await ws.receive_json(timeout=2.0) + await ws.send_json( + { + "method": "browser.controller.result", + "params": { + "command_id": command["params"]["command_id"], + "ok": True, + "result": {"family": "remote-api"}, + }, + } + ) + assert await asyncio.wait_for(pending, timeout=2.0) == { + "family": "remote-api" + } + await ws.close() diff --git a/tests/gateway/test_browser_control_broker.py b/tests/gateway/test_browser_control_broker.py new file mode 100644 index 0000000000..6f910f87e6 --- /dev/null +++ b/tests/gateway/test_browser_control_broker.py @@ -0,0 +1,159 @@ +import threading +import time + +import pytest + +from gateway.browser_control_broker import ( + BrowserControlBroker, + ControllerCancelled, + ControllerScope, + TicketInvalid, +) + + +def _scope(**overrides): + values = { + "principal_id": "principal-fixture", + "profile_id": "default", + "session_id": "session-fixture", + "controller_id": "controller-fixture", + "browser_profile_id": "browser-profile-fixture", + "transport_family": "local-api", + "capabilities": frozenset({"controller.noop"}), + } + values.update(overrides) + return ControllerScope(**values) + + +def test_registration_ticket_is_short_lived_single_use_and_identity_bound(): + now = [100.0] + broker = BrowserControlBroker(ticket_ttl=30.0, clock=lambda: now[0]) + scope = _scope() + + ticket = broker.mint_ticket(scope) + assert len(ticket.value) >= 32 + assert ticket.expires_at == 130.0 + assert broker.consume_ticket(ticket.value) == scope + with pytest.raises(TicketInvalid, match="unknown|consumed"): + broker.consume_ticket(ticket.value) + + expired = broker.mint_ticket(scope) + now[0] = 131.0 + with pytest.raises(TicketInvalid, match="expired"): + broker.consume_ticket(expired.value) + + +def test_controller_selection_requires_exact_principal_profile_session_controller_and_browser_profile(): + broker = BrowserControlBroker() + scope = _scope() + broker.attach(scope, lambda _frame: None) + + assert broker.select(scope, "controller.noop") is not None + for field, value in ( + ("principal_id", "other-principal"), + ("profile_id", "other-profile"), + ("session_id", "other-session"), + ("controller_id", "other-controller"), + ("browser_profile_id", "other-browser-profile"), + ("transport_family", "remote-api"), + ): + assert broker.select(_scope(**{field: value}), "controller.noop") is None + assert broker.select(scope, "browser_navigate") is None + + +def test_noop_round_trip_uses_controller_and_returns_result_without_enabling_browser_actions(): + broker = BrowserControlBroker(command_timeout=1.0) + scope = _scope() + + def send(frame): + assert frame["method"] == "browser.controller.command" + assert frame["params"]["action"] == "controller.noop" + broker.complete( + frame["params"]["command_id"], + ok=True, + result={"echo": frame["params"]["arguments"]["echo"]}, + ) + + broker.attach(scope, send) + result = broker.dispatch( + scope, + action="controller.noop", + arguments={"echo": "phase-4"}, + tool_call_id="tool-call-fixture", + ) + assert result == {"echo": "phase-4"} + assert broker.pending_count == 0 + assert broker.select(scope, "browser_navigate") is None + + +def test_cancellation_targets_only_the_matching_pending_command_and_cleans_up(): + broker = BrowserControlBroker(command_timeout=2.0) + scope = _scope() + frames = [] + command_ready = threading.Event() + + def send(frame): + frames.append(frame) + if frame["method"] == "browser.controller.command": + command_ready.set() + + broker.attach(scope, send) + outcome = {} + + def run_dispatch(): + try: + broker.dispatch( + scope, + action="controller.noop", + arguments={}, + tool_call_id="tool-call-cancelled", + ) + except Exception as exc: # asserted below + outcome["error"] = exc + + thread = threading.Thread(target=run_dispatch) + thread.start() + assert command_ready.wait(timeout=1.0) + + assert broker.cancel(scope, tool_call_id="wrong-tool-call") is False + assert broker.cancel(scope, tool_call_id="tool-call-cancelled") is True + thread.join(timeout=1.0) + + assert not thread.is_alive() + assert isinstance(outcome.get("error"), ControllerCancelled) + cancel_frames = [frame for frame in frames if frame["method"] == "browser.controller.cancel"] + assert len(cancel_frames) == 1 + assert cancel_frames[0]["params"]["command_id"] == frames[0]["params"]["command_id"] + assert broker.pending_count == 0 + + +def test_detach_fails_pending_work_closed_and_late_completion_is_ignored(): + broker = BrowserControlBroker(command_timeout=2.0) + scope = _scope() + command_id = [] + command_ready = threading.Event() + + def send(frame): + if frame["method"] == "browser.controller.command": + command_id.append(frame["params"]["command_id"]) + command_ready.set() + + broker.attach(scope, send) + outcome = {} + + def run_dispatch(): + try: + broker.dispatch(scope, action="controller.noop", arguments={}) + except Exception as exc: # asserted below + outcome["error"] = exc + + thread = threading.Thread(target=run_dispatch) + thread.start() + assert command_ready.wait(timeout=1.0) + broker.detach(scope) + thread.join(timeout=1.0) + + assert not thread.is_alive() + assert isinstance(outcome.get("error"), ControllerCancelled) + assert broker.complete(command_id[0], ok=True, result={}) is False + assert broker.pending_count == 0 diff --git a/tests/gateway/test_browser_control_broker_hardening.py b/tests/gateway/test_browser_control_broker_hardening.py new file mode 100644 index 0000000000..b78aec407b --- /dev/null +++ b/tests/gateway/test_browser_control_broker_hardening.py @@ -0,0 +1,331 @@ +import threading + +import pytest + +from gateway.browser_control_broker import ( + BrowserControlBroker, + browser_control_enabled, + ControllerCancelled, + ControllerScope, + ControllerRejected, + ControllerTimeout, + ControllerUnavailable, +) + + +def _scope(**overrides): + values = { + "principal_id": "principal-fixture", + "profile_id": "default", + "session_id": "session-fixture", + "controller_id": "controller-fixture", + "browser_profile_id": "browser-profile-fixture", + "transport_family": "local-api", + "capabilities": frozenset({"controller.noop"}), + } + values.update(overrides) + return ControllerScope(**values) + + +def _start_pending(broker, scope, *, tool_call_id="tool-call-fixture"): + outcome = {} + ready = threading.Event() + frames = [] + + def send(frame): + frames.append(frame) + if frame["method"] == "browser.controller.command": + ready.set() + + broker.attach(scope, send, owner="owner-fixture") + + def run(): + try: + outcome["result"] = broker.dispatch( + scope, + action="controller.noop", + arguments={}, + tool_call_id=tool_call_id, + ) + except Exception as exc: + outcome["error"] = exc + + thread = threading.Thread(target=run) + thread.start() + assert ready.wait(timeout=1.0) + return thread, outcome, frames + + +def test_detach_emits_cancel_before_controller_is_removed(): + broker = BrowserControlBroker(command_timeout=1.0) + scope = _scope() + thread, outcome, frames = _start_pending(broker, scope) + + broker.detach(scope) + thread.join(timeout=1.0) + + assert isinstance(outcome.get("error"), ControllerCancelled) + assert [frame["method"] for frame in frames] == [ + "browser.controller.command", + "browser.controller.cancel", + ] + + +def test_dispatch_revalidates_selected_controller_after_detach_race(): + broker = BrowserControlBroker(command_timeout=0.2) + scope = _scope() + broker.attach(scope, lambda _frame: None) + selected = threading.Event() + resume = threading.Event() + original_select = broker.select + + def paused_select(candidate_scope, capability): + controller = original_select(candidate_scope, capability) + selected.set() + assert resume.wait(timeout=1.0) + return controller + + broker.select = paused_select + outcome = {} + + def run(): + try: + broker.dispatch(scope, action="controller.noop", arguments={}) + except Exception as exc: + outcome["error"] = exc + + thread = threading.Thread(target=run) + thread.start() + assert selected.wait(timeout=1.0) + broker.detach(scope) + resume.set() + thread.join(timeout=1.0) + + assert not thread.is_alive() + assert isinstance(outcome.get("error"), ControllerUnavailable) + assert broker.pending_count == 0 + + +def test_completion_requires_the_same_scope_as_the_pending_command(): + broker = BrowserControlBroker(command_timeout=1.0) + scope = _scope() + thread, outcome, frames = _start_pending(broker, scope) + command_id = frames[0]["params"]["command_id"] + + assert broker.complete( + command_id, + scope=_scope(principal_id="other-principal"), + ok=True, + result={"unsafe": True}, + ) is False + assert thread.is_alive() + assert broker.complete( + command_id, + scope=scope, + ok=True, + result={"safe": True}, + ) is True + thread.join(timeout=1.0) + + assert outcome.get("result") == {"safe": True} + assert broker.pending_count == 0 + + +def test_reattach_cancels_pending_work_from_the_previous_controller_generation(): + broker = BrowserControlBroker(command_timeout=1.0) + scope = _scope() + thread, outcome, frames = _start_pending(broker, scope) + old_command_id = frames[0]["params"]["command_id"] + + broker.attach(scope, lambda _frame: None, owner="replacement-owner") + thread.join(timeout=1.0) + + assert not thread.is_alive() + assert isinstance(outcome.get("error"), ControllerCancelled) + assert broker.complete(old_command_id, scope=scope, ok=True, result={}) is False + + +def test_session_lookup_fails_closed_on_ambiguity_and_owner_detach_is_scoped(): + broker = BrowserControlBroker() + first = _scope(controller_id="controller-one") + second = _scope(controller_id="controller-two") + other = _scope( + session_id="other-session", + controller_id="controller-other", + transport_family="cloud-ticket-ws", + ) + broker.attach(first, lambda _frame: None, owner="owner-shared") + broker.attach(second, lambda _frame: None, owner="owner-shared") + broker.attach(other, lambda _frame: None, owner="owner-other") + + assert broker.scope_for_session( + session_id="session-fixture", + principal_id="principal-fixture", + transport_family="local-api", + ) is None + assert broker.scope_for_session( + session_id="other-session", + principal_id="principal-fixture", + transport_family="cloud-ticket-ws", + ) == other + + assert broker.detach_owner("owner-shared") == 2 + assert broker.scope_for_session( + session_id="other-session", + principal_id="principal-fixture", + transport_family="cloud-ticket-ws", + ) == other + assert broker.detach_owner("missing-owner") == 0 + + broker.reset() + assert broker.scope_for_session( + session_id="other-session", + principal_id="principal-fixture", + transport_family="cloud-ticket-ws", + ) is None + assert broker.pending_count == 0 + + +def test_session_lookup_requires_exact_server_principal_and_transport_family(): + broker = BrowserControlBroker() + local = _scope( + principal_id="principal:api:local", + controller_id="controller-local", + transport_family="local-api", + ) + remote = _scope( + principal_id="principal:api:remote", + controller_id="controller-remote", + transport_family="remote-api", + ) + broker.attach(local, lambda _frame: None, owner="owner-local") + broker.attach(remote, lambda _frame: None, owner="owner-remote") + + assert broker.scope_for_session(session_id="session-fixture") is None + assert broker.scope_for_session( + session_id="session-fixture", + principal_id="principal:api:local", + transport_family="local-api", + ) == local + assert broker.scope_for_session( + session_id="session-fixture", + principal_id="principal:api:local", + transport_family="remote-api", + ) is None + assert broker.scope_for_session( + session_id="session-fixture", + principal_id="principal:api:attacker", + transport_family="local-api", + ) is None + + +def test_feature_flag_requires_literal_boolean_true(): + assert browser_control_enabled({}) is False + assert browser_control_enabled( + {"browser": {"extension_control": {"enabled": False}}} + ) is False + assert browser_control_enabled( + {"browser": {"extension_control": {"enabled": True}}} + ) is True + for ambiguous in ("true", "false", "yes", 1, [], {}): + assert browser_control_enabled( + {"browser": {"extension_control": {"enabled": ambiguous}}} + ) is False + + +def test_transport_teardown_can_cancel_waiters_without_writing_to_closing_peer(): + broker = BrowserControlBroker(command_timeout=1.0) + scope = _scope() + thread, outcome, frames = _start_pending(broker, scope) + + assert broker.detach_owner("owner-fixture", notify_controller=False) == 1 + thread.join(timeout=1.0) + + assert isinstance(outcome.get("error"), ControllerCancelled) + assert [frame["method"] for frame in frames] == ["browser.controller.command"] + assert broker.pending_count == 0 + + +def test_stale_owner_teardown_cannot_detach_replacement_controller_generation(): + broker = BrowserControlBroker() + scope = _scope() + first_owner = object() + live_owner = object() + broker.attach(scope, lambda frame: None, owner=first_owner) + broker.attach(scope, lambda frame: None, owner=live_owner) + + broker.detach( + scope, + owner=first_owner, + notify_controller=False, + ) + + selected = broker.select(scope, "controller.noop") + assert selected is not None + assert selected.owner is live_owner + + +def test_completion_winning_at_timeout_boundary_is_not_misreported_as_timeout(): + broker = BrowserControlBroker(command_timeout=0.01) + scope = _scope() + + class BoundaryEvent: + def set(self): + pass + + def wait(self, timeout): + assert broker.complete( + command_id, + scope=scope, + ok=True, + result={"boundary": "completed"}, + ) + return False + + def send(frame): + nonlocal command_id + command_id = frame["params"]["command_id"] + broker._pending[command_id].event = BoundaryEvent() + + command_id = "" + broker.attach(scope, send) + assert broker.dispatch(scope, action="controller.noop") == { + "boundary": "completed" + } + + +def test_timeout_marks_terminal_and_emits_cancel_to_controller(): + broker = BrowserControlBroker(command_timeout=0.01) + scope = _scope() + frames = [] + broker.attach(scope, frames.append) + + with pytest.raises(ControllerTimeout): + broker.dispatch( + scope, + action="controller.noop", + tool_call_id="tool-timeout", + ) + + assert [frame["method"] for frame in frames] == [ + "browser.controller.command", + "browser.controller.cancel", + ] + assert broker.pending_count == 0 + + +def test_non_boolean_success_values_fail_closed(): + broker = BrowserControlBroker(command_timeout=1.0) + scope = _scope() + + def send(frame): + assert broker.complete( + frame["params"]["command_id"], + scope=scope, + ok="false", + result={"spoofed": True}, + ) + + broker.attach(scope, send) + with pytest.raises(ControllerRejected): + broker.dispatch(scope, action="controller.noop") diff --git a/tests/gateway/test_browser_control_cloud.py b/tests/gateway/test_browser_control_cloud.py new file mode 100644 index 0000000000..0d40941045 --- /dev/null +++ b/tests/gateway/test_browser_control_cloud.py @@ -0,0 +1,294 @@ +import threading +from types import SimpleNamespace + +import pytest + +from gateway.browser_control_broker import ControllerRejected, get_browser_control_broker +from hermes_cli import web_server +from hermes_cli.dashboard_auth.ws_tickets import _reset_for_tests, mint_ticket +from tui_gateway import server +from tui_gateway.ws import WSTransport +from tui_gateway.methods_browser_control import _broker_event_writer, _principal_digest + + +def _fake_ticket_ws(ticket): + return SimpleNamespace( + query_params={"ticket": ticket}, + client=SimpleNamespace(host="203.0.113.7"), + url=SimpleNamespace(path="/api/ws"), + ) + + +@pytest.fixture +def gated_dashboard(): + previous = getattr(web_server.app.state, "auth_required", False) + web_server.app.state.auth_required = True + try: + yield + finally: + web_server.app.state.auth_required = previous + _reset_for_tests() + + +def test_dashboard_ticket_identity_is_carried_forward_without_trusting_rpc_params(gated_dashboard): + _reset_for_tests() + ticket = mint_ticket(user_id="user-fixture", provider="provider-fixture") + ws = _fake_ticket_ws(ticket) + + assert web_server._ws_auth_ok(ws) is True + assert ws._hermes_auth_identity == { + "user_id": "user-fixture", + "provider": "provider-fixture", + } + assert web_server._ws_auth_ok(_fake_ticket_ws(ticket)) is False + + +def test_ws_transport_records_only_server_authenticated_identity(): + loop = SimpleNamespace() + identity = {"user_id": "user-fixture", "provider": "provider-fixture"} + transport = WSTransport( + SimpleNamespace(), + loop, + peer="identity-test", + auth_identity=identity, + ) + assert transport.auth_identity == identity + + +def test_cloud_agent_context_binds_registration_principal_and_transport_family(): + from gateway.session_context import clear_session_vars, get_session_env + + identity = {"user_id": "user-fixture", "provider": "provider-fixture"} + transport = SimpleNamespace(auth_identity=identity) + server._sessions["context-session-fixture"] = { + "transport": transport, + "session_key": "stored-context-session", + "profile": "default", + "agent": SimpleNamespace(session_id="context-session-fixture"), + } + tokens = [] + try: + tokens = server._set_session_context("stored-context-session") + assert get_session_env("HERMES_BROWSER_CONTROL_PRINCIPAL") == _principal_digest( + identity + ) + assert ( + get_session_env("HERMES_BROWSER_CONTROL_TRANSPORT_FAMILY") + == "cloud-ticket-ws" + ) + finally: + clear_session_vars(tokens) + server._sessions.pop("context-session-fixture", None) + + +def test_cloud_event_writer_surfaces_closed_or_failed_transport_immediately(): + class RaisingTransport: + def write(self, _frame): + raise ConnectionError("fixture transport closed") + + class FalseTransport: + def write(self, _frame): + return False + + frame = {"method": "browser.controller.command", "params": {"command_id": "fixture"}} + with pytest.raises(ConnectionError, match="fixture transport closed"): + _broker_event_writer(RaisingTransport(), "session-fixture")(frame) + with pytest.raises(ConnectionError, match="failed"): + _broker_event_writer(FalseTransport(), "session-fixture")(frame) + + +def test_cloud_principal_digest_is_unambiguous_across_identity_components(): + assert _principal_digest({"user_id": "a:b", "provider": "c"}) != _principal_digest( + {"user_id": "a", "provider": "b:c"} + ) + + +@pytest.mark.parametrize( + "identity", + [None, {}, {"user_id": "server-internal", "provider": "server-internal"}], +) +def test_cloud_controller_registration_rejects_missing_or_internal_identity(monkeypatch, identity): + monkeypatch.setattr( + "gateway.browser_control_broker.browser_control_enabled", lambda: True + ) + + class Transport: + auth_identity = identity + + def write(self, _frame): + return True + + transport = Transport() + server._sessions["session-fixture"] = { + "transport": transport, + "session_key": "stored-session-fixture", + "profile": "default", + } + try: + response = server.dispatch( + { + "jsonrpc": "2.0", + "id": 1, + "method": "browser.controller.register", + "params": { + "session_id": "session-fixture", + "controller_id": "controller-fixture", + "browser_profile_id": "browser-profile-fixture", + "capabilities": ["controller.noop"], + "principal_id": "spoofed-client-principal", + }, + }, + transport, + ) + assert response["error"]["code"] == 4403 + finally: + server._sessions.pop("session-fixture", None) + + +def test_cloud_gateway_noop_round_trip_is_bound_to_ticket_identity_and_session_transport(monkeypatch): + monkeypatch.setattr( + "gateway.browser_control_broker.browser_control_enabled", lambda: True + ) + broker = get_browser_control_broker() + broker.reset() + frames = [] + ready = threading.Event() + + class Transport: + auth_identity = { + "user_id": "user-fixture", + "provider": "provider-fixture", + } + + def write(self, frame): + frames.append(frame) + if frame.get("method") == "event": + ready.set() + return True + + transport = Transport() + server._sessions["session-fixture"] = { + "transport": transport, + "session_key": "stored-session-fixture", + "profile": "default", + } + try: + registration = server.dispatch( + { + "jsonrpc": "2.0", + "id": 1, + "method": "browser.controller.register", + "params": { + "protocol_version": 1, + "session_id": "session-fixture", + "controller_id": "controller-fixture", + "browser_profile_id": "browser-profile-fixture", + "capabilities": ["controller.noop", "browser_navigate"], + "principal_id": "spoofed-client-principal", + }, + }, + transport, + ) + scope_payload = registration["result"]["scope"] + assert scope_payload["principal_id"] != "spoofed-client-principal" + assert scope_payload["transport_family"] == "cloud-ticket-ws" + assert scope_payload["capabilities"] == ["controller.noop"] + + missing_identity = server.dispatch( + { + "jsonrpc": "2.0", + "id": 41, + "method": "browser.controller.register", + "params": { + "session_id": "session-fixture", + "controller_id": "", + "browser_profile_id": "", + "capabilities": ["controller.noop"], + }, + }, + transport=transport, + ) + assert missing_identity["error"]["code"] == 4403 + + scope = broker.scope_for_session( + session_id="session-fixture", + principal_id=scope_payload["principal_id"], + transport_family="cloud-ticket-ws", + ) + assert scope is not None + outcome = {} + + def dispatch_noop(): + outcome["result"] = broker.dispatch( + scope, + action="controller.noop", + arguments={"echo": "cloud"}, + tool_call_id="tool-call-cloud", + ) + + thread = threading.Thread(target=dispatch_noop) + thread.start() + assert ready.wait(timeout=1.0) + command_event = frames[-1] + assert command_event["method"] == "event" + assert command_event["params"]["type"] == "browser.controller.command" + command_id = command_event["params"]["payload"]["command_id"] + + result_response = server.dispatch( + { + "jsonrpc": "2.0", + "id": 2, + "method": "browser.controller.result", + "params": { + "session_id": "session-fixture", + "command_id": command_id, + "ok": True, + "result": {"echo": "cloud"}, + }, + }, + transport, + ) + assert result_response["result"]["accepted"] is True + thread.join(timeout=1.0) + assert outcome["result"] == {"echo": "cloud"} + + rejected_outcome = {} + ready.clear() + + def dispatch_rejected(): + try: + rejected_outcome["result"] = broker.dispatch( + scope, + action="controller.noop", + arguments={"echo": "reject"}, + tool_call_id="tool-call-cloud-rejected", + ) + except Exception as exc: # asserted below + rejected_outcome["error"] = exc + + rejected_thread = threading.Thread(target=dispatch_rejected, daemon=True) + rejected_thread.start() + assert ready.wait(timeout=1.0) + rejected_event = frames[-1] + rejected_response = server.dispatch( + { + "jsonrpc": "2.0", + "id": 8, + "method": "browser.controller.result", + "params": { + "session_id": "session-fixture", + "command_id": rejected_event["params"]["payload"]["command_id"], + "ok": "false", + "error": {"code": "controller_rejected", "message": "fixture rejection"}, + }, + }, + transport=transport, + ) + assert rejected_response["result"]["accepted"] is True + rejected_thread.join(timeout=1.0) + assert isinstance(rejected_outcome.get("error"), ControllerRejected) + assert "controller_rejected" in str(rejected_outcome["error"]) + assert broker.pending_count == 0 + finally: + broker.reset() + server._sessions.pop("session-fixture", None) diff --git a/tests/tools/test_browser_extension_router.py b/tests/tools/test_browser_extension_router.py new file mode 100644 index 0000000000..66dcf4ea3e --- /dev/null +++ b/tests/tools/test_browser_extension_router.py @@ -0,0 +1,198 @@ +import pytest + +from tools.browser_extension_router import route_browser_tool, routed_browser_handler + + +class FakeBroker: + def __init__(self, *, scope=None, selected=None, result=None, error=None): + self.scope = scope + self.selected = selected + self.result = result + self.error = error + self.calls = [] + + def scope_for_session(self, **identity): + self.calls.append(("scope", identity)) + return self.scope + + def select(self, scope, action): + self.calls.append(("select", scope, action)) + return self.selected + + def dispatch(self, scope, *, action, arguments, tool_call_id=""): + self.calls.append(("dispatch", scope, action, arguments, tool_call_id)) + if self.error: + raise self.error + return self.result + + +def test_feature_off_calls_existing_backend_once_without_touching_broker(): + broker = FakeBroker() + fallbacks = [] + args = {"url": "https://example.test"} + + result = route_browser_tool( + "browser_navigate", + args, + fallback=lambda: fallbacks.append(args.copy()) or "legacy-result", + broker=broker, + enabled=False, + session_id="session-fixture", + task_id="task-fixture", + tool_call_id="tool-call-fixture", + ) + + assert result == "legacy-result" + assert fallbacks == [{"url": "https://example.test"}] + assert broker.calls == [] + + +@pytest.mark.parametrize( + "scope,selected", + [(None, None), ("scope-fixture", None)], +) +def test_no_exact_capable_controller_preserves_existing_backend(scope, selected): + broker = FakeBroker(scope=scope, selected=selected) + fallbacks = [] + + result = route_browser_tool( + "browser_navigate", + {"url": "https://example.test"}, + fallback=lambda: fallbacks.append(True) or "legacy-result", + broker=broker, + enabled=True, + session_id="session-fixture", + task_id="task-fixture", + tool_call_id="tool-call-fixture", + ) + + assert result == "legacy-result" + assert fallbacks == [True] + assert not any(call[0] == "dispatch" for call in broker.calls) + + +def test_selected_controller_receives_immutable_arguments_and_context(): + broker = FakeBroker( + scope="scope-fixture", + selected="connection-fixture", + result='{"ok": true, "source": "browser-extension"}', + ) + args = {"url": "https://example.test"} + + result = route_browser_tool( + "browser_navigate", + args, + fallback=lambda: pytest.fail("selected controller must not call fallback"), + broker=broker, + enabled=True, + session_id="session-fixture", + task_id="task-fixture", + principal_id="principal-fixture", + transport_family="local-api", + tool_call_id="tool-call-fixture", + ) + + assert result == '{"ok": true, "source": "browser-extension"}' + assert args == {"url": "https://example.test"} + assert broker.calls == [ + ( + "scope", + { + "session_id": "session-fixture", + "task_id": "task-fixture", + "principal_id": "principal-fixture", + "transport_family": "local-api", + }, + ), + ("select", "scope-fixture", "browser_navigate"), + ( + "dispatch", + "scope-fixture", + "browser_navigate", + {"url": "https://example.test"}, + "tool-call-fixture", + ), + ] + + +def test_selected_controller_failure_never_retries_through_existing_backend(): + broker = FakeBroker( + scope="scope-fixture", + selected="connection-fixture", + error=TimeoutError("controller timed out"), + ) + fallbacks = [] + + with pytest.raises(TimeoutError, match="controller timed out"): + route_browser_tool( + "browser_navigate", + {"url": "https://example.test"}, + fallback=lambda: fallbacks.append(True) or "unsafe-retry", + broker=broker, + enabled=True, + session_id="session-fixture", + task_id="task-fixture", + principal_id="principal-fixture", + transport_family="local-api", + tool_call_id="tool-call-fixture", + ) + + assert fallbacks == [] + + +def test_missing_server_bound_identity_falls_back_without_querying_broker(): + broker = FakeBroker(scope="attacker-scope", selected="attacker-controller") + fallbacks = [] + + result = route_browser_tool( + "browser_navigate", + {"url": "https://example.test"}, + fallback=lambda: fallbacks.append(True) or "legacy-result", + broker=broker, + enabled=True, + session_id="session-fixture", + ) + + assert result == "legacy-result" + assert fallbacks == [True] + assert broker.calls == [] + + +def test_routed_handler_reads_server_bound_identity_from_session_context(monkeypatch): + from gateway import browser_control_broker + from gateway.session_context import clear_session_vars, set_session_vars + + broker = FakeBroker( + scope="scope-fixture", + selected="connection-fixture", + result="controller-result", + ) + monkeypatch.setattr(browser_control_broker, "browser_control_enabled", lambda: True) + monkeypatch.setattr( + browser_control_broker, "get_browser_control_broker", lambda: broker + ) + tokens = set_session_vars( + session_id="session-fixture", + browser_control_principal="principal-fixture", + browser_control_transport_family="cloud-ticket-ws", + ) + try: + result = routed_browser_handler( + "browser_navigate", + {"url": "https://example.test"}, + fallback=lambda: pytest.fail("bound controller must be selected"), + tool_call_id="tool-call-fixture", + ) + finally: + clear_session_vars(tokens) + + assert result == "controller-result" + assert broker.calls[0] == ( + "scope", + { + "session_id": "session-fixture", + "task_id": None, + "principal_id": "principal-fixture", + "transport_family": "cloud-ticket-ws", + }, + ) diff --git a/tests/tools/test_browser_extension_router_wiring.py b/tests/tools/test_browser_extension_router_wiring.py new file mode 100644 index 0000000000..0a81b9c567 --- /dev/null +++ b/tests/tools/test_browser_extension_router_wiring.py @@ -0,0 +1,92 @@ +"""Wiring regression tests for the Phase 4 browser extension router. + +These guard the *registry wiring* — that every ``browser_*`` handler routes +through :func:`tools.browser_extension_router.routed_browser_handler` with +the tool's action name, its raw args, and its identity kwargs, instead of +calling the legacy backend directly. The routing contract itself is tested +by ``test_browser_extension_router.py``; here we only pin the plumbing. +""" + +import pytest + +from tools.registry import registry + + +@pytest.fixture(autouse=True) +def _route_spy(monkeypatch): + """Replace the wrapper with a spy that records the route and then runs + the legacy fallback, so each test proves the handler is wired without + exercising real routing or a real browser backend.""" + calls = [] + + def spy(action, args, *, fallback, task_id=None, session_id=None, tool_call_id=None): + calls.append( + { + "action": action, + "args": dict(args), + "task_id": task_id, + "session_id": session_id, + "tool_call_id": tool_call_id, + } + ) + return fallback() + + import tools.browser_tool as browser_tool + import tools.browser_cdp_tool as browser_cdp_tool + + monkeypatch.setattr(browser_tool, "routed_browser_handler", spy) + monkeypatch.setattr(browser_cdp_tool, "routed_browser_handler", spy) + monkeypatch.setattr(browser_tool, "browser_navigate", lambda url="", task_id=None: "legacy-nav") + monkeypatch.setattr(browser_cdp_tool, "browser_cdp", lambda *a, **k: "legacy-cdp") + return calls + + +BROWSER_ACTIONS = [ + "browser_navigate", + "browser_snapshot", + "browser_click", + "browser_type", + "browser_scroll", + "browser_back", + "browser_press", + "browser_get_images", + "browser_vision", + "browser_console", +] + + +def test_every_browser_registry_handler_routes_through_wrapper(_route_spy): + for name in BROWSER_ACTIONS: + _route_spy.clear() + handler = registry.get_entry(name).handler + args = {"url": "https://example.test", "ref": "@e1", "text": "hi"} + result = handler(dict(args), task_id="task-fixture", session_id="session-fixture") + assert result is not None + assert len(_route_spy) == 1, f"{name} did not route through the wrapper" + route = _route_spy[0] + assert route["action"] == name + assert route["task_id"] == "task-fixture" + assert route["session_id"] == "session-fixture" + + +def test_browser_navigate_forwards_raw_args_and_identity(_route_spy): + handler = registry.get_entry("browser_navigate").handler + args = {"url": "https://example.test"} + result = handler(dict(args), task_id="task-fixture", session_id="session-fixture") + assert result == "legacy-nav" + route = _route_spy[0] + assert route["args"] == args + # The router must not mutate the args dict. + assert args == {"url": "https://example.test"} + + +def test_browser_cdp_handler_routes_through_wrapper(_route_spy): + handler = registry.get_entry("browser_cdp").handler + args = {"method": "Target.getTargets", "params": {"filter": []}} + result = handler(dict(args), task_id="task-fixture", session_id="session-fixture") + assert result == "legacy-cdp" + route = _route_spy[0] + assert route["action"] == "browser_cdp" + assert route["args"] == args + assert route["task_id"] == "task-fixture" + assert route["session_id"] == "session-fixture" diff --git a/tools/browser_cdp_tool.py b/tools/browser_cdp_tool.py index eccd8f8fc1..8c23f8a6d7 100644 --- a/tools/browser_cdp_tool.py +++ b/tools/browser_cdp_tool.py @@ -23,6 +23,7 @@ import logging from typing import Any, Dict, Optional from tools.registry import registry, tool_error +from tools.browser_extension_router import routed_browser_handler logger = logging.getLogger(__name__) @@ -671,13 +672,19 @@ registry.register( name="browser_cdp", toolset="browser-cdp", schema=BROWSER_CDP_SCHEMA, - handler=lambda args, **kw: browser_cdp( - method=args.get("method", ""), - params=args.get("params"), - target_id=args.get("target_id"), - frame_id=args.get("frame_id"), - timeout=args.get("timeout", 30.0), + handler=lambda args, **kw: routed_browser_handler( + "browser_cdp", + args, + fallback=lambda: browser_cdp( + method=args.get("method", ""), + params=args.get("params"), + target_id=args.get("target_id"), + frame_id=args.get("frame_id"), + timeout=args.get("timeout", 30.0), + task_id=kw.get("task_id"), + ), task_id=kw.get("task_id"), + session_id=kw.get("session_id"), ), check_fn=_browser_cdp_check, emoji="🧪", diff --git a/tools/browser_extension_router.py b/tools/browser_extension_router.py new file mode 100644 index 0000000000..7a0429e02e --- /dev/null +++ b/tools/browser_extension_router.py @@ -0,0 +1,200 @@ +"""Phase 4 registry-level browser extension router. + +This module is the *agent-side* half of the browser-extension-control +feature: it decides, for one registry ``browser_*`` handler invocation, +whether the command is executed by an attached extension controller (via +the :mod:`gateway.browser_control_broker`) or by the existing legacy +browser backend. + +Routing contract (exercised by ``tests/tools/test_browser_extension_router.py``): + +- **Feature off ⇒ legacy, untouched.** When ``enabled`` is false the broker + is never touched and ``fallback()`` is called exactly once. This is the + default: ``browser.extension_control.enabled`` is false unless explicitly + configured, so every real browser action keeps its exact legacy path. + +- **No exact server-bound scope ⇒ legacy.** ``broker.scope_for_session(...)`` + must return exactly one attached controller scope for the caller's session, + authenticated principal, and transport family. Missing identity, no match, + or ambiguity preserves the existing backend. + +- **No exact capable controller ⇒ legacy.** ``broker.select(scope, action)`` + must return a controller whose capability set contains the action. + Controllers currently register with only ``controller.noop``, so real + browser actions never match and always fall back. + +- **Selected controller ⇒ authoritative.** Once a controller is selected the + command is dispatched to it and its result returned; the legacy backend + is *never* retried, even when the controller fails (timeout, cancellation, + rejection, transport error all propagate to the caller). + +- **Arguments are never mutated.** ``args`` is passed through untouched; + the broker copies arguments into its command frame itself. + +The lazy wrapper :func:`routed_browser_handler` is what the ``browser_*`` +registry handlers call. It resolves the feature flag and the process-local +broker lazily on every invocation so importing this module (or +``tools.browser_tool``) never pulls in the gateway, and so a mid-process +config change is honored without restart. +""" + +from __future__ import annotations + +import logging +from typing import Any, Callable, Dict, Optional + +logger = logging.getLogger(__name__) + + +def route_browser_tool( + action: str, + args: Dict[str, Any], + *, + fallback: Callable[[], Any], + broker: Any, + enabled: bool, + session_id: Optional[str] = None, + task_id: Optional[str] = None, + principal_id: Optional[str] = None, + transport_family: Optional[str] = None, + tool_call_id: Optional[str] = "", +) -> Any: + """Route one browser action through the extension-control broker. + + Parameters + ---------- + action: + Registry tool name / controller capability, e.g. ``"browser_navigate"``. + args: + Tool arguments as received from the model. Never mutated. + fallback: + The existing backend handler, called exactly once when the router + decides the extension path must not run (feature off, no scope, or + no capable controller). Must be a zero-argument callable. + broker: + Object exposing ``scope_for_session(**identity) -> scope|None``, + ``select(scope, capability) -> controller|None`` and + ``dispatch(scope, *, action, arguments, tool_call_id)``. The real + implementation is ``gateway.browser_control_broker``. + enabled: + Feature flag; false bypasses the broker entirely. + session_id/task_id: + Caller session hints forwarded to ``scope_for_session``. + principal_id/transport_family: + Server-bound caller identity. Both are mandatory when the feature is + enabled; missing values fail closed to the existing backend. + tool_call_id: + Caller tool-call id forwarded verbatim to ``dispatch``. + + Returns + ------- + The legacy backend's return value when falling back, or the controller's + completion result when routed. Exceptions from a selected controller are + propagated — the legacy backend is never retried after selection. + """ + if not enabled: + return fallback() + + if not str(principal_id or "").strip() or not str(transport_family or "").strip(): + return fallback() + + scope = broker.scope_for_session( + session_id=session_id, + task_id=task_id, + principal_id=principal_id, + transport_family=transport_family, + ) + if scope is None: + # No unambiguous attached session scope: preserve existing backend. + return fallback() + + controller = broker.select(scope, action) + if controller is None: + # No controller capable of this exact action: preserve existing backend. + return fallback() + + # A controller was selected: it is authoritative. Never retry through the + # existing backend, whatever happens here. + return broker.dispatch( + scope, action=action, arguments=args, tool_call_id=tool_call_id + ) + + +def current_tool_call_id() -> str: + """Return the active tool_call_id, or ``""`` when none is bound. + + The agent executor binds the id via + ``tools.approval.set_current_observability_context`` immediately before + registry dispatch, so the registry handler (and this router) can read it + back from the same context. Bare/offline callers have no binding. + """ + try: + from tools.approval import _approval_tool_call_id + + return _approval_tool_call_id.get() or "" + except Exception: + return "" + + +def routed_browser_handler( + action: str, + args: Dict[str, Any], + *, + fallback: Callable[[], Any], + task_id: Optional[str] = None, + session_id: Optional[str] = None, + principal_id: Optional[str] = None, + transport_family: Optional[str] = None, + tool_call_id: Optional[str] = None, +) -> Any: + """Lazy registry-handler route wrapper for ``browser_*`` tools. + + Resolves the Phase 4 feature flag and process-local broker lazily so the + default (feature off) path costs one cached config read and an immediate + fallback, and so importing ``tools.browser_tool`` never imports the + gateway. When the gateway cannot be imported or the feature is off, the + legacy handler runs unchanged. + """ + try: + from gateway.browser_control_broker import ( + browser_control_enabled, + get_browser_control_broker, + ) + except Exception as exc: # pragma: no cover - defensive, gateway always present + logger.debug( + "browser extension router unavailable (%s); using legacy backend", + exc, + ) + return fallback() + + if not browser_control_enabled(): + return fallback() + + if tool_call_id is None: + tool_call_id = current_tool_call_id() + + try: + from gateway.session_context import get_session_env + + session_id = session_id or get_session_env("HERMES_SESSION_ID", "") or None + principal_id = principal_id or get_session_env( + "HERMES_BROWSER_CONTROL_PRINCIPAL", "" + ) or None + transport_family = transport_family or get_session_env( + "HERMES_BROWSER_CONTROL_TRANSPORT_FAMILY", "" + ) or None + except Exception: + pass + + return route_browser_tool( + action, + args, + fallback=fallback, + broker=get_browser_control_broker(), + enabled=True, + session_id=session_id, + task_id=task_id, + principal_id=principal_id, + transport_family=transport_family, + tool_call_id=tool_call_id, + ) diff --git a/tools/browser_tool.py b/tools/browser_tool.py index 34f38baa1e..3cd08451f3 100644 --- a/tools/browser_tool.py +++ b/tools/browser_tool.py @@ -5390,14 +5390,29 @@ if __name__ == "__main__": # Registry # --------------------------------------------------------------------------- from tools.registry import registry, tool_error +from tools.browser_extension_router import routed_browser_handler _BROWSER_SCHEMA_MAP = {s["name"]: s for s in BROWSER_TOOL_SCHEMAS} + +def _browser_router_kw(kw: dict) -> dict: + """Identity kwargs forwarded to the extension router wrapper.""" + return { + "task_id": kw.get("task_id"), + "session_id": kw.get("session_id"), + } + + registry.register( name="browser_navigate", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_navigate"], - handler=lambda args, **kw: browser_navigate(url=args.get("url", ""), task_id=kw.get("task_id")), + handler=lambda args, **kw: routed_browser_handler( + "browser_navigate", + args, + fallback=lambda: browser_navigate(url=args.get("url", ""), task_id=kw.get("task_id")), + **_browser_router_kw(kw), + ), check_fn=check_browser_requirements, emoji="🌐", ) @@ -5405,8 +5420,13 @@ registry.register( name="browser_snapshot", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_snapshot"], - handler=lambda args, **kw: browser_snapshot( - full=args.get("full", False), task_id=kw.get("task_id"), user_task=kw.get("user_task")), + handler=lambda args, **kw: routed_browser_handler( + "browser_snapshot", + args, + fallback=lambda: browser_snapshot( + full=args.get("full", False), task_id=kw.get("task_id"), user_task=kw.get("user_task")), + **_browser_router_kw(kw), + ), check_fn=check_browser_requirements, emoji="📸", ) @@ -5414,7 +5434,12 @@ registry.register( name="browser_click", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_click"], - handler=lambda args, **kw: browser_click(ref=args.get("ref", ""), task_id=kw.get("task_id")), + handler=lambda args, **kw: routed_browser_handler( + "browser_click", + args, + fallback=lambda: browser_click(ref=args.get("ref", ""), task_id=kw.get("task_id")), + **_browser_router_kw(kw), + ), check_fn=check_browser_requirements, emoji="👆", ) @@ -5422,7 +5447,12 @@ registry.register( name="browser_type", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_type"], - handler=lambda args, **kw: browser_type(ref=args.get("ref", ""), text=args.get("text", ""), task_id=kw.get("task_id")), + handler=lambda args, **kw: routed_browser_handler( + "browser_type", + args, + fallback=lambda: browser_type(ref=args.get("ref", ""), text=args.get("text", ""), task_id=kw.get("task_id")), + **_browser_router_kw(kw), + ), check_fn=check_browser_requirements, emoji="⌨️", ) @@ -5430,7 +5460,12 @@ registry.register( name="browser_scroll", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_scroll"], - handler=lambda args, **kw: browser_scroll(direction=args.get("direction", "down"), task_id=kw.get("task_id")), + handler=lambda args, **kw: routed_browser_handler( + "browser_scroll", + args, + fallback=lambda: browser_scroll(direction=args.get("direction", "down"), task_id=kw.get("task_id")), + **_browser_router_kw(kw), + ), check_fn=check_browser_requirements, emoji="📜", ) @@ -5438,7 +5473,12 @@ registry.register( name="browser_back", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_back"], - handler=lambda args, **kw: browser_back(task_id=kw.get("task_id")), + handler=lambda args, **kw: routed_browser_handler( + "browser_back", + args, + fallback=lambda: browser_back(task_id=kw.get("task_id")), + **_browser_router_kw(kw), + ), check_fn=check_browser_requirements, emoji="◀️", ) @@ -5446,7 +5486,12 @@ registry.register( name="browser_press", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_press"], - handler=lambda args, **kw: browser_press(key=args.get("key", ""), task_id=kw.get("task_id")), + handler=lambda args, **kw: routed_browser_handler( + "browser_press", + args, + fallback=lambda: browser_press(key=args.get("key", ""), task_id=kw.get("task_id")), + **_browser_router_kw(kw), + ), check_fn=check_browser_requirements, emoji="⌨️", ) @@ -5455,7 +5500,12 @@ registry.register( name="browser_get_images", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_get_images"], - handler=lambda args, **kw: browser_get_images(task_id=kw.get("task_id")), + handler=lambda args, **kw: routed_browser_handler( + "browser_get_images", + args, + fallback=lambda: browser_get_images(task_id=kw.get("task_id")), + **_browser_router_kw(kw), + ), check_fn=check_browser_requirements, emoji="🖼️", ) @@ -5463,7 +5513,12 @@ registry.register( name="browser_vision", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_vision"], - handler=lambda args, **kw: browser_vision(question=args.get("question", ""), annotate=args.get("annotate", False), task_id=kw.get("task_id")), + handler=lambda args, **kw: routed_browser_handler( + "browser_vision", + args, + fallback=lambda: browser_vision(question=args.get("question", ""), annotate=args.get("annotate", False), task_id=kw.get("task_id")), + **_browser_router_kw(kw), + ), check_fn=check_browser_vision_requirements, emoji="👁️", ) @@ -5471,7 +5526,12 @@ registry.register( name="browser_console", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_console"], - handler=lambda args, **kw: browser_console(clear=args.get("clear", False), expression=args.get("expression"), task_id=kw.get("task_id")), + handler=lambda args, **kw: routed_browser_handler( + "browser_console", + args, + fallback=lambda: browser_console(clear=args.get("clear", False), expression=args.get("expression"), task_id=kw.get("task_id")), + **_browser_router_kw(kw), + ), check_fn=check_browser_requirements, emoji="🖥️", ) diff --git a/tui_gateway/methods_browser_control.py b/tui_gateway/methods_browser_control.py new file mode 100644 index 0000000000..ef0d7240ef --- /dev/null +++ b/tui_gateway/methods_browser_control.py @@ -0,0 +1,308 @@ +"""Browser controller registration / result routing (Phase 4 Cloud). + +The dashboard's browser controller (the extension that physically drives a +browser) registers itself over the authenticated ``/api/ws`` JSON-RPC +gateway. Everything here is bound to the **server-minted identity** that the +dashboard auth layer stamped onto the WS connection: ``hermes_cli.web_server`` +consumes the single-use ticket and records ``ws._hermes_auth_identity``, the +WS transport carries it as ``WSTransport.auth_identity``, and the client can +never name its own principal (a spoofed ``principal_id`` param is ignored and +replaced by a server-derived digest of the authenticated identity). + +Registration attaches the shared transport-neutral broker +(:mod:`gateway.browser_control_broker`) with the calling transport as owner; +broker command/cancel frames are wrapped as standard Gateway ``event`` frames +(``type`` = broker method name, ``payload`` = broker params, plus the owning +``session_id``) so the dashboard consumes the same envelope as every other +gateway event. ``browser.controller.result`` resolves a pending command only +when the request arrives on the same transport that owns the session, and +only for the exact attached scope — the broker's exact-scope ``complete`` is +the last line of defense against cross-tenant completion. + +Phase 4 is deliberately minimal: the only capability a controller may hold is +``controller.noop``, exercised end-to-end by +``tests/gateway/test_browser_control_cloud.py``. + +Note on handler globals: ``HandlerRegistry.install`` (method_ctx.py) rebinds +each handler's ``__globals__`` onto server.py's namespace, so handler bodies +may only reference names server.py defines/imports (``_ok``, ``_err``, +``_sessions``, ``_sessions_lock``, ``current_transport``, ``logger``, ...). +This module's own helpers and constants are therefore captured through +keyword-default arguments, which ``install`` preserves. +""" + +from __future__ import annotations + +import hashlib +import logging + +from .method_ctx import HandlerRegistry + +logger = logging.getLogger(__name__) + +_registry = HandlerRegistry() +method = _registry.method + +#: Capabilities a Cloud/dashboard controller may register in Phase 4. Any +#: capability outside this set is silently filtered out (fail closed: an +#: empty intersection rejects the registration). +_CONTROLLER_CAPABILITIES = frozenset({"controller.noop"}) + +#: Transport family stamped into every scope attached from this gateway. The +#: broker's exact-match contract treats it as an identity field, so an API +#: transport can never address a dashboard controller (and vice versa). +_CLOUD_TRANSPORT_FAMILY = "cloud-ticket-ws" + +#: JSON-RPC error code for identity / session / flag denials (forbidden). +_ERR_FORBIDDEN = 4403 + +#: Identity recorded for server-spawned WS clients (see +#: ``hermes_cli.dashboard_auth.ws_tickets``) — never allowed to act as a +#: browser controller. +_INTERNAL_USER_ID = "server-internal" +_INTERNAL_PROVIDER = "server-internal" + + +def _is_authenticated_identity(identity: object) -> bool: + """True for a server-minted, non-internal ``{user_id, provider}`` identity.""" + if not isinstance(identity, dict): + return False + user_id = identity.get("user_id") + provider = identity.get("provider") + if not isinstance(user_id, str) or not user_id.strip(): + return False + if not isinstance(provider, str) or not provider.strip(): + return False + if user_id == _INTERNAL_USER_ID and provider == _INTERNAL_PROVIDER: + return False + return True + + +def _principal_digest(identity: dict) -> str: + """Server-derived principal id: a digest of the server-minted identity. + + The client-supplied ``principal_id`` RPC param is never trusted; the + digest is deterministic (stable across reconnects for the same user) but + unspoofable by a peer that does not hold the authenticated identity. + """ + raw = f"{identity.get('provider')}\x00{identity.get('user_id')}" + digest = hashlib.sha256(raw.encode("utf-8")).hexdigest() + return f"principal:dashboard:{digest[:32]}" + + +def _broker_event_writer(transport: object, session_id: str): + """Wrap broker command/cancel frames as standard Gateway event frames. + + The broker's send callback receives transport-neutral envelopes like + ``{"method": "browser.controller.command", "params": {...}}``; the + dashboard speaks Gateway events, so we re-envelope them: + ``{"jsonrpc": "2.0", "method": "event", "params": {"type": , + "session_id": , "payload": }}``. + """ + + def send(frame: dict) -> None: + try: + accepted = transport.write( + { + "jsonrpc": "2.0", + "method": "event", + "params": { + "type": frame.get("method"), + "session_id": session_id, + "payload": frame.get("params"), + }, + } + ) + except Exception: + logger.exception( + "browser controller event write failed session=%s frame=%s", + session_id, + frame.get("method"), + ) + raise + if accepted is False: + raise ConnectionError("browser controller event write failed") + + return send + + +@method("browser.controller.register") +def _( + rid, + params: dict, + _family=_CLOUD_TRANSPORT_FAMILY, + _caps=_CONTROLLER_CAPABILITIES, + _forbidden=_ERR_FORBIDDEN, + _identity_ok=_is_authenticated_identity, + _digest=_principal_digest, + _event_writer=_broker_event_writer, +) -> dict: + """Attach this connection as the browser controller for one session. + + Fails closed (4403) unless *every* gate passes: + + * the ``browser.extension_control.enabled`` feature flag is on; + * the calling transport holds a server-authenticated, non-internal + identity (``WSTransport.auth_identity`` — never the RPC params); + * the named session exists in the live session registry and its + ``transport`` is exactly the calling transport; + * at least one requested capability survives the filter to + ``controller.noop``. + + The returned ``scope`` names a server-derived ``principal_id``, the + ``cloud-ticket-ws`` transport family, and the filtered capability set. + """ + from gateway import browser_control_broker + + if not browser_control_broker.browser_control_enabled(): + return _err( + rid, + _forbidden, + "browser.extension_control.enabled is not set", + ) + + transport = current_transport() + identity = getattr(transport, "auth_identity", None) + if not _identity_ok(identity): + return _err( + rid, + _forbidden, + "browser.controller.register requires an authenticated " + "non-internal identity", + ) + + session_id = str(params.get("session_id") or "") + with _sessions_lock: + session = _sessions.get(session_id) + if session is None or session.get("transport") is not transport: + return _err( + rid, + _forbidden, + "session is not owned by this transport", + ) + + controller_id = str(params.get("controller_id") or "").strip() + browser_profile_id = str(params.get("browser_profile_id") or "").strip() + profile_id = str(session.get("profile") or "").strip() + if not controller_id or not browser_profile_id or not profile_id: + return _err( + rid, + _forbidden, + "controller_id, browser_profile_id, and server session profile are required", + ) + + requested = params.get("capabilities") or [] + capabilities = frozenset(cap for cap in requested if cap in _caps) + if not capabilities: + return _err( + rid, + _forbidden, + "no permitted controller capabilities requested", + ) + + scope = browser_control_broker.ControllerScope( + principal_id=_digest(identity), + profile_id=profile_id, + session_id=session_id, + controller_id=controller_id, + browser_profile_id=browser_profile_id, + transport_family=_family, + capabilities=capabilities, + ) + + broker = browser_control_broker.get_browser_control_broker() + broker.attach( + scope, + _event_writer(transport, session_id), + owner=transport, + ) + + return _ok( + rid, + { + "scope": { + "principal_id": scope.principal_id, + "profile_id": scope.profile_id, + "session_id": scope.session_id, + "controller_id": scope.controller_id, + "browser_profile_id": scope.browser_profile_id, + "transport_family": scope.transport_family, + "capabilities": sorted(scope.capabilities), + } + }, + ) + + +@method("browser.controller.result") +def _( + rid, + params: dict, + _family=_CLOUD_TRANSPORT_FAMILY, + _forbidden=_ERR_FORBIDDEN, + _identity_ok=_is_authenticated_identity, + _digest=_principal_digest, +) -> dict: + """Deliver one controller command result back to the broker. + + Only the transport that owns the session may resolve its commands, and + only against the exact scope attached for that session (the broker's + exact-scope ``complete`` rejects any other scope). ``accepted`` is + ``False`` for unknown / already-resolved / cancelled command ids — the + broker's idempotent answer, surfaced verbatim. + """ + from gateway import browser_control_broker + + transport = current_transport() + identity = getattr(transport, "auth_identity", None) + if not _identity_ok(identity): + return _err(rid, _forbidden, "authenticated controller identity required") + session_id = str(params.get("session_id") or "") + with _sessions_lock: + session = _sessions.get(session_id) + if session is None or session.get("transport") is not transport: + return _err( + rid, + _forbidden, + "session is not owned by this transport", + ) + + command_id = str(params.get("command_id") or "") + if not command_id: + return _err(rid, _forbidden, "command_id required") + + broker = browser_control_broker.get_browser_control_broker() + scope = broker.scope_for_session( + session_id=session_id, + principal_id=_digest(identity), + transport_family=_family, + ) + if scope is None: + return _err( + rid, + _forbidden, + "no controller registered for this session", + ) + # Defense in depth: the exact-scope complete below already rejects any + # foreign scope, but the owner check makes the "same transport" rule + # explicit at this layer too. + controller = broker.select(scope, "controller.noop") + if controller is None or controller.owner is not transport: + return _err( + rid, + _forbidden, + "controller is not owned by this transport", + ) + + ok = params.get("ok") is True + accepted = broker.complete( + command_id, + scope=scope, + ok=ok, + result=params.get("result") if ok else params.get("error"), + ) + return _ok(rid, {"accepted": accepted}) + + +def register(server) -> None: + """Bind this module's handlers onto ``server``'s globals and registry.""" + _registry.install(server) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 5b6a177569..ff08824ea3 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -3433,6 +3433,8 @@ def _set_session_context( # it instead of falling back to the gateway launch dir. resolved = cwd if cwd is not None else _cwd_for_session_key(session_key) source = _resolve_session_platform() + browser_control_principal = "" + browser_control_transport_family = "" # Derive the live conversation id so terminal/execute_code subprocesses # can read HERMES_SESSION_ID. Without this, set_session_vars leaves the # session-id contextvar as "" (explicitly empty), and the subprocess-env @@ -3450,11 +3452,22 @@ def _set_session_context( session_id = ( getattr(sess.get("agent"), "session_id", None) or session_key ) + transport = sess.get("transport") + identity = getattr(transport, "auth_identity", None) + if _methods_browser_control._is_authenticated_identity(identity): + browser_control_principal = ( + _methods_browser_control._principal_digest(identity) + ) + browser_control_transport_family = ( + _methods_browser_control._CLOUD_TRANSPORT_FAMILY + ) break return set_session_vars( session_key=session_key, session_id=session_id, source=source, + browser_control_principal=browser_control_principal, + browser_control_transport_family=browser_control_transport_family, cwd=resolved, ui_session_id=ui_session_id, cron_session="", @@ -15621,6 +15634,7 @@ def _mcp_summarize_server(name, cfg): # noqa: E402 # Imported at the end of this module so every global the handlers close # over already exists; register() rebinds them onto this namespace. from . import ( # noqa: E402 + methods_browser_control as _methods_browser_control, methods_complete as _methods_complete, methods_config as _methods_config, methods_images as _methods_images, @@ -15631,6 +15645,7 @@ from . import ( # noqa: E402 ) for _m in ( + _methods_browser_control, _methods_session, _methods_prompt, _methods_config, diff --git a/tui_gateway/ws.py b/tui_gateway/ws.py index 073c4ac149..1f2c7b0316 100644 --- a/tui_gateway/ws.py +++ b/tui_gateway/ws.py @@ -89,10 +89,19 @@ class WSTransport: loop: asyncio.AbstractEventLoop, *, peer: str = "unknown", + auth_identity: dict | None = None, ) -> None: self._ws = ws self._loop = loop self._peer = peer + #: Server-verified identity carried from the WS-upgrade credential + #: (dashboard ticket / internal credential) — stamped by + #: ``hermes_cli.web_server._ws_auth_reason`` onto the WS object and + #: passed through ``handle_ws``. None for transports that + #: authenticated via the legacy token path or stdio. RPC params can + #: never populate this: it is the only identity authority for + #: browser-controller registration. + self.auth_identity = auth_identity self._closed = False # Token-coalescing buffer (CF-2). Streamed token frames land here and a # short timer flushes the batch. The lock guards the buffer + the @@ -283,8 +292,16 @@ def _disable_nagle(ws: Any) -> None: _log.debug("ws TCP_NODELAY skip: %s", exc) -async def handle_ws(ws: Any) -> None: - """Run one WebSocket session. Wire-compatible with ``tui_gateway.entry``.""" +async def handle_ws(ws: Any, *, auth_identity: dict | None = None) -> None: + """Run one WebSocket session. Wire-compatible with ``tui_gateway.entry``. + + *auth_identity* is the server-minted ``{user_id, provider}`` recorded at + WS-upgrade authentication (``hermes_cli.web_server._ws_auth_reason``); it + is stored on the transport as ``WSTransport.auth_identity`` and is the + only identity authority for browser-controller registration. Existing + callers (stdio-free harnesses, the embedded TUI child) omit it and get a + ``None`` transport identity — unchanged behaviour. + """ peer = _ws_peer_label(ws) transport: WSTransport | None = None messages = 0 @@ -301,7 +318,12 @@ async def handle_ws(ws: Any) -> None: _disable_nagle(ws) _log.info("ws accepted peer=%s", peer) - transport = WSTransport(ws, asyncio.get_running_loop(), peer=peer) + transport = WSTransport( + ws, + asyncio.get_running_loop(), + peer=peer, + auth_identity=auth_identity, + ) # resolve_skin() reads config + initializes the skin engine — # synchronous I/O + CPU work that should not block the event loop @@ -431,6 +453,22 @@ async def handle_ws(ws: Any) -> None: detached_sessions = 0 if transport is not None: server.unregister_live_transport(transport) + + # Owner-safely detach browser controllers this transport + # registered (Phase 4 Cloud). The socket itself is closing, so no + # peer cancel write is attempted; every server-side pending command + # is still failed closed immediately. + try: + from gateway.browser_control_broker import ( + get_browser_control_broker, + ) + + get_browser_control_broker().detach_owner( + transport, notify_controller=False + ) + except Exception: + _log.exception("ws browser-controller detach failed peer=%s", peer) + transport.close() try: