diff --git a/AGENTS.md b/AGENTS.md index ab7b0986..c3c3241c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -117,7 +117,7 @@ and deserialized through a registry. | Package | Contents | |---|---| -| `env/` | `Env` base class: a subclass declares `type`, implements `from_dict` and `async deploy(...) -> DeployedEnv`, and may override `reset`, `close` and `load_file_artifact_universe`; teardown closes each env a run deployed before terminating its sandboxes by id. Built-ins in `env/envs/`: `MCPServerEnv` (`mcp_server`), `MultiEnv` (`multi`, MCP servers plus websites), `ServiceDBEnv` (`service_db`, PostgreSQL), `WebsiteEnv` (`website`, one site: a server image, `backend_image` in its env.toml, serving the pages and API, ready when its card's `website/v1` `ready` answers a `WebsiteReadiness` that says so, as `agentenv_protocol.website` (`website_readiness`, `env_readiness`) defines a website env's health for the deploy and `DeployedEnv.readiness()` alike, or, for a card without `ready`, when its `health` path answers 200, which is deprecated; the gateway container's `env/gateway/constants.py` mirrors the URI and op, a test keeping them equal; a separate frontend image is deprecated, warned of at put and optional in a bundle, and a stored `backend_docker_image_artifact` reads as the server image), `GatewayEnv` (`gateway_server`), `ComputerDriverEnv` (`computer_driver`, `env/envs/computer/driver_env.py`: a computer driver's files and what its class declares for each platform it supports, read at put in a process of its own, `agentenv_protocol.computer._declare`, where the driver's own dependencies needn't be installed, with the `port` it serves on and its install by OS, `[install.]` (`env/envs/computer/utils/install.py`): files, apt `packages` with extra `sources`, a `python` requirements file for a venv, `downloads` by URL and sha256, `run` scripts, the services the base's init runs, and `ports`; it deploys only as a computer env's driver), `ComputerEnv` (`computer`, in `env/envs/computer/`: a base computer and a driver, matched at put by gate 1 before anything is stored, then composed by `composer.py` into an image for each form and arch on a builder of the put's sandbox provider, by default the env's (`utils/builder.py`; `--sandbox` on `env build`, `bundle register` and `run`): this machine's Docker for `local`, else a VM of the provider, which fetches the put's stored files by signed URL, takes the rest inline and pushes to the image store itself, so it builds only with stores a remote VM reaches; each arch goes to the first link that runs it natively, else one that emulates it (a VM registers QEMU; `--no-emulate` leaves it `skipped` instead, for a later put), and its pair records the provider that built it (`builder`); a run or register that reuses a computer env composes what it has left (`pending`), a run only the forms its sandbox's deploys boot (`booted_forms`: on each link, the first form it boots natively, by the deploy's own rule, `boot.first_native`; a compose of a disk adds its container), and a deploy composes first what it boots that no compose started (`ComputerEnv.compose_unstarted`); tagged `agentenv-computers:` in the image store by a key of the image's Dockerfile, build context and lock so any env with the same inputs reuses it; every layer's packages, requirements and downloads are resolved at put in a container of the base for each arch (`utils/packages.py`, `utils/resolve_in_base.py`), stored in the object store by sha256 under `agentenv-packages/`, and listed in a lock stored by a digest of its inputs, which any env with the same inputs reuses, and builds install only the lock's files, every install step `RUN --network=none`; each pair's status goes into its metadata, `composed`, the arches built are written as its computer, a `computer` artifact over one index, and `agent-env env build` resumes what a put left; gate 3 (`utils/probing.py`) then boots each native build on the put's sandbox provider and probes it at `full`, making it `ready` or `failed`, and a build no provider runs natively stays `unverified`. Its deploy, through the `computer` env provider (`providers/env_providers/env_computer_provider.py`), boots its first `ready` build on a sandbox provider that runs its arch natively (`native_platforms()`), with no gateway in front and the driver's card as the env card, then checks its `display`; `deploy_env` `env_params` take a `form`, `computer_settings` (deploy-phase values of the base's settings; a secret one, here or on the env, only as a bare `secret:`/`env:` reference) and `driver_settings` (the driver's `ComputerSettings`, such as its default coordinate space); a `network` that allows nothing runs it on local Docker's internal network behind a relay (`providers/sandbox_providers/local_docker/relay/`); `reset` and `close` have the driver end its recordings first (`computer-recording/v1` `end_all`, each uploading what's left and its manifest under the grant it started with), and `reset` then boots a fresh container in its place (`reset_container`), keeping the computer's address but not an agent's MCP session to the old driver; its `load_file_artifact_universe` writes files under the session user's home through the driver's `data/add`, `Downloads` by default, each file's bytes through a read grant when the store's grants reach it and its network allows everything, else inline (`env/envs/computer/utils/inside.py`)), `ComputerAppEnv` (`computer_app`, `env/envs/computer/app_env.py`: an app on a computer, a `ComputerAppEnvironment` class its env.toml's `module` names, read at put by the same `_declare`, which declares each OS with its own decorator (`@linux(...)`, or `@install(os_type, spec)` for any OS type) holding its install, its launch and, optionally, a skill and settings and build settings that add to the class's there, and install steps and ready checks for every OS or those their `os=` names; everything that reads its settings, launch or steps asks for the computer's OS type; deployed alone it is a one-app computer env, a `ComputerEnv` of the computer and driver its env.toml names, the composer installing it under `/opt/agentenv/apps//` with its services as supervisord programs beside its own server (`agentenv_protocol.computer.serve_app`, in a venv with the protocol's `app` extra, on `.agentenv/server.sock` in its folder), which adds its launcher, launches a `boot` app once per boot and says through `computer-app/v1` whether it's ready, as the driver's `computer-ready` asks, its card named after it, and `env_params` `app_settings` handing it values of its `Settings` in `AGENTENV_APP_SETTINGS`, a secret one only as a bare reference, over its env.toml's `[settings]` and a computer env's `settings` beside it in `apps`, each by top-level field (`params.app_settings`; checked at put, a required one left to the deploy; the deploy record keeps the merged values as `app_settings`, secrets withheld); its env.toml's `build_settings`, values of its `BuildSettings` checked at put, are written to `.agentenv/build_settings.json` in its folder, so they're part of the image's key, its install steps read them, and a deploy can't change them). An env type whose services take environment data defines `load_environment_artifact` and `load_environment_universe_artifact`; `load_artifact` refuses an environment artifact aimed at a type that doesn't, at save and before it loads. An env built on the protocol SDK can serve agents a skill (`@agent_skill`, `skill/v1`), which `add_skills` `env_ids` installs under its environment name (behind a gateway, each server's under that server's name); a skill with files beside its SKILL.md goes as a bundle stored once per content under `agentenv-env-skills/` in the object store. `env/gateway/gateway.py` is the env-side gateway server, container-only code. Its site route, `/website//`, never forwards a site's data plane or card (`/agentenv/…`, `/.well-known/agent-env.json`), which agentenv reaches through `/svc//`; a site with no frontend image is served by its server image there. While its clock is armed it serves agents `get_time` (`env/gateway/get_time.py`), which answers in UTC and, given an IANA `time_zone`, also in that zone; executors' working `hours` (`env/gateway/triggers.py`) take a fixed `utc_offset` or an IANA `time_zone`. | +| `env/` | `Env` base class: a subclass declares `type`, implements `from_dict` and `async deploy(...) -> DeployedEnv`, and may override `reset`, `close` and `load_file_artifact_universe`; teardown closes each env a run deployed before terminating its sandboxes by id. Built-ins in `env/envs/`: `MCPServerEnv` (`mcp_server`), `MultiEnv` (`multi`, MCP servers plus websites), `ServiceDBEnv` (`service_db`, PostgreSQL), `WebsiteEnv` (`website`, one site: a server image, `backend_image` in its env.toml, serving the pages and API, ready when its card's `website/v1` `ready` answers a `WebsiteReadiness` that says so, as `agentenv_protocol.website` (`website_readiness`, `env_readiness`) defines a website env's health for the deploy and `DeployedEnv.readiness()` alike, or, for a card without `ready`, when its `health` path answers 200, which is deprecated; the gateway container's `env/gateway/constants.py` mirrors the URI and op, a test keeping them equal; a separate frontend image is deprecated, warned of at put and optional in a bundle, and a stored `backend_docker_image_artifact` reads as the server image), `GatewayEnv` (`gateway_server`), `ComputerDriverEnv` (`computer_driver`, `env/envs/computer/driver_env.py`: a computer driver's files and what its class declares for each platform it supports, read at put in a process of its own, `agentenv_protocol.computer._declare`, where the driver's own dependencies needn't be installed, with the `port` it serves on and its install by OS, `[install.]` (`env/envs/computer/utils/install.py`): files, apt `packages` with extra `sources`, a `python` requirements file for a venv, `downloads` by URL and sha256, `run` scripts, the services the base's init runs, and `ports`; it deploys only as a computer env's driver), `ComputerEnv` (`computer`, in `env/envs/computer/`: a base computer and a driver, matched at put by gate 1 before anything is stored, then composed by `composer.py` into an image for each form and arch on a builder of the put's sandbox provider, by default the env's (`utils/builder.py`; `--sandbox` on `env build`, `bundle register` and `run`): this machine's Docker for `local`, else a VM of the provider, which fetches the put's stored files by signed URL, takes the rest inline and pushes to the image store itself, so it builds only with stores a remote VM reaches; each arch goes to the first link that runs it natively, else one that emulates it (a VM registers QEMU; `--no-emulate` leaves it `skipped` instead, for a later put), and its pair records the provider that built it (`builder`); a run or register that reuses a computer env composes what it has left (`pending`), a run only the forms its sandbox's deploys boot (`booted_forms`: on each link, the first form it boots natively, by the deploy's own rule, `boot.first_native`; a compose of a disk adds its container), and a deploy composes first what it boots that no compose started (`ComputerEnv.compose_unstarted`); tagged `agentenv-computers:` in the image store by a key of the image's Dockerfile, build context and lock so any env with the same inputs reuses it; every layer's packages, requirements and downloads are resolved at put in a container of the base for each arch (`utils/packages.py`, `utils/resolve_in_base.py`), stored in the object store by sha256 under `agentenv-packages/`, and listed in a lock stored by a digest of its inputs, which any env with the same inputs reuses, and builds install only the lock's files, every install step `RUN --network=none`; each pair's status goes into its metadata, `composed`, the arches built are written as its computer, a `computer` artifact over one index, and `agent-env env build` resumes what a put left; gate 3 (`utils/probing.py`) then boots each native build on the put's sandbox provider and probes it at `full`, making it `ready` or `failed`, and a build no provider runs natively stays `unverified`. Its deploy, through the `computer` env provider (`providers/env_providers/env_computer_provider.py`), boots its first `ready` build on a sandbox provider that runs its arch natively (`native_platforms()`), with no gateway in front and the driver's card as the env card, then checks its `display`; `deploy_env` `env_params` take a `form`, `computer_settings` (deploy-phase values of the base's settings; a secret one, here or on the env, only as a bare `secret:`/`env:` reference) and `driver_settings` (the driver's `ComputerSettings`, such as its default coordinate space); a `network` that allows nothing runs it on local Docker's internal network behind a relay (`providers/sandbox_providers/local_docker/relay/`); `reset` and `close` have the driver end its recordings first (`computer-recording/v1` `end_all`, each uploading what's left and its manifest under the grant it started with), and `reset` then boots a fresh container in its place (`reset_container`), keeping the computer's address but not an agent's MCP session to the old driver; its `load_file_artifact_universe` writes files under the session user's home through the driver's `data/add`, `Downloads` by default, each file's bytes through a read grant when the store's grants reach it and its network allows everything, else inline (`env/envs/computer/utils/inside.py`)), `ComputerAppEnv` (`computer_app`, `env/envs/computer/app_env.py`: an app on a computer, a `ComputerAppEnvironment` class its env.toml's `module` names, read at put by the same `_declare`, which declares each OS with its own decorator (`@linux(...)`, or `@install(os_type, spec)` for any OS type) holding its install, its launch and, optionally, a skill and settings and build settings that add to the class's there, and install steps and ready checks for every OS or those their `os=` names; everything that reads its settings, launch or steps asks for the computer's OS type; deployed alone it is a one-app computer env, a `ComputerEnv` of the computer and driver its env.toml names, the composer installing it under `/opt/agentenv/apps//` with its services as supervisord programs beside its own server (`agentenv_protocol.computer.serve_app`, in a venv with the protocol's `app` extra, on `.agentenv/server.sock` in its folder), which adds its launcher, launches a `boot` app once per boot and says through `computer-app/v1` whether it's ready, as the driver's `computer-ready` asks, its card named after it, and `env_params` `app_settings` handing it values of its `Settings` in `AGENTENV_APP_SETTINGS`, a secret one only as a bare reference, over its env.toml's `[settings]` and a computer env's `settings` beside it in `apps`, each by top-level field (`params.app_settings`; checked at put, a required one left to the deploy; the deploy record keeps the merged values as `app_settings`, secrets withheld); its env.toml's `build_settings`, values of its `BuildSettings` checked at put, are written to `.agentenv/build_settings.json` in its folder, so they're part of the image's key, its install steps read them, and a deploy can't change them). An env type whose services take environment data defines `load_environment_artifact` and `load_environment_universe_artifact`; `load_artifact` refuses an environment artifact aimed at a type that doesn't, at save and before it loads. An env built on the protocol SDK can serve agents a skill (`@agent_skill`, `skill/v1`), which `add_skills` `env_ids` installs under its environment name (behind a gateway, each server's under that server's name); a skill with files beside its SKILL.md goes as a bundle stored once per content under `agentenv-env-skills/` in the object store. `env/gateway/gateway.py` is the env-side gateway server, container-only code. Its site route, `/website//`, never forwards a site's data plane or card (`/agentenv/…`, `/.well-known/agent-env.json`), which agentenv reaches through `/svc//`; a site with no frontend image is served by its server image there. Its CDP route, `/browser//cdp//` (`env/gateway/browser_proxy.py`), reaches the website browser behind the `/svc` child ``, found from that child's card (it, or an app among its children, offering `browser/v1` `cdp` with `/cdp/` in its `computer-app/v1` `agent_routes`): it relays the browser's discovery and WebSocket both ways, each leg pinged every 20 s, and every CDP address the browser hands out, in `json/version`, `json/list` and its `cdp` answer through `/svc//`, comes back at the gateway, everything before `/cdp//` becoming the address the caller reached the gateway at and `/browser/`; each connection is a `browser_connection` trajectory event, without its token. While its clock is armed it serves agents `get_time` (`env/gateway/get_time.py`), which answers in UTC and, given an IANA `time_zone`, also in that zone; executors' working `hours` (`env/gateway/triggers.py`) take a fixed `utc_offset` or an IANA `time_zone`. | | `artifact/` | Immutable versioned artifacts; every `put()` creates a new version. The document goes to the document store, binary payloads to the object store. Every artifact but an environment universe, whose `metadata` key holds its legacy metadata file ids, carries a `metadata` dict outside its versioned fields, changed by compare-and-swap (`merge_metadata`). Built-ins: `CliArtifact` (`cli`), `ComputerArtifact` (`computer`, a base computer: its architectures, OS, session user, settings and container form, whose image `put` reads from its registry into `metadata.validated_images`), `FileArtifact` (`file`), `FileArtifactUniverse` (`file_artifact_universe`), `DockerImageArtifact` (`docker_image`), `EnvironmentArtifact` (`environment`: its environment's files by path, stored as the one wrapped `file_artifact_ref` every environment artifact had when it has one file at its filename, else as `file_artifact_refs`; `get_file_artifacts()` reads either, and a load hands each file to the env's `data/add` with its path, which a pre-v1 load, taking one file, refuses), `EnvironmentUniverseArtifact` (`environment_universe`), `SkillArtifact` (`skill`); `[artifacts] type_aliases` maps another stored spelling of a type onto one of them. A `DockerImageArtifact` holds a tar.gz of its image, or, written by `put_ref` (`--image-ref` on `env mcp-server put` and `a2a-agent put`), only a registry reference with its tag pinned to a digest, which sandboxes pull, or a published image's tag with its `image_digest` and its build context, which sandboxes pull by the digest and build from the context when they can't (the infra images, and a bundle agent built from a shipped agent's sources, `Plan.published`), or, written by `put_context`, only its build context, which a VM sandbox builds before running it (`VmSandbox.load_docker_images`) and Modal builds before creating its container (`ModalSandboxProvider.prepare_image`, outside a fallback chain's deadline); anything else that runs images by name (a container on the local provider, a lone server on any provider but Modal) refuses such an image. A build context (`utils/docker/context.py`) is what `docker build` would send, `.dockerignore` applied, written so equal files give equal bytes and an equal `source_digest`. | | `task/`, `task_step/` | A `Task` holds its `TaskStep`s inline; `Task.run()` executes them as a DAG. `depends_on` (None means all prior steps) gates a step, independent steps run concurrently, `fail_task_on_error` makes a failure fatal or tolerated, `retry_config` rolls a failed span back through the step journal and re-dispatches it. Built-in steps live in `task_step/task_steps/` (`deploy_env`, `deploy_agent`, `prompt_agent`, the verifiers under `verifiers/`, among them `start_computer_recording` and `end_computer_recording`, and more). On an env whose card serves computer-exec/v1 (a computer), `run_code` with `env_id` and `run_container_unit_tests_verifier` with `env_id` run inside it as its session user through the driver: files staged with `data/add` and read with `data/get`, commands run with `run`; `env_outcome_verifier` hands a `verify` that takes an `env_url` keyword the env's address. A computer's framework-only routes (everything but MCP and the card) take the per-deploy control token its provider boots it with (`DeployedEnv.control_token`, `AGENTENV_CONTROL_TOKEN`); framework calls send `deployed.control_headers`, `Task.run` runs every step under `protocol_v1.authorized` with each deployed env's (a `verify` included), and a `run_code` script inside the computer gets `input["env"]["headers"]`. `teardown_run` and the `teardown_sandboxes` step share one core (`task/teardown.py` `tear_down`): the recordings the run left open end with fresh grants, each env still up closes (`Env.close`; a computer env ends any recording still open), what stayed open is registered from its namespace, then every sandbox is terminated by id. `task/episode/` runs a task with some of its agents played from outside, as a training loop does (`EpisodeSpec`). | | `store/` | Four store ABCs with local and cloud implementations: `DocumentStore` (SQLite, MongoDB, DynamoDB, Firestore with MongoDB compatibility), `ObjectStore` (filesystem, S3, Cloud Storage), `ImageStore` (local OCI registry, ECR), `SecretStore` (env vars or file, AWS Secrets Manager, Google Cloud Secret Manager). `VersionedEntityStore` implements the shared versioned get/put logic, `QueryBuilder` is the immutable chained query API, `store/base.py` holds the error types. `store/routing.py` is the local namespaces, `@local` and `@hf` (`is_local_id`), each with a documents file of its own: while the CLI or a `build_app` server runs, a local id and what a local task's run records go to local stores beside the configured ones, and that run's writes to the configured stores are refused. A new backend must pass the conformance kits in `tst/store/`. | diff --git a/src/agent_env/env/gateway/browser_proxy.py b/src/agent_env/env/gateway/browser_proxy.py new file mode 100644 index 00000000..4f5a2cc9 --- /dev/null +++ b/src/agent_env/env/gateway/browser_proxy.py @@ -0,0 +1,330 @@ +"""The gateway's CDP route: ``/browser//cdp//...`` reaches the website browser behind the ``/svc`` child +````, its discovery and its WebSocket both, and every CDP address the browser hands out, in discovery or in its +``cdp`` answer, comes back naming the gateway at the address the caller reached it at. + +The browser builds those addresses from the address its own server was reached at, so behind the gateway they would +name the browser's host and a client following one would leave the gateway. The gateway rewrites them by path, +replacing everything before ``/cdp//`` with its own address and ``/browser/``, so a rewrite holds whatever +host an address was minted at.""" +from __future__ import annotations + +import asyncio +import contextlib +import logging +import posixpath +import re +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any +from urllib.parse import urlsplit + +import httpx +import websockets +from starlette.requests import Request +from starlette.responses import JSONResponse, Response +from starlette.websockets import WebSocket, WebSocketDisconnect +from websockets.asyncio.client import ClientConnection, connect + +# Relative import — gateway runs as a standalone container with no agent_env.* +from .constants import EXT_BROWSER_URI, EXT_COMPUTER_APP_URI +from .rest_proxy import HOP_BY_HOP, fetch_child_card, operations, upstream_headers, upstream_url + +logger = logging.getLogger(__name__) + +#: The route under a browser's own that its CDP addresses sit under, which its card opens to agents. +CDP_ROUTE = "/cdp/" +#: How often each leg of a relayed connection is pinged, so an idle one outlives a tunnel's idle timeout. +PING_INTERVAL_S = 20.0 +#: The fields of a discovery answer, ``json/version`` or each ``json/list`` entry, that carry a CDP address. +DISCOVERY_FIELDS = frozenset({"webSocketDebuggerUrl", "devtoolsFrontendUrl"}) +#: The fields of a browser's ``cdp`` answer that do. +CDP_ANSWER_FIELDS = frozenset({"http_url", "ws_url"}) +#: A Host header that names a host, and its port or not, and nothing else. +HOST = re.compile(r"(?:[A-Za-z0-9.-]+|\[[0-9A-Fa-f:.]+\])(?::[0-9]{1,5})?") +#: A CDP address's path from its token on. +ADDRESS = re.compile(r"/cdp/[^/?#]+(?:/[^?#]*)?$") +#: A CDP address's token, which no log line shows. +TOKEN = re.compile(r"(/cdp/)[^/\s?#\"']+") +#: A WebSocket close code that means the connection ended without one, which a close can't send on. +UNSENDABLE_CLOSES = (1005, 1006, 1015) +#: A close for a connection the browser can't be found for, or refuses. +REFUSED = 1008 +#: A close for a connection the browser didn't answer, or that failed in between. +UNREACHABLE = 1011 + + +@dataclass(frozen=True) +class Browser: + """A website browser behind a ``/svc`` child: where its routes sit under the child's (``prefix``), and the path + of its ``cdp`` method, as the child's card names them.""" + + prefix: str + cdp_path: str + + +def browsers(card: dict) -> list[Browser]: + """The website browsers a ``/svc`` child's card, as the child serves it, offers: the child itself, or each app + among its children, as a computer's are, its routes under ``/svc/``. One counts only when its card opens + its CDP route to agents, the only one the gateway relays without the child's control token.""" + offered = [("", card)] + [(f"/svc/{child['name']}", child) for child in card.get("children_environments") or [] + if isinstance(child, dict) and isinstance(child.get("name"), str)] + found = [] + for prefix, browser_card in offered: + cdp_path = _method_path(browser_card, EXT_BROWSER_URI, "cdp") + if cdp_path is not None and CDP_ROUTE in (_params(browser_card, EXT_COMPUTER_APP_URI).get("agent_routes") or []): + found.append(Browser(prefix, cdp_path)) + return found + + +def at_gateway(url: str, base: str, key: str) -> str: + """``url``, a CDP address a browser handed out, at the gateway: everything before its ``/cdp//`` becomes + ``base`` and ``/browser/``, a WebSocket URL staying one. A value that isn't such an address is kept.""" + parts = urlsplit(url) + match = ADDRESS.search(parts.path) + if parts.scheme not in ("http", "https", "ws", "wss") or match is None: + return url + secure, host = base.startswith("https://"), base.split("://", 1)[1] + scheme = ("wss" if secure else "ws") if parts.scheme in ("ws", "wss") else ("https" if secure else "http") + return f"{scheme}://{host}/browser/{key}{match.group(0)}" + (f"?{parts.query}" if parts.query else "") + + +def rewritten(body: Any, fields: frozenset[str], base: str, key: str) -> Any: + """A JSON answer, an object or a list of them, with each of ``fields`` holding an address at the gateway.""" + if isinstance(body, list): + return [rewritten(item, fields, base, key) for item in body] + if isinstance(body, dict): + return {name: at_gateway(value, base, key) if name in fields and isinstance(value, str) else value + for name, value in body.items()} + return body + + +def public_base(connection: Request | WebSocket) -> str | None: + """The address the client reached the gateway at, scheme, host and port: the Host it sent, with the scheme a + TLS-terminating tunnel in front says it was reached by (``X-Forwarded-Proto``), else the connection's own. + None for a Host that isn't a host and port. A rewrite with it goes back only to the client that sent it.""" + host = connection.headers.get("host", "") + if not HOST.fullmatch(host): + return None + proto = connection.headers.get("x-forwarded-proto", "").split(",")[0].strip().lower() + if proto not in ("http", "https"): + proto = "https" if connection.url.scheme in ("https", "wss") else "http" + return f"{proto}://{host}" + + +class TokenRedaction(logging.Filter): + """Takes the token out of each CDP address a log line names: a request's path, or a URL forwarded to.""" + + def filter(self, record: logging.LogRecord) -> bool: + message = record.getMessage() + if CDP_ROUTE in message: + record.msg, record.args = TOKEN.sub(r"\1", message), () + return True + + +class BrowserRoutes: + """The ``/browser//cdp//`` route over a gateway's ``/svc`` children, and the rewrite of a browser's + ``cdp`` answer on their ``/svc`` route. ``log_event`` writes a trajectory event, ``attribution`` names a role's + caller for one, and ``role_of`` reads the role a connection's ``AgentEnv-Role`` header names.""" + + def __init__(self, rest_proxy_urls: dict[str, str], timeout: httpx.Timeout, + log_event: Callable[[dict], Awaitable[Any]], attribution: Callable[[str], dict], + role_of: Callable[[Request | WebSocket], str]): + self._urls = rest_proxy_urls + self._timeout = timeout + self._log_event = log_event + self._attribution = attribution + self._role_of = role_of + self._found: dict[str, list[Browser]] = {} + + async def browser(self, key: str) -> Browser | Response: + """The website browser behind the ``/svc`` child ``key``, from its card, read on first use and kept once it + names one: a 404 for a key no child has or whose card offers none, a 409 for one offering several, and a 503 + while its card can't be read. A card naming none is read again next time, since a computer lists an app + whose server isn't answering yet by its name alone.""" + base = self._urls.get(key) + if base is None: + return _error(404, f"no browser {key!r}") + found = self._found.get(key) + if found is None: + async with httpx.AsyncClient() as client: + card = await fetch_child_card(client, key, base) + if card is None: + return _error(503, f"{key}'s card can't be read") + found = browsers(card) + if found: + self._found[key] = found + if len(found) != 1: + return _error(409 if found else 404, f"{key} has {len(found) or 'no'} website browsers") + return found[0] + + async def discovery(self, request: Request) -> Response: + """A request under a CDP address, discovery's ``json/version`` and ``json/list``, forwarded to the browser, + with each address in a JSON answer at the gateway.""" + key = request.path_params["key"] + browser = await self.browser(key) + if isinstance(browser, Response): + return browser + base = public_base(request) + if base is None: + return _error(400, "the Host header isn't a host and port") + headers = [(name, value) for name, value in request.headers.raw if name.lower() not in HOP_BY_HOP] + try: + async with httpx.AsyncClient(timeout=self._timeout) as client: + answer = await client.request(request.method, self._target(key, browser, request), headers=headers) + except httpx.HTTPError as e: + return _error(502, f"the browser didn't answer: {type(e).__name__}") + return _answered(answer, DISCOVERY_FIELDS, base, key) + + async def cdp_answer(self, request: Request, base_url: str, role: str) -> Response | None: + """A browser's ``cdp`` answer to a call through its child's ``/svc//`` route, with its addresses at the + gateway; None for any other ``/svc`` request, which goes through as it is.""" + path = posixpath.normpath("/" + request.path_params.get("path", "")) + if request.method != "POST" or not path.endswith("/cdp"): + return None + key = request.path_params["service_name"] + browser = await self.browser(key) + if not isinstance(browser, Browser) or path != browser.cdp_path: + return None + base = public_base(request) + if base is None: + return _error(400, "the Host header isn't a host and port") + try: + async with httpx.AsyncClient(timeout=self._timeout) as client: + answer = await client.post(upstream_url(base_url, request), content=await request.body(), + headers=upstream_headers(request, role)) + except httpx.HTTPError as e: + return _error(502, f"the browser didn't answer: {type(e).__name__}") + return _answered(answer, CDP_ANSWER_FIELDS, base, key) + + async def relay(self, websocket: WebSocket) -> None: + """A WebSocket under a CDP address relayed both ways to the browser until either side closes, each leg pinged + every ``PING_INTERVAL_S``; refused before it opens when its browser can't be found, refuses it or doesn't + answer. Each attempt is a ``browser_connection`` trajectory event once it ends, without its token.""" + key = websocket.path_params["key"] + event = {"event_type": "browser_connection", "browser": key, "path": websocket.path_params.get("path", ""), + **self._attribution(self._role_of(websocket))} + browser = await self.browser(key) + if isinstance(browser, Response): + await websocket.close(code=REFUSED) + await self._log_event({**event, "opened_at": None, "closed_at": _now(), "close_code": REFUSED, + "closed_by": "gateway", "status": browser.status_code}) + return + headers = [(name, value) for name, value in websocket.headers.items() + if name.encode() not in HOP_BY_HOP and not name.startswith("sec-websocket-")] + offered = [p.strip() for p in websocket.headers.get("sec-websocket-protocol", "").split(",") if p.strip()] + try: + upstream = await connect( + _websocket_form(self._target(key, browser, websocket)), additional_headers=headers, + subprotocols=offered or None, max_size=None, compression=None, user_agent_header=None, + ping_interval=PING_INTERVAL_S, ping_timeout=PING_INTERVAL_S, open_timeout=self._timeout.connect) + except websockets.InvalidStatus as e: # an address the browser didn't mint, or a route it refuses + await websocket.close(code=REFUSED) + await self._log_event({**event, "opened_at": None, "closed_at": _now(), "close_code": REFUSED, + "closed_by": "browser", "status": e.response.status_code}) + return + except (OSError, ValueError, asyncio.TimeoutError, websockets.WebSocketException) as e: + await websocket.close(code=UNREACHABLE) + await self._log_event({**event, "opened_at": None, "closed_at": _now(), "close_code": UNREACHABLE, + "closed_by": "gateway", "error": type(e).__name__}) + return + opened, closed_by, code, error = _now(), "gateway", 1001, None + try: + async with upstream: + await websocket.accept(subprotocol=upstream.subprotocol) + closed_by, code, error = await _both_ways(websocket, upstream) + finally: + if error is not None: + logger.warning(f"browser {key}: a relayed connection failed with {error}") + await self._log_event({**event, "opened_at": opened, "closed_at": _now(), "close_code": code, + "closed_by": closed_by, **({"error": error} if error else {})}) + + def _target(self, key: str, browser: Browser, connection: Request | WebSocket) -> str: + """The browser's own URL for a request under ``/browser//``: the child's, the browser's prefix and the + path from ``cdp/`` on as it came, still percent-encoded, and its query.""" + raw = (connection.scope.get("raw_path") or connection.url.path.encode()).decode("latin-1") + tail = raw.split("/", 3)[3] # "/browser//cdp/..." -> "cdp/..." + query = connection.scope.get("query_string", b"").decode("latin-1") + return f"{self._urls[key].rstrip('/')}{browser.prefix}/{tail}" + (f"?{query}" if query else "") + + +async def _both_ways(client: WebSocket, browser: ClientConnection) -> tuple[str, int, str | None]: + """Each message from either side sent to the other until one side closes, then the other closed with its code: + which side closed first, and its code. When neither closed but relaying failed, both are closed as ``UNREACHABLE`` + and the failure's type is the third value, never its message, which may hold the page's data.""" + ended: list[tuple[str, int]] = [] + + async def inbound() -> None: + while True: + message = await client.receive() + if message["type"] == "websocket.disconnect": + code = message.get("code") or 1000 + ended.append(("client", code)) + await browser.close(1000 if code in UNSENDABLE_CLOSES else code) + return + await browser.send(message["text"] if message.get("text") is not None else message["bytes"]) + + async def outbound() -> None: + with contextlib.suppress(websockets.ConnectionClosed): + async for message in browser: + if isinstance(message, str): + await client.send_text(message) + else: + await client.send_bytes(message) + code = browser.close_code or 1000 + ended.append(("browser", code)) + with contextlib.suppress(RuntimeError, WebSocketDisconnect): + await client.close(1000 if code in UNSENDABLE_CLOSES else code) + + tasks = [asyncio.ensure_future(inbound()), asyncio.ensure_future(outbound())] + try: + done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + if ended: + return (*ended[0], None) + failure = next((task.exception() for task in done if not task.cancelled() and task.exception()), None) + if failure is None: + return "gateway", 1001, None + with contextlib.suppress(RuntimeError, WebSocketDisconnect): + await client.close(UNREACHABLE) + with contextlib.suppress(websockets.WebSocketException, OSError): + await browser.close(UNREACHABLE) + return "gateway", UNREACHABLE, type(failure).__name__ + + +def _answered(answer: httpx.Response, fields: frozenset[str], base: str, key: str) -> Response: + """The browser's ``answer``, a JSON one with each of ``fields`` at the gateway, anything else as it came.""" + media_type = answer.headers.get("content-type") + if media_type and media_type.split(";")[0].strip() == "application/json": + try: + return JSONResponse(rewritten(answer.json(), fields, base, key), status_code=answer.status_code) + except ValueError: + pass + return Response(answer.content, status_code=answer.status_code, media_type=media_type) + + +def _method_path(card: dict, uri: str, name: str) -> str | None: + params = _params(card, uri) + return next((path for method, _, _, _, path in operations(params) if method == name), None) if params else None + + +def _params(card: dict, uri: str) -> dict: + for extension in (card.get("capabilities") or {}).get("extensions") or []: + if isinstance(extension, dict) and extension.get("uri") == uri and isinstance(extension.get("params"), dict): + return extension["params"] + return {} + + +def _websocket_form(url: str) -> str: + return "wss://" + url.removeprefix("https://") if url.startswith("https://") else "ws://" + url.removeprefix("http://") + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _error(status: int, message: str) -> Response: + return JSONResponse({"error": message}, status_code=status) diff --git a/src/agent_env/env/gateway/constants.py b/src/agent_env/env/gateway/constants.py index 30bd0003..20f27957 100644 --- a/src/agent_env/env/gateway/constants.py +++ b/src/agent_env/env/gateway/constants.py @@ -80,6 +80,8 @@ def data_plane_load_timeout_s(size_bytes: int | None) -> int: EXT_CHANGES_URI = "urn:agentenv:changes/v1" EXT_NOTIFICATIONS_URI = "urn:agentenv:notifications/v1" EXT_WEBSITE_URI = "urn:agentenv:website/v1" +EXT_BROWSER_URI = "urn:agentenv:browser/v1" +EXT_COMPUTER_APP_URI = "urn:agentenv:computer-app/v1" WEBSITE_READY_OP = "ready" # Rows per GET /changes page: the default, and the most a client may ask for. diff --git a/src/agent_env/env/gateway/gateway.py b/src/agent_env/env/gateway/gateway.py index b7dd2209..623948a9 100644 --- a/src/agent_env/env/gateway/gateway.py +++ b/src/agent_env/env/gateway/gateway.py @@ -32,6 +32,7 @@ from pydantic import create_model from starlette.requests import Request from starlette.responses import Response, StreamingResponse +from starlette.routing import WebSocketRoute from anyio import ClosedResourceError if TYPE_CHECKING: @@ -67,6 +68,7 @@ GatewayMode, tool_name_clashes_text, ) +from .browser_proxy import PING_INTERVAL_S, BrowserRoutes, TokenRedaction from .caller import Caller from .changes import ChangeAttribution from .child_transport import child_http_client @@ -80,6 +82,11 @@ # RFC 9110's safe methods: a request using one asks for no change, so the proxy measures no write window for it. _SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "TRACE"}) +# A CDP address at the gateway: the website browser behind the /svc child , discovery and its WebSocket. +BROWSER_ROUTE = "/browser/{key}/cdp/{token}/{path:path}" +# The loggers that write a request's path or a forwarded URL: uvicorn's, for HTTP requests and WebSocket handshakes, +# and httpx's, for each request the gateway forwards. +_TOKEN_LOGGERS = ("uvicorn.access", "uvicorn.error", "httpx") _STEP_API = {"source": "step_api"} @@ -127,6 +134,12 @@ def __init__( self.website_urls = website_urls or {} self.rest_proxy_urls = rest_proxy_urls or {} self._site_reserved: dict[str, frozenset[str]] = {} + self._browsers = BrowserRoutes(self.rest_proxy_urls, self.REST_PROXY_TIMEOUT, self._log_event, + self._attribution, self._request_role) + for name in _TOKEN_LOGGERS: # each names a CDP address's path, its token in it, at INFO + log = logging.getLogger(name) + if not any(isinstance(f, TokenRedaction) for f in log.filters): + log.addFilter(TokenRedaction()) self.gateway_mode = gateway_mode self._service_db_url = service_db_url self._db_conn: psycopg2.extensions.connection | None = None @@ -188,6 +201,10 @@ async def list_website_urls() -> list[str]: async def proxy_to_service(request: Request) -> Response: return await self._proxy_rest_request(request) + @self._mcp.custom_route(BROWSER_ROUTE, methods=["GET", "HEAD"]) + async def browser_discovery(request: Request) -> Response: + return await self._browsers.discovery(request) + if self.website_urls: @self._mcp.custom_route( "/website/{service_name}/{path:path}", @@ -396,6 +413,8 @@ def run(self) -> None: host=self._mcp.settings.host, port=self._mcp.settings.port, log_level=self._mcp.settings.log_level.lower(), + ws_ping_interval=PING_INTERVAL_S, + ws_ping_timeout=PING_INTERVAL_S, ) server = uvicorn.Server(config) if self._resolve_i6pn_host(): @@ -418,6 +437,8 @@ def _asgi_app(self): lifespan wrapped around the MCP session manager's.""" inner = self._mcp.streamable_http_app() inner.router.lifespan_context = self._process_lifespan(inner.router.lifespan_context) + if self.rest_proxy_urls: + inner.router.routes.append(WebSocketRoute(BROWSER_ROUTE, self._browsers.relay)) async def app(scope, receive, send): if scope["type"] != "http": @@ -602,6 +623,8 @@ async def _proxy_rest_request(self, request: Request, url_map: dict[str, str] | return self._json_response({"error": f"Unknown service: {service_name}"}, 404) role = self._request_role(request) mcp_server = url_map is self.rest_proxy_urls and service_name.startswith("mcp-") + if url_map is self.rest_proxy_urls and (cdp := await self._browsers.cdp_answer(request, base_url, role)) is not None: + return cdp if request.method in _SAFE_METHODS: return await forward(request, base_url, role=role, timeout=self.REST_PROXY_TIMEOUT, mcp_server=mcp_server) changelog_before = self._query_changelog_id() diff --git a/src/agent_env/env/gateway/requirements.txt b/src/agent_env/env/gateway/requirements.txt index 89a76f21..26132898 100644 --- a/src/agent_env/env/gateway/requirements.txt +++ b/src/agent_env/env/gateway/requirements.txt @@ -3,3 +3,4 @@ httpx mcp>=1.25,<2 pydantic>=2.0 tzdata +websockets>=13 diff --git a/tst/integration/computer/browser_computer_test.py b/tst/integration/computer/browser_computer_test.py index d0604d56..5c095b76 100644 --- a/tst/integration/computer/browser_computer_test.py +++ b/tst/integration/computer/browser_computer_test.py @@ -18,8 +18,6 @@ import platform import re import shlex -import shutil -import subprocess import time import urllib.parse import uuid @@ -54,8 +52,6 @@ ARCH = {"x86_64": "amd64", "amd64": "amd64", "aarch64": "arm64", "arm64": "arm64"}[platform.machine().lower()] #: How long an effect may take to reach the page, or the computer to be ready again; how often it's asked meanwhile. WAIT_S, POLL_S = 120.0, 0.5 -#: Playwright and Puppeteer, which connect to a cdp address from Node. -CDP_CLIENTS = Path(__file__).parent / "cdp_clients" #: A WebSocket handshake for the browser's relay sent straight to its app server's socket, as code in the computer's #: session can, with the token it's given; prints the status line it gets. #: The lines of the computer's logs that hold either of two tokens, each shown masked, then one path the driver @@ -72,22 +68,6 @@ """ -@pytest.fixture(scope="module") -def cdp_clients(tmp_path_factory) -> Path: - """The CDP clients installed in a folder of their own, or the test skipped without Node 20 or later and npm.""" - node, npm = shutil.which("node"), shutil.which("npm") - version = subprocess.run([node, "--version"], capture_output=True, text=True).stdout if node else "" - if not npm or not (match := re.match(r"v(\d+)", version)) or int(match.group(1)) < 20: - pytest.skip("the CDP clients need Node 20 or later and npm") - folder = tmp_path_factory.mktemp("cdp-clients") - for name in ("package.json", "connect.mjs"): - shutil.copy(CDP_CLIENTS / name, folder / name) - installed = subprocess.run([npm, "install", "--no-audit", "--no-fund", "--loglevel=error"], cwd=folder, - capture_output=True, text=True, timeout=600) - assert installed.returncode == 0, installed.stderr - return folder - - # A page whose title says, as JSON, what reached it: clicks and double clicks on its red box, its input's value, how far # it has scrolled and the last keys pressed. PAGE = """loading