diff --git a/packages/agentenv-protocol/README.md b/packages/agentenv-protocol/README.md index 560821c68..cd7e676d1 100644 --- a/packages/agentenv-protocol/README.md +++ b/packages/agentenv-protocol/README.md @@ -907,6 +907,31 @@ are the profile's template. The base class adds: - `files/choose {paths}`, files in `~/Uploads` for the file chooser open on it, which headless, and so alone, it intercepts; - `page/state`, the active tab's id, URL and title and the dialog or file chooser open on it. + - `cookies/get {urls?}`, `cookies/set {cookies}` and `cookies/delete {name?, domain?, path?}`, on its default + context, each cookie as Playwright's `storageState` has it, CDP's `Network.Cookie` fields (`name`, `value`, + `domain`, `path`, `expires`, `httpOnly`, `secure`, `sameSite`), so a storage state exported from Playwright loads + as is, and a partitioned cookie's `partitionKey` (`{topLevelSite, hasCrossSiteAncestor}`), without which it would + be another cookie. `get` returns the cookies a request to any of `urls` would send, as Playwright picks them, else + every one; `set` takes a `url`, or a `domain` and a `path` (`/` by default), and may carry `max_age` in place of + `expires`: seconds from when it's set, on the computer's clock, which the browser's cookies are judged by. It sets + them all at once, so a cookie the browser refuses leaves none set. `delete` removes every cookie with all the + fields it's given, every cookie with none, partitioned or not, and says how many. A cookie refused names its + fields, never its value; + - `local_storage/get {origins?}`, `local_storage/set {origins}` and `local_storage/clear {origins?}`, each origin + as `storageState` has it, `{origin, localStorage: [{name, value}]}`. CDP reaches an origin's local storage only + through a page of it, so each goes through a hidden page agentenv sends to the origin and serves itself, past any + service worker of the origin: the site gets no request and runs no script. The origins answered are as the browser + names them (`https://xn--bcher-kva.de` for `https://bücher.de`). With no `origins`, every origin a page or a + same-site frame loaded since its server started, and those agentenv set. A cross-site iframe's local storage is + partitioned under the site that embeds it, and isn't read, as Playwright's `storageState` doesn't read it either. + Agents' CDP connections see each hidden page come and go as a target of type `other`; + - `downloads/list` and `downloads/get {guid, write_namespace?}`, a download and, completed, its file: uploaded + under the `write_namespace` grant when there is one, else inline, as `data/get` returns files. Inline, the whole + file is read into memory and sent base64 in one answer, so a large one belongs under a grant. + + The record's entries are the models in `agentenv_protocol.website_browser` (`OpenTab`, with the tab that opened + it as `opener`; `ClosedTab`; `Download`; `Dialog`; `LaunchEntry`; `BrowserRecord`), every field present and null + until it's known. - `cdp {user}`, `user`'s CDP address once the browser is ready, `http_url` to discover from and `ws_url` to connect to, both at the address the call reached it at with a token agentenv mints in the path: the same one each time it's asked for. Through the driver they're under diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser.py deleted file mode 100644 index 8b549e628..000000000 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser.py +++ /dev/null @@ -1,984 +0,0 @@ -"""WebsiteBrowserEnvironment: a website browser as a computer app, run inside a computer or alone. A subclass says how -its build installs and starts; this class adds what agentenv owns, the same in both: its flags after the subclass's -command, the unattended-use policies, a CDP connection over the browser's debugging pipe, the settings every browser -takes, a fresh cache and the deploy's prompt answers at every launch, another window for a launch while it runs, its -skill and ``browser/v1`` on its card. It runs alone, headless, on a computer with no display, a browser computer's, -whose screen is its page; which site proxy it goes through comes from a variable agentenv sets, never from its -settings.""" -from __future__ import annotations - -import asyncio -import contextlib -import hashlib -import itertools -import json -import logging -import os -import shutil -from collections.abc import AsyncIterator -from datetime import datetime, timezone -from pathlib import Path -from typing import Any, Awaitable, Callable, ClassVar, Literal, Optional - -from pydantic import BaseModel, Field, field_validator -from starlette.requests import Request -from starlette.responses import JSONResponse, Response -from starlette.websockets import WebSocket - -from .agent_env_environment import add_websocket_route, agent_skill, extension -from .browser_cdp import ROUTE, Relay, Tokens, reached, websocket_form -from .computer import AppDeclaration, ComputerAppEnvironment, Launch, install_step, ready_check -from .computer import app as computer_app -from .computer.app import SOCKET -from .computer.errors import DeviceUnavailable, InvalidArgument - -logger = logging.getLogger(__name__) - -SITE_PROXY_ENV = "AGENTENV_SITE_PROXY" -BROWSER = "urn:agentenv:browser/v1" -#: The flags agentenv sets, by prefix, which a subclass's command may not carry: among them every way to open a debugging -#: port, since the browser is reached only over its pipe. -OWNED_FLAGS = ("--user-data-dir", "--remote-debugging-", "--remote-allow-origins", "--proxy-", "--no-proxy-server", - "--headless", "--window-size", "--lang", "--force-webrtc-ip-handling-policy", "--disk-cache-dir", - "--disable-dev-shm-usage") -#: Below this much /dev/shm, as Docker's 64 MiB default, the browser keeps its shared memory in files instead. -SHM_MIN_BYTES = 1 << 30 -#: Chromium's content setting that denies a permission without a prompt. -BLOCK = 2 -#: Who closed a tab: agentenv over its own connection, or someone it can't tell; or how its browser ended, on its own -#: or stopped by its server. -AGENTENV, UNKNOWN, EXIT, STOP = "agentenv", "unknown", "exit", "stop" -#: How long the active tab's check waits on each page. -ACTIVE_CHECK_TIMEOUT_S = 2.0 -#: How often a call that waits for the browser to be ready checks it. -READY_POLL_S = 0.5 -#: How long a tab opened headless may take to be attached and held to its window's size. -PIN_TIMEOUT_S = 10.0 -#: How much of a download each read hashes. -HASH_CHUNK_BYTES = 1 << 20 - -#: Who answered a dialog when agentenv didn't. -AGENT = "agent" -#: The targets agentenv's connection follows. -PAGES = [{"type": "page"}] -#: The CDP input a computer's page driver may send to the active tab, and how long its screenshot may take. -PAGE_INPUT = frozenset({"Input.dispatchMouseEvent", "Input.dispatchKeyEvent", "Input.insertText"}) -CAPTURE_TIMEOUT_S, UNNAVIGATED_CAPTURE_TIMEOUT_S = 3.0, 1.0 -#: Where files a page's file chooser may be given come from, under the session user's home. -UPLOADS = "Uploads" -#: Alone, loopback goes through the proxy too. In a computer, the apps on loopback beside it stay reachable; ``<-loopback>`` -#: removes Chromium's own rule, so ``*.localhost`` names still don't. -ALONE_BYPASS = "<-loopback>" -COMPUTER_BYPASS = "<-loopback>;127.0.0.1;[::1];localhost" -#: Managed policies for unattended use: no sign-in, sync, password manager, autofill, metrics, promotions, translation, -#: background mode, component updates, default-browser check, download prompt or warning bar for its flags. -POLICIES: dict[str, Any] = { - "BrowserSignin": 0, - "SyncDisabled": True, - "PasswordManagerEnabled": False, - "AutofillSettings": [{"url_pattern": "*", "blocked_types": ["all"]}], - "MetricsReportingEnabled": False, - "PromotionalTabsEnabled": False, - "TranslateEnabled": False, - "BackgroundModeEnabled": False, - "ComponentUpdatesEnabled": False, - "DefaultBrowserSettingEnabled": False, - "PromptForDownloadLocation": False, - "CommandLineFlagSecurityWarningsEnabled": False, - "DownloadDirectory": "${user_home}/Downloads", -} -#: Where an agent's files to upload are, and where downloads land, under the session user's home. -UPLOADS = "Uploads" -DOWNLOADS = "Downloads" -SKILL = f"""# Website browser - -- Open sites by their real names, such as `https://slack.com`: the environment serves its sites under those names. -- Files to upload are in `~/{UPLOADS}`; downloads land in `~/{DOWNLOADS}`. -""" -#: The fds a Chromium on ``--remote-debugging-pipe`` reads its CDP messages from and writes its answers to. -PIPE_IN, PIPE_OUT = 3, 4 -#: The largest CDP message the pipe takes, such as a screenshot's. -MAX_MESSAGE_BYTES = 256 * 1024 * 1024 - - -class DevToolsError(RuntimeError): - """A CDP call the browser answered with an error.""" - - -class DevToolsPipe: - """A CDP connection over a Chromium's debugging pipe, each message JSON ended by a NUL. It connects anew at every - launch; a call with no browser connected, or whose browser exits before answering, raises ConnectionError. Each - event, a message with no ``id``, goes to every listener, and each closer hears when the browser closes the pipe.""" - - #: How long a call waits for the browser's answer when it names no wait of its own. - call_timeout_s: ClassVar[float] = 30.0 - - def __init__(self) -> None: - self._ids = itertools.count(1) - self._waiting: dict[int, asyncio.Future] = {} - self._writer: Optional[asyncio.WriteTransport] = None - self._reading: Optional[asyncio.Task] = None - self._listeners: list[Callable[[dict], None]] = [] - self._closers: list[Callable[[], None]] = [] - - def listen(self, listener: Callable[[dict], None]) -> None: - """Hand ``listener`` each event: its ``method``, ``params`` and, on a page's session, ``sessionId``.""" - self._listeners.append(listener) - - def on_close(self, closer: Callable[[], None]) -> None: - """Call ``closer`` when the browser closes its end of the pipe, not when this end closes it.""" - self._closers.append(closer) - - async def connect(self, to_browser: int, from_browser: int) -> None: - """Take over this end of a new pipe, ending any connection to a browser launched before.""" - self.close() - loop = asyncio.get_running_loop() - reader = asyncio.StreamReader(limit=MAX_MESSAGE_BYTES) - await loop.connect_read_pipe(lambda: asyncio.StreamReaderProtocol(reader), os.fdopen(from_browser, "rb", 0)) - self._writer, _ = await loop.connect_write_pipe(asyncio.Protocol, os.fdopen(to_browser, "wb", 0)) - self._reading = asyncio.ensure_future(self._read(reader)) - - async def call(self, method: str, params: Optional[dict] = None, *, session_id: Optional[str] = None, - timeout_s: Optional[float] = None) -> dict: - """``method``'s result, or DevToolsError with the browser's message.""" - answered = await self._answer(method, params, session_id, - self.call_timeout_s if timeout_s is None else timeout_s) - if "error" in answered: - raise DevToolsError(f"{method}: {answered['error'].get('message', answered['error'])}") - return answered.get("result", {}) - - async def request(self, method: str, params: Optional[dict] = None, *, session_id: Optional[str] = None) -> dict: - """The browser's whole answer to ``method``, its ``result`` or its ``error``, however long it takes: what a - relay passes on as it is.""" - return await self._answer(method, params, session_id, None) - - async def _answer(self, method: str, params: Optional[dict], session_id: Optional[str], - timeout_s: Optional[float]) -> dict: - if self._writer is None or self._writer.is_closing(): - raise ConnectionError("no browser is connected over its debugging pipe") - message_id = next(self._ids) - answer = asyncio.get_running_loop().create_future() - self._waiting[message_id] = answer - message = {"id": message_id, "method": method, "params": params or {}, - **({"sessionId": session_id} if session_id else {})} - self._writer.write(json.dumps(message).encode() + b"\0") - try: - return await asyncio.wait_for(answer, timeout_s) - finally: - self._waiting.pop(message_id, None) - - def close(self) -> None: - if self._reading is not None: - self._reading.cancel() - if self._writer is not None: - self._writer.close() - self._fail_waiting() - self._writer = self._reading = None - - async def _read(self, reader: asyncio.StreamReader) -> None: - try: - while True: - message = json.loads((await reader.readuntil(b"\0"))[:-1]) - if "id" not in message: - for listener in self._listeners: - _heard(listener, message) - continue - answer = self._waiting.get(message.get("id")) - if answer is not None and not answer.done(): - answer.set_result(message) - except (asyncio.IncompleteReadError, ConnectionError): - if self._writer is not None: - self._writer.close() - self._fail_waiting() - for closer in self._closers: - _heard(closer) - - def _fail_waiting(self) -> None: - for answer in self._waiting.values(): - if not answer.done(): - answer.set_exception(ConnectionError("the browser closed its debugging pipe")) - - -def _heard(callback: Callable, *args: Any) -> None: - try: - callback(*args) - except Exception: - logger.exception("a DevToolsPipe listener failed") - -class BrowserControl: - """agentenv's control of a browser over its pipe: at each launch it attaches to every page with flattened sessions - and keeps a record of the browser since this server started, its open tabs in opened order, those closed and who - closed them, its downloads with their size and sha256, and its JavaScript dialogs, which it answers when - ``dialogs`` says to: always leaving a page that asks before it's left, so no answer of agentenv's blocks a - navigation.""" - - def __init__(self, devtools: DevToolsPipe, dialogs: str, viewport: Optional[tuple[int, int]] = None) -> None: - self.devtools, self.dialogs, self.viewport = devtools, dialogs, viewport - #: When this record starts: a server that restarts starts another. - self.since = _now() - self.tabs: dict[str, dict] = {} - self.closed_tabs: list[dict] = [] - self.downloads: dict[str, dict] = {} - self.dialog_log: list[dict] = [] - self._sessions: dict[str, str] = {} - self._hidden: set[str] = set() - self._closing: set[str] = set() - self._open_dialogs: dict[str, dict] = {} - self._answering: set[str] = set() - #: The file chooser each page's session opened that waits, headless, and the tab agentenv last brought to the - #: front. - self._choosers: dict[str, dict] = {} - self._front: Optional[str] = None - #: Each tab's latest capture, which shows while a dialog keeps its page from painting; and, by session, what - #: input sent there waits on to learn a dialog opened. - self._shots: dict[str, str] = {} - self._dialog_opened: dict[str, asyncio.Event] = {} - self._pinned: dict[str, asyncio.Event] = {} - self._tasks: set[asyncio.Future] = set() - devtools.listen(self._event) - - async def attach(self) -> None: - """Follow a browser just launched: any tabs the one before left open closed by its server's stop, every page - this one has or opens attached, and its downloads reported, which go where its own policies say, as Chrome - downloads unattended: into ``~/Downloads``, a name downloaded again kept beside the first.""" - self.ended(STOP) - await self.devtools.call("Target.setDiscoverTargets", {"discover": True, "filter": PAGES}) - await self.devtools.call("Target.setAutoAttach", {"autoAttach": True, - "waitForDebuggerOnStart": self.viewport is not None, - "flatten": True, "filter": PAGES}) - await self.devtools.call("Browser.setDownloadBehavior", {"behavior": "default", "eventsEnabled": True}) - - def ended(self, how: str) -> None: - """The browser is gone, ``how`` it ended: its open tabs are closed by that, and its sessions with them.""" - now = _now() - self.closed_tabs += [{**tab, "closed_at": now, "closed_by": how} for tab in self.tabs.values()] - self.tabs.clear() - self._sessions.clear() - self._open_dialogs.clear() - self._choosers.clear() - self._shots.clear() - self._dialog_opened.clear() - self._pinned.clear() - - async def close_tab(self, target: str) -> None: - """Close the tab ``target``, which the record says agentenv closed.""" - self._closing.add(target) - await self.devtools.call("Target.closeTarget", {"targetId": target}) - - async def tab_list(self) -> list[dict]: - """The open tabs in opened order, each with its URL and title now and whether it's active.""" - await self._refresh() - active = await self.active_tab() - return [{"id": tab["id"], "url": tab["url"], "title": tab["title"], "active": tab["id"] == active} - for tab in self.tabs.values()] - - async def open_tab(self, url: Optional[str] = None) -> str: - """A new tab at ``url``, else blank, brought to the front; its id. Headless, it opens blank and goes to ``url`` - once it's held to the window's size, so its page loads at that size.""" - held = self.viewport is not None and url - start = "about:blank" if held else url or "about:blank" - target = (await self.devtools.call("Target.createTarget", {"url": start}))["targetId"] - if held: - await asyncio.wait_for(self._pinned.setdefault(target, asyncio.Event()).wait(), PIN_TIMEOUT_S) - session = next(s for s, t in self._sessions.items() if t == target) - await self.devtools.call("Page.navigate", {"url": url}, session_id=session) - await self._bring_to_front(target) - return target - - async def activate(self, target: str) -> None: - """The open tab ``target`` brought to the front, the active one.""" - self._open(target) - await self._bring_to_front(target) - - async def close(self, target: str) -> None: - """The open tab ``target`` closed, which the record says agentenv closed.""" - self._open(target) - await self.close_tab(target) - - async def navigate(self, url: str) -> None: - """The active tab opened at ``url``; InvalidArgument, with the browser's reason, when it can't.""" - answer = await self.devtools.call("Page.navigate", {"url": url}, session_id=await self.active_session()) - if error := answer.get("errorText"): - raise InvalidArgument(f"the browser couldn't open {url}: {error}") - - async def history(self, direction: str) -> None: - """The active tab's page reloaded, or the one before or after it in its history opened.""" - session = await self.active_session() - if direction == "reload": - await self.devtools.call("Page.reload", {}, session_id=session) - return - if direction not in ("back", "forward"): - raise InvalidArgument(f"direction is back, forward or reload, not {direction!r}") - history = await self.devtools.call("Page.getNavigationHistory", session_id=session) - index = history["currentIndex"] + (-1 if direction == "back" else 1) - if not 0 <= index < len(history["entries"]): - raise InvalidArgument(f"the active tab has no page to go {direction} to") - await self.devtools.call("Page.navigateToHistoryEntry", {"entryId": history["entries"][index]["id"]}, - session_id=session) - - async def answer_dialog(self, accept: bool, text: Optional[str] = None) -> None: - """The dialog open on the active tab answered, as the agent's answer; InvalidArgument with none open.""" - active = await self.active_tab() - session = next((s for s, dialog in self._open_dialogs.items() if dialog["tab"] == active), None) - if session is None: - raise InvalidArgument("no dialog is open on the page") - await self.devtools.call("Page.handleJavaScriptDialog", { - "accept": accept, **({"promptText": text} if text is not None else {})}, session_id=session) - - async def choose_files(self, paths: list[str]) -> None: - """The file chooser open on the active tab given ``paths``, files in the session user's ``~/Uploads``; - InvalidArgument for one outside it or missing, for several where it takes one, or with no chooser open.""" - active = await self.active_tab() - session = next((s for s in self._choosers if self._sessions.get(s) == active), None) - if session is None: - raise InvalidArgument("no file chooser is open on the page") - uploads = (Path.home() / UPLOADS).resolve() - files = [] - for path in paths: - file = (uploads / path).resolve() - if not file.is_relative_to(uploads): - raise InvalidArgument(f"{path} isn't in ~/{UPLOADS}") - if not file.is_file(): - raise InvalidArgument(f"there's no file {path} in ~/{UPLOADS}") - files.append(str(file)) - chooser = self._choosers[session] - if len(files) > 1 and not chooser["multiple"]: - raise InvalidArgument("the file chooser takes one file") - await self.devtools.call("DOM.setFileInputFiles", {"files": files, "backendNodeId": chooser["node"]}, - session_id=session) - self._choosers.pop(session, None) - - async def state(self) -> dict: - """The active tab's id, URL and title now, and the dialog and file chooser open on it.""" - await self._refresh() - active = await self.active_tab() - if active is None: - raise DeviceUnavailable("the browser has no tab open") - return self._state_of(active) - - def _state_of(self, target: str) -> dict: - tab = self.tabs.get(target) or {"url": "", "title": ""} - state: dict[str, Any] = {"tab": target, "url": tab["url"], "title": tab["title"]} - if dialog := next((d for d in self._open_dialogs.values() if d["tab"] == target), None): - state["dialog"] = {"type": dialog["type"], "message": dialog["message"], - "default_text": dialog["default_text"]} - if chooser := next((c for s, c in self._choosers.items() if self._sessions.get(s) == target), None): - state["file_chooser"] = {"multiple": chooser["multiple"]} - return state - - def _open(self, target: str) -> None: - if target not in self.tabs: - raise InvalidArgument(f"no tab {target} is open; list the tabs for their ids") - - async def _bring_to_front(self, target: str) -> None: - await self.devtools.call("Target.activateTarget", {"targetId": target}) - if (session := next((s for s, t in self._sessions.items() if t == target), None)) is not None: - with contextlib.suppress(DevToolsError): - await self.devtools.call("Page.bringToFront", session_id=session) - self._front = target - - @contextlib.asynccontextmanager - async def hidden_page(self, url: str = "about:blank") -> AsyncIterator[str]: - """A page of agentenv's own for reading state, in no window and never in the record; its session's id. It's - closed after.""" - target = (await self.devtools.call("Target.createTarget", - {"url": url, "hidden": True, "background": True}))["targetId"] - self._hidden.add(target) - self.tabs.pop(target, None) - try: - yield (await self.devtools.call("Target.attachToTarget", {"targetId": target, "flatten": True}))["sessionId"] - finally: - with contextlib.suppress(ConnectionError, DevToolsError): - await self.devtools.call("Target.closeTarget", {"targetId": target}) - - async def active_tab(self) -> Optional[str]: - """The tab a dialog is open on, whose page it blocks from saying; else the visible tab whose page has focus, as - a hidden tab's page that had input still says it has; else the one agentenv last brought to the front, else the - first visible one: CDP has no event for a tab made active.""" - if dialogs := [dialog["tab"] for dialog in self._open_dialogs.values() if dialog["tab"] in self.tabs]: - return dialogs[-1] - visible = None - for session, target in list(self._sessions.items()): - if target not in self.tabs: - continue - try: - state = (await self.devtools.call( - "Runtime.evaluate", {"expression": "[document.visibilityState, document.hasFocus()]", - "returnByValue": True}, session_id=session, - timeout_s=ACTIVE_CHECK_TIMEOUT_S))["result"]["value"] - except (ConnectionError, DevToolsError, asyncio.TimeoutError, KeyError): - continue - if state[0] == "visible" and state[1]: - return target - if state[0] == "visible" and visible is None: - visible = target - return self._front if self._front in self.tabs else visible - - async def record(self) -> dict: - """The record ``since`` it started: open ``tabs``, each with its URL and title now and whether it's ``active``, - ``closed_tabs``, ``downloads`` and ``dialogs``.""" - await self._refresh() - active = await self.active_tab() - return {"since": self.since, "tabs": [{**tab, "active": tab["id"] == active} for tab in self.tabs.values()], - "closed_tabs": list(self.closed_tabs), "downloads": list(self.downloads.values()), - "dialogs": list(self.dialog_log)} - - async def active_session(self) -> str: - """The session of the active tab, else of the first open one; DeviceUnavailable with no tab open.""" - active = await self.active_tab() or next(iter(self.tabs), None) - session = next((session for session, target in self._sessions.items() if target == active), None) - if session is None: - raise DeviceUnavailable("the browser has no tab open") - return session - - async def capture(self) -> Optional[str]: - """The active tab's page as it shows now, a PNG, base64; None for a tab that paints nothing within - ``CAPTURE_TIMEOUT_S``, which a desktop browser shows blank, or sooner for a tab with no URL yet, such as a new - tab whose navigation the browser blocked, which may never paint.""" - return await self._capture(await self.active_session()) - - async def capture_page(self) -> dict: - """The active tab's capture, ``png``, else ``blank``, with its ``page`` state, of that tab whatever becomes - active meanwhile.""" - session = await self.active_session() - target = self._sessions[session] - png = await self._capture(session) - await self._refresh(target) - return {**({"png": png} if png is not None else {"blank": True}), "page": self._state_of(target)} - - async def _capture(self, session: str) -> Optional[str]: - """The tab's page as it shows now; with a dialog open on it, which keeps it from painting, the last it took.""" - target = self._sessions[session] - if session in self._open_dialogs: - return self._shots.get(target) - navigated = self.tabs.get(target, {}).get("url") - try: - shot = await self.devtools.call("Page.captureScreenshot", {"format": "png", "optimizeForSpeed": True}, - session_id=session, - timeout_s=CAPTURE_TIMEOUT_S if navigated else UNNAVIGATED_CAPTURE_TIMEOUT_S) - except asyncio.TimeoutError: - return self._shots.get(target) if session in self._open_dialogs else None - self._shots[target] = shot["data"] - return shot["data"] - - async def dispatch(self, events: list[dict]) -> None: - """Each of ``events``, a CDP input call, sent to the active tab in order, until one opens a dialog left to the - agent: the page answers that one only once the dialog is answered, so it finishes then, and the rest aren't - sent. A dialog answered for the agent is answered at once, and the rest go on. InvalidArgument for any other - call.""" - if refused := [event.get("method") for event in events if event.get("method") not in PAGE_INPUT]: - raise InvalidArgument(f"page input takes {', '.join(sorted(PAGE_INPUT))}, not {refused}") - session = await self.active_session() - opened = self._dialog_opened.setdefault(session, asyncio.Event()) - for event in events: - if self.dialogs == AGENT and session in self._open_dialogs: - return - opened.clear() - sent = asyncio.ensure_future(self.devtools.call(event["method"], event.get("params") or {}, - session_id=session)) - waiting = asyncio.ensure_future(opened.wait()) - try: - await asyncio.wait({sent, waiting}, return_when=asyncio.FIRST_COMPLETED) - finally: - waiting.cancel() - if not sent.done(): - if self.dialogs != AGENT: - await sent - continue - self._tasks.add(sent) - sent.add_done_callback(self._finished) - return - sent.result() - - async def _refresh(self, target: Optional[str] = None) -> None: - """The URL and title of the tab ``target``, or of every open tab, as the browser has them now: it sends no - event when a page's title changes.""" - try: - infos = (await self.devtools.call("Target.getTargets", {"filter": PAGES}, - timeout_s=ACTIVE_CHECK_TIMEOUT_S))["targetInfos"] - except (ConnectionError, DevToolsError, asyncio.TimeoutError, KeyError): - return - for info in infos: - if target in (None, info["targetId"]) and (tab := self.tabs.get(info["targetId"])) is not None: - tab.update(url=info.get("url", tab["url"]), title=info.get("title", tab["title"])) - - def _event(self, message: dict) -> None: - """An event on this connection's own sessions; one on a session an agent's CDP connection holds is the - agent's.""" - session = message.get("sessionId") - handler = getattr(self, "_on_" + message.get("method", "").replace(".", "_"), None) - if handler is not None and (session is None or session in self._sessions): - handler(message.get("params", {}), session) - - def _on_Target_targetCreated(self, params: dict, session: Optional[str]) -> None: - info = params["targetInfo"] - if info.get("type") == "page" and info["targetId"] not in self._hidden and info["targetId"] not in self.tabs: - self.tabs[info["targetId"]] = {"id": info["targetId"], "url": info.get("url", ""), - "title": info.get("title", ""), "opened_at": _now()} - - def _on_Target_targetInfoChanged(self, params: dict, session: Optional[str]) -> None: - info = params["targetInfo"] - if (tab := self.tabs.get(info["targetId"])) is not None: - tab.update(url=info.get("url", tab["url"]), title=info.get("title", tab["title"])) - - def _on_Target_targetDestroyed(self, params: dict, session: Optional[str]) -> None: - target = params["targetId"] - if (tab := self.tabs.pop(target, None)) is not None: - self.closed_tabs.append({**tab, "closed_at": _now(), - "closed_by": AGENTENV if target in self._closing else UNKNOWN}) - self._closing.discard(target) - self._hidden.discard(target) - - def _on_Target_attachedToTarget(self, params: dict, session: Optional[str]) -> None: - if params["targetInfo"].get("type") == "page": - self._sessions[params["sessionId"]] = params["targetInfo"]["targetId"] - self._spawn(self._let_run(params["sessionId"])) - - async def _let_run(self, session: str) -> None: - """A page it attached to, enabled and, headless, held to its window's size, then told to run: a page attached as - it opens, a popup say, or any new page headless, waits for that before it runs.""" - target = self._sessions.get(session) - try: - await self.devtools.call("Page.enable", session_id=session) - if self.viewport is not None: - width, height = self.viewport - await self.devtools.call("Emulation.setDeviceMetricsOverride", { - "width": width, "height": height, "deviceScaleFactor": 1, "mobile": False}, session_id=session) - await self.devtools.call("Page.setInterceptFileChooserDialog", {"enabled": True}, session_id=session) - finally: - try: - await self.devtools.call("Runtime.runIfWaitingForDebugger", session_id=session) - finally: - if target is not None: - self._pinned.setdefault(target, asyncio.Event()).set() - - def _on_Target_detachedFromTarget(self, params: dict, session: Optional[str]) -> None: - if (target := self._sessions.pop(params["sessionId"], None)) is not None: - self._shots.pop(target, None) - self._pinned.pop(target, None) - self._choosers.pop(params["sessionId"], None) - self._dialog_opened.pop(params["sessionId"], None) - - def _on_Page_loadEventFired(self, params: dict, session: Optional[str]) -> None: - self._choosers.pop(session, None) - if (target := self._sessions.get(session)) is not None: - self._spawn(self._refresh(target)) - - def _on_Page_fileChooserOpened(self, params: dict, session: Optional[str]) -> None: - if session in self._sessions: - self._choosers[session] = {"node": params.get("backendNodeId"), - "multiple": params.get("mode") == "selectMultiple"} - - def _on_Page_javascriptDialogOpening(self, params: dict, session: Optional[str]) -> None: - dialog = {"tab": self._sessions.get(session), "type": params.get("type"), "message": params.get("message", ""), - "default_text": params.get("defaultPrompt", ""), "url": params.get("url", ""), "opened_at": _now(), - "answer": None, "by": None} - self.dialog_log.append(dialog) - self._open_dialogs[session] = dialog - if (opened := self._dialog_opened.get(session)) is not None: - opened.set() - if self.dialogs != AGENT: - self._answering.add(session) - leave = params.get("type") == "beforeunload" - self._spawn(self.devtools.call("Page.handleJavaScriptDialog", { - "accept": leave or self.dialogs == "accept", "promptText": params.get("defaultPrompt", "")}, - session_id=session)) - - def _on_Page_javascriptDialogClosed(self, params: dict, session: Optional[str]) -> None: - if (dialog := self._open_dialogs.pop(session, None)) is not None: - dialog.update(answer="accepted" if params.get("result") else "dismissed", answered_at=_now(), - by=AGENTENV if session in self._answering else AGENT, - **({"input": params["userInput"]} if params.get("userInput") else {})) - self._answering.discard(session) - - def _on_Browser_downloadWillBegin(self, params: dict, session: Optional[str]) -> None: - self.downloads[params["guid"]] = {"guid": params["guid"], "url": params.get("url", ""), - "filename": params.get("suggestedFilename", ""), "state": "inProgress", - "started_at": _now()} - - def _on_Browser_downloadProgress(self, params: dict, session: Optional[str]) -> None: - if (download := self.downloads.get(params["guid"])) is None: - return - download["state"] = params["state"] - if params["state"] == "completed": - if not params.get("filePath"): - download["ended_at"] = _now() - return - download["path"] = params["filePath"] - self._spawn(self._hashed(download, Path(params["filePath"]))) - elif params["state"] == "canceled": - download["ended_at"] = _now() - - async def _hashed(self, download: dict, path: Path) -> None: - size, sha256 = await asyncio.to_thread(_size_and_sha256, path) - download.update(size=size, sha256=sha256, ended_at=_now()) - - def _spawn(self, work: Awaitable) -> None: - task = asyncio.ensure_future(work) - self._tasks.add(task) - task.add_done_callback(self._finished) - - def _finished(self, task: asyncio.Future) -> None: - self._tasks.discard(task) - if not task.cancelled() and (error := task.exception()) is not None: - logger.warning(f"the browser's control connection: {type(error).__name__}: {error}") - - -def _size_and_sha256(path: Path) -> tuple[int, str]: - digest = hashlib.sha256() - with path.open("rb") as file: - while chunk := file.read(HASH_CHUNK_BYTES): - digest.update(chunk) - return path.stat().st_size, digest.hexdigest() - - -def _now() -> str: - return datetime.now(timezone.utc).isoformat() - - -class Window(BaseModel): - width: int = Field(1280, gt=0) - height: int = Field(800, gt=0) - - -class BrowserSettings(BaseModel): - """The settings every website browser takes, per deploy; a browser's own ``Settings`` adds to them.""" - - #: Opened in tabs at launch, the first active. - start_urls: list[str] = Field(default_factory=lambda: ["about:blank"]) - #: Its window when it runs alone; in a computer the screen sets it. - window: Window = Field(default_factory=Window) - locale: str = "en-US" - #: The IANA time zone its pages see. - time_zone: str = "UTC" - #: Whether a page asking for the location, or to show notifications, is denied without a prompt or asks. - geolocation: Literal["deny", "ask"] = "deny" - notifications: Literal["deny", "ask"] = "deny" - #: Who answers a page's JavaScript dialogs: the agent, or agentenv accepting or dismissing each. - dialogs: Literal["agent", "accept", "dismiss"] = "agent" - - @field_validator("start_urls") - @classmethod - def _no_flags(cls, urls: list[str]) -> list[str]: - if flags := [url for url in urls if url.startswith("-")]: - raise ValueError(f"start_urls are URLs, not flags: {flags}") - return urls - - -class WebsiteBrowserEnvironment(ComputerAppEnvironment): - """Subclass it as any computer app, with ``@computer_app`` and ``@linux``, its launch running the browser itself, - and set ``version``, its build, and ``policy_dir``, where that build reads managed policies. ``policies`` adds its - own to agentenv's, and ``profile`` is its profile folder under the session user's home, holding what its install - steps put there.""" - - Settings = BrowserSettings - version: ClassVar[str] = "" - policy_dir: ClassVar[Path] = Path("/etc/chromium/policies/managed") - policies: ClassVar[dict[str, Any]] = {} - profile: ClassVar[str] = ".config/chromium" - #: How long its ready check waits on the browser's answer over its pipe. - devtools_timeout_s: ClassVar[float] = 5.0 - agent_routes = (ROUTE,) - - def __init__(self) -> None: - #: Its CDP connection, over the pipe its server launches it with. - self.devtools = DevToolsPipe() - self._following: Optional[asyncio.Future] = None - super().__init__() - for folder in (UPLOADS, DOWNLOADS): - (Path.home() / folder).mkdir(exist_ok=True) - #: agentenv's control of it over that connection, and the record it keeps. - self.control = BrowserControl(self.devtools, self.settings.dialogs, - (self.settings.window.width, self.settings.window.height) if _alone() - else None) - #: The CDP addresses it has handed out, and the relay agents' connections to them go through. - self.cdp_tokens = Tokens(SOCKET.parent / "cdp-tokens.json") - self.relay = Relay(self.devtools, dialogs_answered=self.settings.dialogs != AGENT) - self.devtools.on_close(self._pipe_closed) - - @classmethod - def declaration(cls) -> AppDeclaration: - """What the class declares, a website browser; ValueError for a launch whose command carries a flag agentenv - owns.""" - for declared in cls._installs.values(): - if declared.launch is not None: - _refuse_owned(cls.__name__, declared.launch) - return super().declaration().model_copy(update={"browser": True}) - - def profile_dir(self) -> Path: - return Path.home() / self.profile - - def cache_dir(self) -> Path: - """Its HTTP cache, kept out of its profile, which holds only its data.""" - return Path.home() / ".cache" / Path(self.profile).name - - def agentenv_flags(self) -> list[str]: - """The flags agentenv owns. A browser alone goes out only through its site proxy, so it won't launch without - one; in a computer it goes through one when the computer has one, loopback aside.""" - settings, alone, proxy = self.settings, _alone(), os.environ.get(SITE_PROXY_ENV) - flags = [f"--user-data-dir={self.profile_dir()}", f"--disk-cache-dir={self.cache_dir()}", - "--remote-debugging-pipe", "--no-first-run", "--no-default-browser-check", f"--lang={settings.locale}"] - if (shm := _shm_bytes()) < SHM_MIN_BYTES: - logger.info(f"/dev/shm holds {shm // 2**20} MiB, under {SHM_MIN_BYTES // 2**20} MiB: the browser keeps its " - "shared memory in files") - flags.append("--disable-dev-shm-usage") - if proxy: - flags += [f"--proxy-server={proxy}", f"--proxy-bypass-list={ALONE_BYPASS if alone else COMPUTER_BYPASS}", - "--force-webrtc-ip-handling-policy=disable_non_proxied_udp"] - elif alone: - raise ValueError(f"a website browser alone goes out only through its site proxy, and {SITE_PROXY_ENV} " - "names none") - if alone: - flags.append(f"--window-size={settings.window.width},{settings.window.height}") - if alone: - flags.append("--headless=new") - return flags - - def _launch_for(self, launch: Launch) -> Launch: - """The subclass's launch with agentenv's flags and the start tabs after its command, and the time zone in its - environment; alone, with no window to match or wait for, its ``ready`` keeping its port and command, and - none when that leaves it nothing, so its ready checks say when it's ready.""" - _refuse_owned(type(self).__name__, launch) - update: dict[str, Any] = {"command": [*launch.command, *self.agentenv_flags(), *self._command_line_tabs()], - "env": {**(launch.env or {}), "TZ": self.settings.time_zone}} - if _alone(): - ready = launch.ready - waits = ready is not None and (ready.port is not None or bool(ready.command)) - update.update(window=None, ready=ready.model_copy(update={"window": False}) if waits else None) - return launch.model_copy(update=update) - - def _on_desktop(self) -> bool: - return not _alone() - - def _command_line_tabs(self) -> list[str]: - """The start tabs its command line opens: headless Chromium takes one, and the rest open over its pipe.""" - return self.settings.start_urls[:1] if _alone() else self.settings.start_urls - - async def _launch_fds(self) -> dict[int, int]: - """Before each launch, an empty cache and the deploy's prompt answers in its profile; and a new debugging pipe: - the browser's ends as its fds 3 and 4, this server's ends its CDP connection, which follows the browser and - then opens the start tabs its command line can't.""" - shutil.rmtree(self.cache_dir(), ignore_errors=True) - self._write_prompt_answers() - to_browser_read, to_browser_write = os.pipe() - from_browser_read, from_browser_write = os.pipe() - await self.devtools.connect(to_browser_write, from_browser_read) - self._following = asyncio.ensure_future(self._follow(self.settings.start_urls[len(self._command_line_tabs()):])) - self._following.add_done_callback(lambda done: done.cancelled() or done.exception()) - return {PIPE_IN: to_browser_read, PIPE_OUT: from_browser_write} - - def _pipe_closed(self) -> None: - """The browser closed its pipe: its tabs closed by its exit when it ended on its own, else by its server's - stop.""" - self.control.ended(EXIT if self._computer_app.exited() else STOP) - self.relay.ended() - - async def _follow(self, urls: list[str]) -> None: - """Its control attached, then each of ``urls`` in a tab behind the active one, in order.""" - await self.control.attach() - for url in urls: - await self.devtools.call("Target.createTarget", {"url": url, "background": True}) - - def _write_prompt_answers(self) -> None: - """Its profile's preferences, which it reads as it starts, denying location and notification prompts or - leaving them to ask, as the deploy's settings say.""" - preferences = self.profile_dir() / "Default" / "Preferences" - try: - current = json.loads(preferences.read_text(encoding="utf-8")) - except (OSError, ValueError): - current = {} - current = current if isinstance(current, dict) else {} - defaults = current.setdefault("profile", {}).setdefault("default_content_setting_values", {}) - for name in ("geolocation", "notifications"): - if getattr(self.settings, name) == "deny": - defaults[name] = BLOCK - else: - defaults.pop(name, None) - preferences.parent.mkdir(parents=True, exist_ok=True) - preferences.write_text(json.dumps(current), encoding="utf-8") - - def _relaunch_on_start(self) -> bool: - return True - - def _launch_running(self) -> Callable[[], Awaitable[None]]: - return self._open_window - - async def _open_window(self) -> None: - """Another window, which a launch of the browser while it runs opens over its pipe: a second copy would only - hand its window to the first and exit, taking its new pipe with it.""" - await self.devtools.call("Target.createTarget", {"url": "about:blank", "newWindow": True}) - - @install_step(scope="system") - def write_policies(self) -> None: - """agentenv's unattended-use policies, with the subclass's over them, where its build reads them.""" - self.policy_dir.mkdir(parents=True, exist_ok=True) - (self.policy_dir / "agentenv.json").write_text(json.dumps({**POLICIES, **self.policies}, indent=1) + "\n", - encoding="utf-8") - - @ready_check - async def devtools_answers(self) -> bool: - """It answers over its pipe, its control follows it and it has opened its start tabs.""" - try: - await self.devtools.call("Browser.getVersion", timeout_s=self.devtools_timeout_s) - except (ConnectionError, DevToolsError, asyncio.TimeoutError): - return False - following = self._following - return following is not None and following.done() and not following.cancelled() and \ - following.exception() is None - - @extension(BROWSER, op="record", method="GET", description=( - "A website browser agentenv runs: its build, the clock its pages see and, alone, its window. record returns " - "what agentenv's own connection to it saw: its open tabs, each with whether it's active, the tabs closed and " - "who closed them, its downloads with their size and sha256, its JavaScript dialogs and who answered each, and " - "each launch, stop and exit of it since the computer booted, with its cause. page/viewport, page/capture and " - "page/input are its active tab's page as a computer's page driver sees it: its size in pixels, which headless " - "is its window's, a PNG of it, and CDP input calls sent to it in order. navigate and history open a URL in " - "the active tab, or go back, forward or reload; tabs/list, tabs/open, tabs/activate and tabs/close manage " - "its tabs by their ids; dialog/answer answers the dialog open on the active tab; files/choose gives the file " - "chooser open on it files in ~/Uploads, headless; and page/state is the active tab's id, URL and title and " - "what is open on it. cdp returns a user's CDP address, http_url to discover from and ws_url to connect to, " - "with a token agentenv mints in its path: each connection to it gets its own browser-level session behind a " - "guard, and record lists them as cdp_connections, each with the commands the guard refused or changed.")) - async def browser_record(self) -> dict: - return {**await self.control.record(), "launches": self._computer_app.launches(), - "cdp_connections": list(self.relay.connections)} - - @extension(BROWSER, op="cdp") - async def browser_cdp(self, user: str, request: Request) -> dict: - """``user``'s CDP address at the address this call reached the browser at, once it's ready: the same one each - time it's asked for, as the server that minted it keeps it.""" - await self._until_ready() - address = self.cdp_tokens.mint(user, reached(request)) - return {"http_url": address.http_url, "ws_url": address.ws_url} - - def create_app(self) -> Any: - app = super().create_app() - for path in ("", "/", "/version", "/version/", "/list", "/list/"): - app.custom_route(f"{ROUTE}{{token}}/json{path}", methods=["GET"])(self._cdp_discovery) - for path in ("", "/{browser}"): - add_websocket_route(app, f"{ROUTE}{{token}}/devtools/browser{path}", self._cdp_connect) - return app - - async def _cdp_discovery(self, request: Request) -> Response: - """``json/version`` and ``json/list`` under a CDP address, answered from the browser, with its WebSocket at the - address the client reached: a browser on a pipe has no endpoints of its own. 403 for a token agentenv didn't - mint.""" - token = request.path_params["token"] - if self.cdp_tokens.address(token) is None: - return JSONResponse({"error": "agentenv didn't mint this CDP address"}, status_code=403) - try: - version = await self.devtools.call("Browser.getVersion") - if request.url.path.rstrip("/").endswith("/json/version"): - return JSONResponse({ - "Browser": version.get("product"), "Protocol-Version": version.get("protocolVersion"), - "User-Agent": version.get("userAgent"), "V8-Version": version.get("jsVersion"), - "WebKit-Version": version.get("revision"), - "webSocketDebuggerUrl": f"{websocket_form(reached(request))}{ROUTE}{token}/devtools/browser"}) - tabs = await self.control.tab_list() - except (ConnectionError, DevToolsError, asyncio.TimeoutError) as e: - return JSONResponse({"error": f"the browser isn't answering: {e}"}, status_code=503) - return JSONResponse([{"id": tab["id"], "type": "page", "title": tab["title"], "url": tab["url"], - "description": "", "devtoolsFrontendUrl": ""} for tab in tabs]) - - async def _cdp_connect(self, websocket: WebSocket) -> None: - address = self.cdp_tokens.address(websocket.path_params["token"]) - if address is None: - await websocket.close(code=1008) - return - await self.relay.serve(websocket, address.user) - - async def _until_ready(self) -> None: - """Back once it's ready, as ``computer-app/v1``'s ``ready`` says, or DeviceUnavailable after its ready - timeout.""" - timeout_s = self._computer_app.timeout_s - deadline = asyncio.get_running_loop().time() + timeout_s - while not (await self._computer_app.readiness()).ready: - if asyncio.get_running_loop().time() > deadline: - raise DeviceUnavailable(f"the browser isn't ready after {timeout_s:g} s") - await asyncio.sleep(READY_POLL_S) - - @extension(BROWSER, op="page/viewport", method="GET") - async def browser_page_viewport(self) -> dict: - window = self.settings.window - return {"width": window.width, "height": window.height} - - @extension(BROWSER, op="page/capture", method="GET") - async def browser_page_capture(self) -> dict: - return await self.control.capture_page() - - @extension(BROWSER, op="page/input") - async def browser_page_input(self, events: list[dict]) -> dict: - await self.control.dispatch(events) - return {} - - @extension(BROWSER, op="page/state", method="GET") - async def browser_page_state(self) -> dict: - return await self.control.state() - - @extension(BROWSER, op="navigate") - async def browser_navigate(self, url: str) -> dict: - await self.control.navigate(url) - return {} - - @extension(BROWSER, op="history") - async def browser_history(self, direction: str) -> dict: - await self.control.history(direction) - return {} - - @extension(BROWSER, op="tabs/list", method="GET") - async def browser_tabs_list(self) -> dict: - return {"tabs": await self.control.tab_list()} - - @extension(BROWSER, op="tabs/open") - async def browser_tabs_open(self, url: Optional[str] = None) -> dict: - return {"tab": await self.control.open_tab(url)} - - @extension(BROWSER, op="tabs/activate") - async def browser_tabs_activate(self, tab: str) -> dict: - await self.control.activate(tab) - return {} - - @extension(BROWSER, op="tabs/close") - async def browser_tabs_close(self, tab: str) -> dict: - await self.control.close(tab) - return {} - - @extension(BROWSER, op="dialog/answer") - async def browser_dialog_answer(self, accept: bool, text: Optional[str] = None) -> dict: - await self.control.answer_dialog(accept, text) - return {} - - @extension(BROWSER, op="files/choose") - async def browser_files_choose(self, paths: list[str]) -> dict: - await self.control.choose_files(paths) - return {} - - def extension_params(self, uri: str) -> dict: - if uri == BROWSER: - params: dict[str, Any] = {"chromium": self.version, "clock": "real"} - if _alone(): - params["window"] = self.settings.window.model_dump() - return params - return super().extension_params(uri) - - @agent_skill - def browser_skill(self) -> tuple[str, str]: - return "How to use this website browser.", SKILL - - -def _alone() -> bool: - """Whether it runs on a computer with no display, a browser computer's, alone and headless.""" - return computer_app.detect().display == "none" - - -def _shm_bytes(path: str = "/dev/shm") -> int: - try: - stats = os.statvfs(path) - except OSError: - return 0 - return stats.f_frsize * stats.f_blocks - - -def _refuse_owned(name: str, launch: Launch) -> None: - if owned := [arg for arg in launch.command if arg.startswith(OWNED_FLAGS)]: - raise ValueError(f"{name}'s launch sets {', '.join(owned)}, which agentenv sets for every website browser") diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/__init__.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/__init__.py new file mode 100644 index 000000000..904656ebc --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/__init__.py @@ -0,0 +1,23 @@ +"""WebsiteBrowserEnvironment: a website browser as a computer app, run inside a computer or alone. A subclass says how +its build installs and starts; this class adds what agentenv owns, the same in both: its flags after the subclass's +command, the unattended-use policies, a CDP connection over the browser's debugging pipe, the settings every browser +takes, a fresh cache and the deploy's prompt answers at every launch, another window for a launch while it runs, its +skill and ``browser/v1`` on its card. It runs alone, headless, on a computer with no display, a browser computer's, +whose screen is its page; which site proxy it goes through comes from a variable agentenv sets, never from its +settings.""" +from .control import (ACTIVE_CHECK_TIMEOUT_S, AGENT, AGENTENV, CAPTURE_TIMEOUT_S, DOWNLOADS, EXIT, HASH_CHUNK_BYTES, + PAGE_INPUT, PAGES, PIN_TIMEOUT_S, STOP, UNKNOWN, UNNAVIGATED_CAPTURE_TIMEOUT_S, UPLOADS, + BrowserControl) +from .devtools import MAX_MESSAGE_BYTES, PIPE_IN, PIPE_OUT, DevToolsError, DevToolsPipe +from .environment import (ALONE_BYPASS, BLOCK, BROWSER, COMPUTER_BYPASS, OWNED_FLAGS, POLICIES, READY_POLL_S, + SHM_MIN_BYTES, SITE_PROXY_ENV, SKILL, BrowserSettings, WebsiteBrowserEnvironment, Window) +from .shapes import (BrowserRecord, ClosedTab, Cookie, CookieToSet, Dialog, Download, LaunchEntry, OpenTab, + OriginStorage, PartitionKey, StorageItem, origin_of) + +__all__ = ["ACTIVE_CHECK_TIMEOUT_S", "AGENT", "AGENTENV", "ALONE_BYPASS", "BLOCK", "BROWSER", "CAPTURE_TIMEOUT_S", + "COMPUTER_BYPASS", "DOWNLOADS", "EXIT", "HASH_CHUNK_BYTES", "MAX_MESSAGE_BYTES", "OWNED_FLAGS", "PAGE_INPUT", + "PAGES", "PIN_TIMEOUT_S", "PIPE_IN", "PIPE_OUT", "POLICIES", "READY_POLL_S", "SHM_MIN_BYTES", + "SITE_PROXY_ENV", "SKILL", "STOP", "UNKNOWN", "UNNAVIGATED_CAPTURE_TIMEOUT_S", "UPLOADS", "BrowserControl", + "BrowserSettings", "DevToolsError", "DevToolsPipe", "WebsiteBrowserEnvironment", "Window", + "BrowserRecord", "ClosedTab", "Cookie", "CookieToSet", "Dialog", "Download", "LaunchEntry", "OpenTab", + "OriginStorage", "PartitionKey", "StorageItem", "origin_of"] diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/control.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/control.py new file mode 100644 index 000000000..082da5a25 --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/control.py @@ -0,0 +1,649 @@ +"""agentenv's control of a browser over its pipe, and the record it keeps.""" +from __future__ import annotations + +import asyncio +import base64 +import contextlib +import hashlib +import logging +import time +from collections.abc import AsyncIterator +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Awaitable, Callable, Optional +from urllib.parse import urlsplit + +from ..computer.errors import DeviceUnavailable, InvalidArgument +from .devtools import DevToolsError, DevToolsPipe +from .shapes import (ClosedTab, Cookie, CookieToSet, Dialog, Download, OpenTab, OriginStorage, StorageItem, + origin_of) + +logger = logging.getLogger(__name__) + +#: Who closed a tab: agentenv over its own connection, or someone it can't tell; or how its browser ended, on its own +#: or stopped by its server. +AGENTENV, UNKNOWN, EXIT, STOP = "agentenv", "unknown", "exit", "stop" +#: How long the active tab's check waits on each page. +ACTIVE_CHECK_TIMEOUT_S = 2.0 +#: How long a tab opened headless may take to be attached and held to its window's size, which is also how long a page +#: of agentenv's own may take to reach an origin, and how often it's asked meanwhile. +PIN_TIMEOUT_S = 10.0 +ARRIVAL_POLL_S = 0.05 +#: How much of a download each read hashes. +HASH_CHUNK_BYTES = 1 << 20 +#: Who answered a dialog when agentenv didn't. +AGENT = "agent" +#: The targets agentenv's connection follows. +PAGES = [{"type": "page"}] +#: The CDP input a computer's page driver may send to the active tab, and how long its screenshot may take. +PAGE_INPUT = frozenset({"Input.dispatchMouseEvent", "Input.dispatchKeyEvent", "Input.insertText"}) +CAPTURE_TIMEOUT_S, UNNAVIGATED_CAPTURE_TIMEOUT_S = 3.0, 1.0 +#: Where an agent's files to upload are, and where downloads land, under the session user's home. +UPLOADS = "Uploads" +DOWNLOADS = "Downloads" +#: The document agentenv serves a page of its own that it sends to an origin, so the site gets no request. +SERVED_PAGE = base64.b64encode(b"").decode() +#: Marks the document a page of agentenv's own leaves, and reads the origin of the one it goes to once it's there, as +#: the browser names it. +LEAVING, ARRIVED = "window.__agentenvLeft = true", "window.__agentenvLeft ? '' : location.origin" +#: Hostnames a secure cookie is sent to over http, as browsers treat them. +LOCAL_HOSTS = ("localhost", "127.0.0.1", "[::1]") + + +class BrowserControl: + """agentenv's control of a browser over its pipe: at each launch it attaches to every page with flattened sessions + and keeps a record of the browser since this server started, its open tabs in opened order, those closed and who + closed them, its downloads with their size and sha256, and its JavaScript dialogs, which it answers when + ``dialogs`` says to: always leaving a page that asks before it's left, so no answer of agentenv's blocks a + navigation.""" + + def __init__(self, devtools: DevToolsPipe, dialogs: str, viewport: Optional[tuple[int, int]] = None) -> None: + self.devtools, self.dialogs, self.viewport = devtools, dialogs, viewport + #: When this record starts: a server that restarts starts another. + self.since = _now() + self.tabs: dict[str, dict] = {} + self.closed_tabs: list[dict] = [] + self.downloads: dict[str, dict] = {} + self.dialog_log: list[dict] = [] + self._sessions: dict[str, str] = {} + self._hidden: set[str] = set() + self._closing: set[str] = set() + self._open_dialogs: dict[str, dict] = {} + self._answering: set[str] = set() + #: The file chooser each page's session opened that waits, headless, and the tab agentenv last brought to the + #: front. + self._choosers: dict[str, dict] = {} + self._front: Optional[str] = None + #: Each tab's latest capture, which shows while a dialog keeps its page from painting; and, by session, what + #: input sent there waits on to learn a dialog opened. + self._shots: dict[str, str] = {} + self._dialog_opened: dict[str, asyncio.Event] = {} + self._pinned: dict[str, asyncio.Event] = {} + self._tasks: set[asyncio.Future] = set() + #: Every http and https origin a page or frame it follows loaded since this server started, and the sessions + #: of agentenv's own pages whose requests it answers itself. + self.origins: set[str] = set() + self._serving: set[str] = set() + devtools.listen(self._event) + + async def attach(self) -> None: + """Follow a browser just launched: any tabs the one before left open closed by its server's stop, every page + this one has or opens attached, and its downloads reported, which go where its own policies say, as Chrome + downloads unattended: into ``~/Downloads``, a name downloaded again kept beside the first.""" + self.ended(STOP) + await self.devtools.call("Target.setDiscoverTargets", {"discover": True, "filter": PAGES}) + await self.devtools.call("Target.setAutoAttach", {"autoAttach": True, + "waitForDebuggerOnStart": self.viewport is not None, + "flatten": True, "filter": PAGES}) + await self.devtools.call("Browser.setDownloadBehavior", {"behavior": "default", "eventsEnabled": True}) + + def ended(self, how: str) -> None: + """The browser is gone, ``how`` it ended: its open tabs are closed by that, and its sessions with them.""" + now = _now() + self.closed_tabs += [{**tab, "closed_at": now, "closed_by": how} for tab in self.tabs.values()] + self.tabs.clear() + self._sessions.clear() + self._open_dialogs.clear() + self._choosers.clear() + self._shots.clear() + self._dialog_opened.clear() + self._pinned.clear() + + async def close_tab(self, target: str) -> None: + """Close the tab ``target``, which the record says agentenv closed.""" + self._closing.add(target) + await self.devtools.call("Target.closeTarget", {"targetId": target}) + + async def tab_list(self) -> list[dict]: + """The open tabs in opened order, each with its URL and title now and whether it's active.""" + await self._refresh() + active = await self.active_tab() + return [OpenTab(**tab, active=tab["id"] == active).model_dump() for tab in self.tabs.values()] + + async def open_tab(self, url: Optional[str] = None) -> str: + """A new tab at ``url``, else blank, brought to the front; its id. Headless, it opens blank and goes to ``url`` + once it's held to the window's size, so its page loads at that size.""" + held = self.viewport is not None and url + start = "about:blank" if held else url or "about:blank" + target = (await self.devtools.call("Target.createTarget", {"url": start}))["targetId"] + if held: + await asyncio.wait_for(self._pinned.setdefault(target, asyncio.Event()).wait(), PIN_TIMEOUT_S) + session = next(s for s, t in self._sessions.items() if t == target) + await self.devtools.call("Page.navigate", {"url": url}, session_id=session) + await self._bring_to_front(target) + return target + + async def activate(self, target: str) -> None: + """The open tab ``target`` brought to the front, the active one.""" + self._open(target) + await self._bring_to_front(target) + + async def close(self, target: str) -> None: + """The open tab ``target`` closed, which the record says agentenv closed.""" + self._open(target) + await self.close_tab(target) + + async def navigate(self, url: str) -> None: + """The active tab opened at ``url``; InvalidArgument, with the browser's reason, when it can't.""" + answer = await self.devtools.call("Page.navigate", {"url": url}, session_id=await self.active_session()) + if error := answer.get("errorText"): + raise InvalidArgument(f"the browser couldn't open {url}: {error}") + + async def history(self, direction: str) -> None: + """The active tab's page reloaded, or the one before or after it in its history opened.""" + session = await self.active_session() + if direction == "reload": + await self.devtools.call("Page.reload", {}, session_id=session) + return + if direction not in ("back", "forward"): + raise InvalidArgument(f"direction is back, forward or reload, not {direction!r}") + history = await self.devtools.call("Page.getNavigationHistory", session_id=session) + index = history["currentIndex"] + (-1 if direction == "back" else 1) + if not 0 <= index < len(history["entries"]): + raise InvalidArgument(f"the active tab has no page to go {direction} to") + await self.devtools.call("Page.navigateToHistoryEntry", {"entryId": history["entries"][index]["id"]}, + session_id=session) + + async def answer_dialog(self, accept: bool, text: Optional[str] = None) -> None: + """The dialog open on the active tab answered, as the agent's answer; InvalidArgument with none open.""" + active = await self.active_tab() + session = next((s for s, dialog in self._open_dialogs.items() if dialog["tab"] == active), None) + if session is None: + raise InvalidArgument("no dialog is open on the page") + await self.devtools.call("Page.handleJavaScriptDialog", { + "accept": accept, **({"promptText": text} if text is not None else {})}, session_id=session) + + async def choose_files(self, paths: list[str]) -> None: + """The file chooser open on the active tab given ``paths``, files in the session user's ``~/Uploads``; + InvalidArgument for one outside it or missing, for several where it takes one, or with no chooser open.""" + active = await self.active_tab() + session = next((s for s in self._choosers if self._sessions.get(s) == active), None) + if session is None: + raise InvalidArgument("no file chooser is open on the page") + uploads = (Path.home() / UPLOADS).resolve() + files = [] + for path in paths: + file = (uploads / path).resolve() + if not file.is_relative_to(uploads): + raise InvalidArgument(f"{path} isn't in ~/{UPLOADS}") + if not file.is_file(): + raise InvalidArgument(f"there's no file {path} in ~/{UPLOADS}") + files.append(str(file)) + chooser = self._choosers[session] + if len(files) > 1 and not chooser["multiple"]: + raise InvalidArgument("the file chooser takes one file") + await self.devtools.call("DOM.setFileInputFiles", {"files": files, "backendNodeId": chooser["node"]}, + session_id=session) + self._choosers.pop(session, None) + + async def state(self) -> dict: + """The active tab's id, URL and title now, and the dialog and file chooser open on it.""" + await self._refresh() + active = await self.active_tab() + if active is None: + raise DeviceUnavailable("the browser has no tab open") + return self._state_of(active) + + def _state_of(self, target: str) -> dict: + tab = self.tabs.get(target) or {"url": "", "title": ""} + state: dict[str, Any] = {"tab": target, "url": tab["url"], "title": tab["title"]} + if dialog := next((d for d in self._open_dialogs.values() if d["tab"] == target), None): + state["dialog"] = {"type": dialog["type"], "message": dialog["message"], + "default_text": dialog["default_text"]} + if chooser := next((c for s, c in self._choosers.items() if self._sessions.get(s) == target), None): + state["file_chooser"] = {"multiple": chooser["multiple"]} + return state + + def _open(self, target: str) -> None: + if target not in self.tabs: + raise InvalidArgument(f"no tab {target} is open; list the tabs for their ids") + + async def _bring_to_front(self, target: str) -> None: + await self.devtools.call("Target.activateTarget", {"targetId": target}) + if (session := next((s for s, t in self._sessions.items() if t == target), None)) is not None: + with contextlib.suppress(DevToolsError): + await self.devtools.call("Page.bringToFront", session_id=session) + self._front = target + + @contextlib.asynccontextmanager + async def hidden_page(self, url: str = "about:blank") -> AsyncIterator[str]: + """A page of agentenv's own for reading state, in no window and never in the record; its session's id. It's + closed after.""" + target = (await self.devtools.call("Target.createTarget", + {"url": url, "hidden": True, "background": True}))["targetId"] + self._hidden.add(target) + self.tabs.pop(target, None) + try: + yield (await self.devtools.call("Target.attachToTarget", {"targetId": target, "flatten": True}))["sessionId"] + finally: + with contextlib.suppress(ConnectionError, DevToolsError): + await self.devtools.call("Target.closeTarget", {"targetId": target}) + + async def active_tab(self) -> Optional[str]: + """The tab a dialog is open on, whose page it blocks from saying; else the visible tab whose page has focus, as + a hidden tab's page that had input still says it has; else the one agentenv last brought to the front, else the + first visible one: CDP has no event for a tab made active.""" + if dialogs := [dialog["tab"] for dialog in self._open_dialogs.values() if dialog["tab"] in self.tabs]: + return dialogs[-1] + visible = None + for session, target in list(self._sessions.items()): + if target not in self.tabs: + continue + try: + state = (await self.devtools.call( + "Runtime.evaluate", {"expression": "[document.visibilityState, document.hasFocus()]", + "returnByValue": True}, session_id=session, + timeout_s=ACTIVE_CHECK_TIMEOUT_S))["result"]["value"] + except (ConnectionError, DevToolsError, asyncio.TimeoutError, KeyError): + continue + if state[0] == "visible" and state[1]: + return target + if state[0] == "visible" and visible is None: + visible = target + return self._front if self._front in self.tabs else visible + + async def record(self) -> dict: + """The record ``since`` it started: open ``tabs``, each with its URL and title now and whether it's ``active``, + ``closed_tabs``, ``downloads`` and ``dialogs``.""" + return {"since": self.since, "tabs": await self.tab_list(), + "closed_tabs": [ClosedTab(**tab).model_dump() for tab in self.closed_tabs], + "downloads": self.download_list(), "dialogs": [Dialog(**dialog).model_dump() for dialog in self.dialog_log]} + + async def cookies(self, urls: Optional[list[str]] = None) -> list[dict]: + """The default context's cookies, or those a request to any of ``urls`` would send, as Playwright picks them.""" + cookies = [Cookie(**{field: raw[field] for field in Cookie.model_fields if field in raw}) + for raw in (await self.devtools.call("Storage.getCookies"))["cookies"]] + return [cookie.model_dump(exclude_none=True) for cookie in cookies + if not urls or any(_sent(cookie, url) for url in urls)] + + async def set_cookies(self, cookies: list[CookieToSet]) -> None: + """``cookies`` set in the default context at once, a ``max_age`` turned into an expiry on this machine's clock, + which the browser's cookies are judged by; InvalidArgument when the browser refuses one, which sets none.""" + try: + await self.devtools.call("Storage.setCookies", {"cookies": [_cdp_cookie(cookie, time.time()) + for cookie in cookies]}) + except DevToolsError as e: + raise InvalidArgument(f"the browser refused the cookies, so it set none of them: {e}") + + async def delete_cookies(self, name: Optional[str] = None, domain: Optional[str] = None, + path: Optional[str] = None) -> int: + """Every cookie of the default context with all of ``name``, ``domain`` and ``path`` given, every cookie with + none given, deleted; how many.""" + gone = [cookie for cookie in await self.cookies() if (name is None or cookie["name"] == name) + and (domain is None or cookie["domain"] == domain) and (path is None or cookie["path"] == path)] + if gone: + async with self.hidden_page() as session: + for cookie in gone: + await self.devtools.call("Network.deleteCookies", { + "name": cookie["name"], "domain": cookie["domain"], "path": cookie["path"], + **({"partitionKey": cookie["partitionKey"]} if cookie.get("partitionKey") else {})}, + session_id=session) + return len(gone) + + async def local_storage(self, origins: Optional[list[str]] = None) -> list[dict]: + """The local storage of each of ``origins``, else of every origin it has seen, as ``storageState`` has it.""" + wanted = _origins(origins) if origins is not None else sorted(self.origins) + found = [] + async with self._at_origins() as visit: + for asked in wanted: + session, origin = await visit(asked) + entries = (await self.devtools.call("DOMStorage.getDOMStorageItems", {"storageId": _storage(origin)}, + session_id=session))["entries"] + found.append(OriginStorage(origin=origin, localStorage=[StorageItem(name=name, value=value) + for name, value in entries]).model_dump()) + return found + + async def set_local_storage(self, origins: list[OriginStorage]) -> None: + """Each origin's items set in its local storage, beside those it has.""" + async with self._at_origins() as visit: + for origin in origins: + session, named = await visit(origin.origin) + for item in origin.localStorage: + await self.devtools.call("DOMStorage.setDOMStorageItem", { + "storageId": _storage(named), "key": item.name, "value": item.value}, session_id=session) + + async def clear_local_storage(self, origins: Optional[list[str]] = None) -> list[str]: + """The local storage of each of ``origins``, else of every origin it has seen, cleared; those origins.""" + wanted, cleared = _origins(origins) if origins is not None else sorted(self.origins), [] + async with self._at_origins() as visit: + for asked in wanted: + session, origin = await visit(asked) + await self.devtools.call("DOMStorage.clear", {"storageId": _storage(origin)}, session_id=session) + cleared.append(origin) + return cleared + + def download_list(self) -> list[dict]: + return [Download(**download).model_dump() for download in self.downloads.values()] + + def download(self, guid: str) -> dict: + """The download ``guid``; InvalidArgument for one it hasn't seen.""" + if guid not in self.downloads: + raise InvalidArgument(f"no download {guid} has begun; list the downloads for their guids") + return Download(**self.downloads[guid]).model_dump() + + @contextlib.asynccontextmanager + async def _at_origins(self) -> AsyncIterator[Callable[[str], Awaitable[tuple[str, str]]]]: + """A page of agentenv's own, ``visit``, that goes to an origin and returns its session and the origin as the + browser names it, once it's that origin's page. agentenv serves it the document, and a service worker of the + origin is bypassed, so the site gets no request and runs no script: CDP reaches an origin's local storage only + through a page of that origin.""" + async with self.hidden_page() as session: + self._serving.add(session) + try: + await self.devtools.call("Network.enable", session_id=session) + await self.devtools.call("Network.setBypassServiceWorker", {"bypass": True}, session_id=session) + await self.devtools.call("Fetch.enable", {"patterns": [{"urlPattern": "*"}]}, session_id=session) + await self.devtools.call("DOMStorage.enable", session_id=session) + + async def visit(origin: str) -> tuple[str, str]: + await self.devtools.call("Runtime.evaluate", {"expression": LEAVING}, session_id=session) + answer = await self.devtools.call("Page.navigate", {"url": f"{origin}/"}, session_id=session) + if error := answer.get("errorText"): + raise InvalidArgument(f"the browser couldn't open a page of {origin}: {error}") + deadline = asyncio.get_running_loop().time() + PIN_TIMEOUT_S + while not (named := (await self.devtools.call("Runtime.evaluate", { + "expression": ARRIVED, "returnByValue": True}, session_id=session))["result"].get("value")): + if asyncio.get_running_loop().time() > deadline: + raise DeviceUnavailable(f"a page of {origin} didn't open within {PIN_TIMEOUT_S:g} s") + await asyncio.sleep(ARRIVAL_POLL_S) + self.origins.add(named) + return session, named + + yield visit + finally: + self._serving.discard(session) + + async def active_session(self) -> str: + """The session of the active tab, else of the first open one; DeviceUnavailable with no tab open.""" + active = await self.active_tab() or next(iter(self.tabs), None) + session = next((session for session, target in self._sessions.items() if target == active), None) + if session is None: + raise DeviceUnavailable("the browser has no tab open") + return session + + async def capture(self) -> Optional[str]: + """The active tab's page as it shows now, a PNG, base64; None for a tab that paints nothing within + ``CAPTURE_TIMEOUT_S``, which a desktop browser shows blank, or sooner for a tab with no URL yet, such as a new + tab whose navigation the browser blocked, which may never paint.""" + return await self._capture(await self.active_session()) + + async def capture_page(self) -> dict: + """The active tab's capture, ``png``, else ``blank``, with its ``page`` state, of that tab whatever becomes + active meanwhile.""" + session = await self.active_session() + target = self._sessions[session] + png = await self._capture(session) + await self._refresh(target) + return {**({"png": png} if png is not None else {"blank": True}), "page": self._state_of(target)} + + async def _capture(self, session: str) -> Optional[str]: + """The tab's page as it shows now; with a dialog open on it, which keeps it from painting, the last it took.""" + target = self._sessions[session] + if session in self._open_dialogs: + return self._shots.get(target) + navigated = self.tabs.get(target, {}).get("url") + try: + shot = await self.devtools.call("Page.captureScreenshot", {"format": "png", "optimizeForSpeed": True}, + session_id=session, + timeout_s=CAPTURE_TIMEOUT_S if navigated else UNNAVIGATED_CAPTURE_TIMEOUT_S) + except asyncio.TimeoutError: + return self._shots.get(target) if session in self._open_dialogs else None + self._shots[target] = shot["data"] + return shot["data"] + + async def dispatch(self, events: list[dict]) -> None: + """Each of ``events``, a CDP input call, sent to the active tab in order, until one opens a dialog left to the + agent: the page answers that one only once the dialog is answered, so it finishes then, and the rest aren't + sent. A dialog answered for the agent is answered at once, and the rest go on. InvalidArgument for any other + call.""" + if refused := [event.get("method") for event in events if event.get("method") not in PAGE_INPUT]: + raise InvalidArgument(f"page input takes {', '.join(sorted(PAGE_INPUT))}, not {refused}") + session = await self.active_session() + opened = self._dialog_opened.setdefault(session, asyncio.Event()) + for event in events: + if self.dialogs == AGENT and session in self._open_dialogs: + return + opened.clear() + sent = asyncio.ensure_future(self.devtools.call(event["method"], event.get("params") or {}, + session_id=session)) + waiting = asyncio.ensure_future(opened.wait()) + try: + await asyncio.wait({sent, waiting}, return_when=asyncio.FIRST_COMPLETED) + finally: + waiting.cancel() + if not sent.done(): + if self.dialogs != AGENT: + await sent + continue + self._tasks.add(sent) + sent.add_done_callback(self._finished) + return + sent.result() + + async def _refresh(self, target: Optional[str] = None) -> None: + """The URL and title of the tab ``target``, or of every open tab, as the browser has them now: it sends no + event when a page's title changes.""" + try: + infos = (await self.devtools.call("Target.getTargets", {"filter": PAGES}, + timeout_s=ACTIVE_CHECK_TIMEOUT_S))["targetInfos"] + except (ConnectionError, DevToolsError, asyncio.TimeoutError, KeyError): + return + for info in infos: + if target in (None, info["targetId"]) and (tab := self.tabs.get(info["targetId"])) is not None: + tab.update(url=info.get("url", tab["url"]), title=info.get("title", tab["title"])) + + def _event(self, message: dict) -> None: + """An event on this connection's own sessions; one on a session an agent's CDP connection holds is the + agent's.""" + session = message.get("sessionId") + handler = getattr(self, "_on_" + message.get("method", "").replace(".", "_"), None) + if handler is not None and (session is None or session in self._sessions or session in self._serving): + handler(message.get("params", {}), session) + + def _on_Target_targetCreated(self, params: dict, session: Optional[str]) -> None: + info = params["targetInfo"] + if info.get("type") == "page" and info["targetId"] not in self._hidden and info["targetId"] not in self.tabs: + self.tabs[info["targetId"]] = {"id": info["targetId"], "url": info.get("url", ""), + "title": info.get("title", ""), "opened_at": _now(), + "opener": info.get("openerId")} + self._saw(info.get("url", "")) + + def _on_Target_targetInfoChanged(self, params: dict, session: Optional[str]) -> None: + info = params["targetInfo"] + if (tab := self.tabs.get(info["targetId"])) is not None: + tab.update(url=info.get("url", tab["url"]), title=info.get("title", tab["title"])) + self._saw(tab["url"]) + + def _on_Page_frameNavigated(self, params: dict, session: Optional[str]) -> None: + if session in self._sessions: + self._saw(params.get("frame", {}).get("url", "")) + + def _on_Fetch_requestPaused(self, params: dict, session: Optional[str]) -> None: + if session in self._serving: + self._spawn(self.devtools.call("Fetch.fulfillRequest", { + "requestId": params["requestId"], "responseCode": 200, + "responseHeaders": [{"name": "Content-Type", "value": "text/html"}], "body": SERVED_PAGE}, + session_id=session)) + + def _saw(self, url: str) -> None: + if (origin := origin_of(url)) is not None: + self.origins.add(origin) + + def _on_Target_targetDestroyed(self, params: dict, session: Optional[str]) -> None: + target = params["targetId"] + if (tab := self.tabs.pop(target, None)) is not None: + self.closed_tabs.append({**tab, "closed_at": _now(), + "closed_by": AGENTENV if target in self._closing else UNKNOWN}) + self._closing.discard(target) + self._hidden.discard(target) + + def _on_Target_attachedToTarget(self, params: dict, session: Optional[str]) -> None: + if params["targetInfo"].get("type") == "page": + self._sessions[params["sessionId"]] = params["targetInfo"]["targetId"] + self._spawn(self._let_run(params["sessionId"])) + + async def _let_run(self, session: str) -> None: + """A page it attached to, enabled and, headless, held to its window's size, then told to run: a page attached as + it opens, a popup say, or any new page headless, waits for that before it runs.""" + target = self._sessions.get(session) + try: + await self.devtools.call("Page.enable", session_id=session) + if self.viewport is not None: + width, height = self.viewport + await self.devtools.call("Emulation.setDeviceMetricsOverride", { + "width": width, "height": height, "deviceScaleFactor": 1, "mobile": False}, session_id=session) + await self.devtools.call("Page.setInterceptFileChooserDialog", {"enabled": True}, session_id=session) + finally: + try: + await self.devtools.call("Runtime.runIfWaitingForDebugger", session_id=session) + finally: + if target is not None: + self._pinned.setdefault(target, asyncio.Event()).set() + + def _on_Target_detachedFromTarget(self, params: dict, session: Optional[str]) -> None: + if (target := self._sessions.pop(params["sessionId"], None)) is not None: + self._shots.pop(target, None) + self._pinned.pop(target, None) + self._choosers.pop(params["sessionId"], None) + self._dialog_opened.pop(params["sessionId"], None) + + def _on_Page_loadEventFired(self, params: dict, session: Optional[str]) -> None: + self._choosers.pop(session, None) + if (target := self._sessions.get(session)) is not None: + self._spawn(self._refresh(target)) + + def _on_Page_fileChooserOpened(self, params: dict, session: Optional[str]) -> None: + if session in self._sessions: + self._choosers[session] = {"node": params.get("backendNodeId"), + "multiple": params.get("mode") == "selectMultiple"} + + def _on_Page_javascriptDialogOpening(self, params: dict, session: Optional[str]) -> None: + dialog = {"tab": self._sessions.get(session), "type": params.get("type"), "message": params.get("message", ""), + "default_text": params.get("defaultPrompt", ""), "url": params.get("url", ""), "opened_at": _now(), + "answer": None, "by": None} + self.dialog_log.append(dialog) + self._open_dialogs[session] = dialog + if (opened := self._dialog_opened.get(session)) is not None: + opened.set() + if self.dialogs != AGENT: + self._answering.add(session) + leave = params.get("type") == "beforeunload" + self._spawn(self.devtools.call("Page.handleJavaScriptDialog", { + "accept": leave or self.dialogs == "accept", "promptText": params.get("defaultPrompt", "")}, + session_id=session)) + + def _on_Page_javascriptDialogClosed(self, params: dict, session: Optional[str]) -> None: + if (dialog := self._open_dialogs.pop(session, None)) is not None: + dialog.update(answer="accepted" if params.get("result") else "dismissed", answered_at=_now(), + by=AGENTENV if session in self._answering else AGENT, + **({"input": params["userInput"]} if params.get("userInput") else {})) + self._answering.discard(session) + + def _on_Browser_downloadWillBegin(self, params: dict, session: Optional[str]) -> None: + self.downloads[params["guid"]] = {"guid": params["guid"], "url": params.get("url", ""), + "filename": params.get("suggestedFilename", ""), "state": "inProgress", + "started_at": _now()} + + def _on_Browser_downloadProgress(self, params: dict, session: Optional[str]) -> None: + if (download := self.downloads.get(params["guid"])) is None: + return + download["state"] = params["state"] + if params["state"] == "completed": + if not params.get("filePath"): + download["ended_at"] = _now() + return + download["path"] = params["filePath"] + self._spawn(self._hashed(download, Path(params["filePath"]))) + elif params["state"] == "canceled": + download["ended_at"] = _now() + + async def _hashed(self, download: dict, path: Path) -> None: + size, sha256 = await asyncio.to_thread(_size_and_sha256, path) + download.update(size=size, sha256=sha256, ended_at=_now()) + + def _spawn(self, work: Awaitable) -> None: + task = asyncio.ensure_future(work) + self._tasks.add(task) + task.add_done_callback(self._finished) + + def _finished(self, task: asyncio.Future) -> None: + self._tasks.discard(task) + if not task.cancelled() and (error := task.exception()) is not None: + logger.warning(f"the browser's control connection: {type(error).__name__}: {error}") + + +def _size_and_sha256(path: Path) -> tuple[int, str]: + digest = hashlib.sha256() + with path.open("rb") as file: + while chunk := file.read(HASH_CHUNK_BYTES): + digest.update(chunk) + return path.stat().st_size, digest.hexdigest() + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _origins(origins: list[str]) -> list[str]: + try: + return [origin_of(origin, strict=True) for origin in origins] + except ValueError as e: + raise InvalidArgument(str(e)) + + +def _storage(origin: str) -> dict: + return {"storageKey": f"{origin}/", "isLocalStorage": True} + + +def _cdp_cookie(cookie: CookieToSet, now: float) -> dict: + """``cookie`` as CDP's ``Storage.setCookies`` takes it.""" + param: dict[str, Any] = {"name": cookie.name, "value": cookie.value, "httpOnly": cookie.httpOnly, + "secure": cookie.secure} + if cookie.url: + param["url"] = cookie.url + if cookie.domain: + param["domain"] = cookie.domain + if cookie.path or cookie.domain: + param["path"] = cookie.path or "/" + if cookie.sameSite: + param["sameSite"] = cookie.sameSite + if cookie.partitionKey: + param["partitionKey"] = cookie.partitionKey.model_dump() + if cookie.max_age is not None: + param["expires"] = now + cookie.max_age + elif cookie.expires is not None and cookie.expires >= 0: + param["expires"] = cookie.expires + return param + + +def _sent(cookie: Cookie, url: str) -> bool: + """Whether a request to ``url`` sends ``cookie``, as Playwright's ``cookies(urls)`` decides: its domain, with or + without a leading dot, is the URL's host or a parent of it, its path a prefix of the URL's, and a secure cookie goes + over https or to a local host.""" + parts = urlsplit(url) + host = (parts.hostname or "").lower() + domain = cookie.domain.lower() if cookie.domain.startswith(".") else f".{cookie.domain.lower()}" + if not f".{host}".endswith(domain) or not (parts.path or "/").startswith(cookie.path): + return False + return not cookie.secure or parts.scheme == "https" or host in LOCAL_HOSTS or host == "::1" + diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/devtools.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/devtools.py new file mode 100644 index 000000000..986097b85 --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/devtools.py @@ -0,0 +1,121 @@ +"""A CDP connection over a Chromium's debugging pipe.""" +from __future__ import annotations + +import asyncio +import itertools +import json +import logging +import os +from typing import Any, Callable, ClassVar, Optional + +logger = logging.getLogger(__name__) + +#: The fds a Chromium on ``--remote-debugging-pipe`` reads its CDP messages from and writes its answers to. +PIPE_IN, PIPE_OUT = 3, 4 +#: The largest CDP message the pipe takes, such as a screenshot's. +MAX_MESSAGE_BYTES = 256 * 1024 * 1024 + + +class DevToolsError(RuntimeError): + """A CDP call the browser answered with an error.""" + + +class DevToolsPipe: + """A CDP connection over a Chromium's debugging pipe, each message JSON ended by a NUL. It connects anew at every + launch; a call with no browser connected, or whose browser exits before answering, raises ConnectionError. Each + event, a message with no ``id``, goes to every listener, and each closer hears when the browser closes the pipe.""" + + #: How long a call waits for the browser's answer when it names no wait of its own. + call_timeout_s: ClassVar[float] = 30.0 + + def __init__(self) -> None: + self._ids = itertools.count(1) + self._waiting: dict[int, asyncio.Future] = {} + self._writer: Optional[asyncio.WriteTransport] = None + self._reading: Optional[asyncio.Task] = None + self._listeners: list[Callable[[dict], None]] = [] + self._closers: list[Callable[[], None]] = [] + + def listen(self, listener: Callable[[dict], None]) -> None: + """Hand ``listener`` each event: its ``method``, ``params`` and, on a page's session, ``sessionId``.""" + self._listeners.append(listener) + + def on_close(self, closer: Callable[[], None]) -> None: + """Call ``closer`` when the browser closes its end of the pipe, not when this end closes it.""" + self._closers.append(closer) + + async def connect(self, to_browser: int, from_browser: int) -> None: + """Take over this end of a new pipe, ending any connection to a browser launched before.""" + self.close() + loop = asyncio.get_running_loop() + reader = asyncio.StreamReader(limit=MAX_MESSAGE_BYTES) + await loop.connect_read_pipe(lambda: asyncio.StreamReaderProtocol(reader), os.fdopen(from_browser, "rb", 0)) + self._writer, _ = await loop.connect_write_pipe(asyncio.Protocol, os.fdopen(to_browser, "wb", 0)) + self._reading = asyncio.ensure_future(self._read(reader)) + + async def call(self, method: str, params: Optional[dict] = None, *, session_id: Optional[str] = None, + timeout_s: Optional[float] = None) -> dict: + """``method``'s result, or DevToolsError with the browser's message.""" + answered = await self._answer(method, params, session_id, + self.call_timeout_s if timeout_s is None else timeout_s) + if "error" in answered: + raise DevToolsError(f"{method}: {answered['error'].get('message', answered['error'])}") + return answered.get("result", {}) + + async def request(self, method: str, params: Optional[dict] = None, *, session_id: Optional[str] = None) -> dict: + """The browser's whole answer to ``method``, its ``result`` or its ``error``, however long it takes: what a + relay passes on as it is.""" + return await self._answer(method, params, session_id, None) + + async def _answer(self, method: str, params: Optional[dict], session_id: Optional[str], + timeout_s: Optional[float]) -> dict: + if self._writer is None or self._writer.is_closing(): + raise ConnectionError("no browser is connected over its debugging pipe") + message_id = next(self._ids) + answer = asyncio.get_running_loop().create_future() + self._waiting[message_id] = answer + message = {"id": message_id, "method": method, "params": params or {}, + **({"sessionId": session_id} if session_id else {})} + self._writer.write(json.dumps(message).encode() + b"\0") + try: + return await asyncio.wait_for(answer, timeout_s) + finally: + self._waiting.pop(message_id, None) + + def close(self) -> None: + if self._reading is not None: + self._reading.cancel() + if self._writer is not None: + self._writer.close() + self._fail_waiting() + self._writer = self._reading = None + + async def _read(self, reader: asyncio.StreamReader) -> None: + try: + while True: + message = json.loads((await reader.readuntil(b"\0"))[:-1]) + if "id" not in message: + for listener in self._listeners: + _heard(listener, message) + continue + answer = self._waiting.get(message.get("id")) + if answer is not None and not answer.done(): + answer.set_result(message) + except (asyncio.IncompleteReadError, ConnectionError): + if self._writer is not None: + self._writer.close() + self._fail_waiting() + for closer in self._closers: + _heard(closer) + + def _fail_waiting(self) -> None: + for answer in self._waiting.values(): + if not answer.done(): + answer.set_exception(ConnectionError("the browser closed its debugging pipe")) + + +def _heard(callback: Callable, *args: Any) -> None: + try: + callback(*args) + except Exception: + logger.exception("a DevToolsPipe listener failed") diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/environment.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/environment.py new file mode 100644 index 000000000..bc4d8bf1c --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/environment.py @@ -0,0 +1,490 @@ +"""WebsiteBrowserEnvironment, its settings, and ``browser/v1`` on its card.""" +from __future__ import annotations + +import asyncio +import json +import logging +import os +import shutil +from pathlib import Path +from typing import Any, Awaitable, Callable, ClassVar, Literal, Optional + +from pydantic import BaseModel, Field, ValidationError, field_validator +from starlette.requests import Request +from starlette.responses import JSONResponse, Response +from starlette.websockets import WebSocket + +from ..agent_env_environment import add_websocket_route, agent_skill, extension +from ..browser_cdp import ROUTE, Relay, Tokens, reached, websocket_form +from ..computer import AppDeclaration, ComputerAppEnvironment, Launch, install_step, ready_check +from ..computer import app as computer_app +from ..computer.app import SOCKET +from ..computer.dataplane import file_parts +from ..computer.errors import DeviceUnavailable, InvalidArgument +from ..transfers import WriteNamespaceGrant +from .control import AGENT, DOWNLOADS, EXIT, STOP, UPLOADS, BrowserControl +from .devtools import PIPE_IN, PIPE_OUT, DevToolsError, DevToolsPipe +from .shapes import BrowserRecord, CookieToSet, OriginStorage + +logger = logging.getLogger(__name__) + +SITE_PROXY_ENV = "AGENTENV_SITE_PROXY" +BROWSER = "urn:agentenv:browser/v1" +#: The flags agentenv sets, by prefix, which a subclass's command may not carry: among them every way to open a debugging +#: port, since the browser is reached only over its pipe. +OWNED_FLAGS = ("--user-data-dir", "--remote-debugging-", "--remote-allow-origins", "--proxy-", "--no-proxy-server", + "--headless", "--window-size", "--lang", "--force-webrtc-ip-handling-policy", "--disk-cache-dir", + "--disable-dev-shm-usage") +#: Below this much /dev/shm, as Docker's 64 MiB default, the browser keeps its shared memory in files instead. +SHM_MIN_BYTES = 1 << 30 +#: Chromium's content setting that denies a permission without a prompt. +BLOCK = 2 +#: How often a call that waits for the browser to be ready checks it. +READY_POLL_S = 0.5 +#: Alone, loopback goes through the proxy too. In a computer, the apps on loopback beside it stay reachable; ``<-loopback>`` +#: removes Chromium's own rule, so ``*.localhost`` names still don't. +ALONE_BYPASS = "<-loopback>" +COMPUTER_BYPASS = "<-loopback>;127.0.0.1;[::1];localhost" +#: Managed policies for unattended use: no sign-in, sync, password manager, autofill, metrics, promotions, translation, +#: background mode, component updates, default-browser check, download prompt or warning bar for its flags. +POLICIES: dict[str, Any] = { + "BrowserSignin": 0, + "SyncDisabled": True, + "PasswordManagerEnabled": False, + "AutofillSettings": [{"url_pattern": "*", "blocked_types": ["all"]}], + "MetricsReportingEnabled": False, + "PromotionalTabsEnabled": False, + "TranslateEnabled": False, + "BackgroundModeEnabled": False, + "ComponentUpdatesEnabled": False, + "DefaultBrowserSettingEnabled": False, + "PromptForDownloadLocation": False, + "CommandLineFlagSecurityWarningsEnabled": False, + "DownloadDirectory": "${user_home}/Downloads", +} +SKILL = f"""# Website browser + +- Open sites by their real names, such as `https://slack.com`: the environment serves its sites under those names. +- Files to upload are in `~/{UPLOADS}`; downloads land in `~/{DOWNLOADS}`. +""" + + +class Window(BaseModel): + width: int = Field(1280, gt=0) + height: int = Field(800, gt=0) + + +class BrowserSettings(BaseModel): + """The settings every website browser takes, per deploy; a browser's own ``Settings`` adds to them.""" + + #: Opened in tabs at launch, the first active. + start_urls: list[str] = Field(default_factory=lambda: ["about:blank"]) + #: Its window when it runs alone; in a computer the screen sets it. + window: Window = Field(default_factory=Window) + locale: str = "en-US" + #: The IANA time zone its pages see. + time_zone: str = "UTC" + #: Whether a page asking for the location, or to show notifications, is denied without a prompt or asks. + geolocation: Literal["deny", "ask"] = "deny" + notifications: Literal["deny", "ask"] = "deny" + #: Who answers a page's JavaScript dialogs: the agent, or agentenv accepting or dismissing each. + dialogs: Literal["agent", "accept", "dismiss"] = "agent" + + @field_validator("start_urls") + @classmethod + def _no_flags(cls, urls: list[str]) -> list[str]: + if flags := [url for url in urls if url.startswith("-")]: + raise ValueError(f"start_urls are URLs, not flags: {flags}") + return urls + + +class WebsiteBrowserEnvironment(ComputerAppEnvironment): + """Subclass it as any computer app, with ``@computer_app`` and ``@linux``, its launch running the browser itself, + and set ``version``, its build, and ``policy_dir``, where that build reads managed policies. ``policies`` adds its + own to agentenv's, and ``profile`` is its profile folder under the session user's home, holding what its install + steps put there.""" + + Settings = BrowserSettings + version: ClassVar[str] = "" + policy_dir: ClassVar[Path] = Path("/etc/chromium/policies/managed") + policies: ClassVar[dict[str, Any]] = {} + profile: ClassVar[str] = ".config/chromium" + #: How long its ready check waits on the browser's answer over its pipe. + devtools_timeout_s: ClassVar[float] = 5.0 + agent_routes = (ROUTE,) + + def __init__(self) -> None: + #: Its CDP connection, over the pipe its server launches it with. + self.devtools = DevToolsPipe() + self._following: Optional[asyncio.Future] = None + super().__init__() + for folder in (UPLOADS, DOWNLOADS): + (Path.home() / folder).mkdir(exist_ok=True) + #: agentenv's control of it over that connection, and the record it keeps. + self.control = BrowserControl(self.devtools, self.settings.dialogs, + (self.settings.window.width, self.settings.window.height) if _alone() + else None) + #: The CDP addresses it has handed out, and the relay agents' connections to them go through. + self.cdp_tokens = Tokens(SOCKET.parent / "cdp-tokens.json") + self.relay = Relay(self.devtools, dialogs_answered=self.settings.dialogs != AGENT) + self.devtools.on_close(self._pipe_closed) + + @classmethod + def declaration(cls) -> AppDeclaration: + """What the class declares, a website browser; ValueError for a launch whose command carries a flag agentenv + owns.""" + for declared in cls._installs.values(): + if declared.launch is not None: + _refuse_owned(cls.__name__, declared.launch) + return super().declaration().model_copy(update={"browser": True}) + + def profile_dir(self) -> Path: + return Path.home() / self.profile + + def cache_dir(self) -> Path: + """Its HTTP cache, kept out of its profile, which holds only its data.""" + return Path.home() / ".cache" / Path(self.profile).name + + def agentenv_flags(self) -> list[str]: + """The flags agentenv owns. A browser alone goes out only through its site proxy, so it won't launch without + one; in a computer it goes through one when the computer has one, loopback aside.""" + settings, alone, proxy = self.settings, _alone(), os.environ.get(SITE_PROXY_ENV) + flags = [f"--user-data-dir={self.profile_dir()}", f"--disk-cache-dir={self.cache_dir()}", + "--remote-debugging-pipe", "--no-first-run", "--no-default-browser-check", f"--lang={settings.locale}"] + if (shm := _shm_bytes()) < SHM_MIN_BYTES: + logger.info(f"/dev/shm holds {shm // 2**20} MiB, under {SHM_MIN_BYTES // 2**20} MiB: the browser keeps its " + "shared memory in files") + flags.append("--disable-dev-shm-usage") + if proxy: + flags += [f"--proxy-server={proxy}", f"--proxy-bypass-list={ALONE_BYPASS if alone else COMPUTER_BYPASS}", + "--force-webrtc-ip-handling-policy=disable_non_proxied_udp"] + elif alone: + raise ValueError(f"a website browser alone goes out only through its site proxy, and {SITE_PROXY_ENV} " + "names none") + if alone: + flags.append(f"--window-size={settings.window.width},{settings.window.height}") + if alone: + flags.append("--headless=new") + return flags + + def _launch_for(self, launch: Launch) -> Launch: + """The subclass's launch with agentenv's flags and the start tabs after its command, and the time zone in its + environment; alone, with no window to match or wait for, its ``ready`` keeping its port and command, and + none when that leaves it nothing, so its ready checks say when it's ready.""" + _refuse_owned(type(self).__name__, launch) + update: dict[str, Any] = {"command": [*launch.command, *self.agentenv_flags(), *self._command_line_tabs()], + "env": {**(launch.env or {}), "TZ": self.settings.time_zone}} + if _alone(): + ready = launch.ready + waits = ready is not None and (ready.port is not None or bool(ready.command)) + update.update(window=None, ready=ready.model_copy(update={"window": False}) if waits else None) + return launch.model_copy(update=update) + + def _on_desktop(self) -> bool: + return not _alone() + + def _command_line_tabs(self) -> list[str]: + """The start tabs its command line opens: headless Chromium takes one, and the rest open over its pipe.""" + return self.settings.start_urls[:1] if _alone() else self.settings.start_urls + + async def _launch_fds(self) -> dict[int, int]: + """Before each launch, an empty cache and the deploy's prompt answers in its profile; and a new debugging pipe: + the browser's ends as its fds 3 and 4, this server's ends its CDP connection, which follows the browser and + then opens the start tabs its command line can't.""" + shutil.rmtree(self.cache_dir(), ignore_errors=True) + self._write_prompt_answers() + to_browser_read, to_browser_write = os.pipe() + from_browser_read, from_browser_write = os.pipe() + await self.devtools.connect(to_browser_write, from_browser_read) + self._following = asyncio.ensure_future(self._follow(self.settings.start_urls[len(self._command_line_tabs()):])) + self._following.add_done_callback(lambda done: done.cancelled() or done.exception()) + return {PIPE_IN: to_browser_read, PIPE_OUT: from_browser_write} + + def _pipe_closed(self) -> None: + """The browser closed its pipe: its tabs closed by its exit when it ended on its own, else by its server's + stop.""" + self.control.ended(EXIT if self._computer_app.exited() else STOP) + self.relay.ended() + + async def _follow(self, urls: list[str]) -> None: + """Its control attached, then each of ``urls`` in a tab behind the active one, in order.""" + await self.control.attach() + for url in urls: + await self.devtools.call("Target.createTarget", {"url": url, "background": True}) + + def _write_prompt_answers(self) -> None: + """Its profile's preferences, which it reads as it starts, denying location and notification prompts or + leaving them to ask, as the deploy's settings say.""" + preferences = self.profile_dir() / "Default" / "Preferences" + try: + current = json.loads(preferences.read_text(encoding="utf-8")) + except (OSError, ValueError): + current = {} + current = current if isinstance(current, dict) else {} + defaults = current.setdefault("profile", {}).setdefault("default_content_setting_values", {}) + for name in ("geolocation", "notifications"): + if getattr(self.settings, name) == "deny": + defaults[name] = BLOCK + else: + defaults.pop(name, None) + preferences.parent.mkdir(parents=True, exist_ok=True) + preferences.write_text(json.dumps(current), encoding="utf-8") + + def _relaunch_on_start(self) -> bool: + return True + + def _launch_running(self) -> Callable[[], Awaitable[None]]: + return self._open_window + + async def _open_window(self) -> None: + """Another window, which a launch of the browser while it runs opens over its pipe: a second copy would only + hand its window to the first and exit, taking its new pipe with it.""" + await self.devtools.call("Target.createTarget", {"url": "about:blank", "newWindow": True}) + + @install_step(scope="system") + def write_policies(self) -> None: + """agentenv's unattended-use policies, with the subclass's over them, where its build reads them.""" + self.policy_dir.mkdir(parents=True, exist_ok=True) + (self.policy_dir / "agentenv.json").write_text(json.dumps({**POLICIES, **self.policies}, indent=1) + "\n", + encoding="utf-8") + + @ready_check + async def devtools_answers(self) -> bool: + """It answers over its pipe, its control follows it and it has opened its start tabs.""" + try: + await self.devtools.call("Browser.getVersion", timeout_s=self.devtools_timeout_s) + except (ConnectionError, DevToolsError, asyncio.TimeoutError): + return False + following = self._following + return following is not None and following.done() and not following.cancelled() and \ + following.exception() is None + + @extension(BROWSER, op="record", method="GET", description=( + "A website browser agentenv runs: its build, the clock its pages see and, alone, its window. record returns " + "what agentenv's own connection to it saw: its open tabs, each with whether it's active, the tabs closed and " + "who closed them, its downloads with their size and sha256, its JavaScript dialogs and who answered each, and " + "each launch, stop and exit of it since the computer booted, with its cause. page/viewport, page/capture and " + "page/input are its active tab's page as a computer's page driver sees it: its size in pixels, which headless " + "is its window's, a PNG of it, and CDP input calls sent to it in order. navigate and history open a URL in " + "the active tab, or go back, forward or reload; tabs/list, tabs/open, tabs/activate and tabs/close manage " + "its tabs by their ids; dialog/answer answers the dialog open on the active tab; files/choose gives the file " + "chooser open on it files in ~/Uploads, headless; and page/state is the active tab's id, URL and title and " + "what is open on it. cookies/get, cookies/set and cookies/delete, and local_storage/get, local_storage/set and " + "local_storage/clear, read and change its default context's cookies and local storage as Playwright's " + "storageState has them, a cookie's max_age set as an expiry on the browser's clock; downloads/list and " + "downloads/get are its downloads, get with its file, uploaded under a write_namespace grant or inline. cdp " + "returns a user's CDP address, http_url to discover from and ws_url to connect to, " + "with a token agentenv mints in its path: each connection to it gets its own browser-level session behind a " + "guard, and record lists them as cdp_connections, each with the commands the guard refused or changed.")) + async def browser_record(self) -> dict: + return BrowserRecord(**await self.control.record(), launches=self._computer_app.launches(), + cdp_connections=list(self.relay.connections)).model_dump() + + @extension(BROWSER, op="cdp") + async def browser_cdp(self, user: str, request: Request) -> dict: + """``user``'s CDP address at the address this call reached the browser at, once it's ready: the same one each + time it's asked for, as the server that minted it keeps it.""" + await self._until_ready() + address = self.cdp_tokens.mint(user, reached(request)) + return {"http_url": address.http_url, "ws_url": address.ws_url} + + def create_app(self) -> Any: + app = super().create_app() + for path in ("", "/", "/version", "/version/", "/list", "/list/"): + app.custom_route(f"{ROUTE}{{token}}/json{path}", methods=["GET"])(self._cdp_discovery) + for path in ("", "/{browser}"): + add_websocket_route(app, f"{ROUTE}{{token}}/devtools/browser{path}", self._cdp_connect) + return app + + async def _cdp_discovery(self, request: Request) -> Response: + """``json/version`` and ``json/list`` under a CDP address, answered from the browser, with its WebSocket at the + address the client reached: a browser on a pipe has no endpoints of its own. 403 for a token agentenv didn't + mint.""" + token = request.path_params["token"] + if self.cdp_tokens.address(token) is None: + return JSONResponse({"error": "agentenv didn't mint this CDP address"}, status_code=403) + try: + version = await self.devtools.call("Browser.getVersion") + if request.url.path.rstrip("/").endswith("/json/version"): + return JSONResponse({ + "Browser": version.get("product"), "Protocol-Version": version.get("protocolVersion"), + "User-Agent": version.get("userAgent"), "V8-Version": version.get("jsVersion"), + "WebKit-Version": version.get("revision"), + "webSocketDebuggerUrl": f"{websocket_form(reached(request))}{ROUTE}{token}/devtools/browser"}) + tabs = await self.control.tab_list() + except (ConnectionError, DevToolsError, asyncio.TimeoutError) as e: + return JSONResponse({"error": f"the browser isn't answering: {e}"}, status_code=503) + return JSONResponse([{"id": tab["id"], "type": "page", "title": tab["title"], "url": tab["url"], + "description": "", "devtoolsFrontendUrl": ""} for tab in tabs]) + + async def _cdp_connect(self, websocket: WebSocket) -> None: + address = self.cdp_tokens.address(websocket.path_params["token"]) + if address is None: + await websocket.close(code=1008) + return + await self.relay.serve(websocket, address.user) + + async def _until_ready(self) -> None: + """Back once it's ready, as ``computer-app/v1``'s ``ready`` says, or DeviceUnavailable after its ready + timeout.""" + timeout_s = self._computer_app.timeout_s + deadline = asyncio.get_running_loop().time() + timeout_s + while not (await self._computer_app.readiness()).ready: + if asyncio.get_running_loop().time() > deadline: + raise DeviceUnavailable(f"the browser isn't ready after {timeout_s:g} s") + await asyncio.sleep(READY_POLL_S) + + @extension(BROWSER, op="page/viewport", method="GET") + async def browser_page_viewport(self) -> dict: + window = self.settings.window + return {"width": window.width, "height": window.height} + + @extension(BROWSER, op="page/capture", method="GET") + async def browser_page_capture(self) -> dict: + return await self.control.capture_page() + + @extension(BROWSER, op="page/input") + async def browser_page_input(self, events: list[dict]) -> dict: + await self.control.dispatch(events) + return {} + + @extension(BROWSER, op="page/state", method="GET") + async def browser_page_state(self) -> dict: + return await self.control.state() + + @extension(BROWSER, op="navigate") + async def browser_navigate(self, url: str) -> dict: + await self.control.navigate(url) + return {} + + @extension(BROWSER, op="history") + async def browser_history(self, direction: str) -> dict: + await self.control.history(direction) + return {} + + @extension(BROWSER, op="tabs/list", method="GET") + async def browser_tabs_list(self) -> dict: + return {"tabs": await self.control.tab_list()} + + @extension(BROWSER, op="tabs/open") + async def browser_tabs_open(self, url: Optional[str] = None) -> dict: + return {"tab": await self.control.open_tab(url)} + + @extension(BROWSER, op="tabs/activate") + async def browser_tabs_activate(self, tab: str) -> dict: + await self.control.activate(tab) + return {} + + @extension(BROWSER, op="tabs/close") + async def browser_tabs_close(self, tab: str) -> dict: + await self.control.close(tab) + return {} + + @extension(BROWSER, op="cookies/get") + async def browser_cookies_get(self, urls: Optional[list[str]] = None) -> dict: + return {"cookies": await self.control.cookies(urls)} + + @extension(BROWSER, op="cookies/set") + async def browser_cookies_set(self, cookies: list[dict]) -> dict: + await self.control.set_cookies(_parsed(CookieToSet, cookies, "cookies")) + return {} + + @extension(BROWSER, op="cookies/delete") + async def browser_cookies_delete(self, name: Optional[str] = None, domain: Optional[str] = None, + path: Optional[str] = None) -> dict: + return {"deleted": await self.control.delete_cookies(name, domain, path)} + + @extension(BROWSER, op="local_storage/get") + async def browser_local_storage_get(self, origins: Optional[list[str]] = None) -> dict: + return {"origins": await self.control.local_storage(origins)} + + @extension(BROWSER, op="local_storage/set") + async def browser_local_storage_set(self, origins: list[dict]) -> dict: + await self.control.set_local_storage(_parsed(OriginStorage, origins, "origins")) + return {} + + @extension(BROWSER, op="local_storage/clear") + async def browser_local_storage_clear(self, origins: Optional[list[str]] = None) -> dict: + return {"origins": await self.control.clear_local_storage(origins)} + + @extension(BROWSER, op="downloads/list", method="GET") + async def browser_downloads_list(self) -> dict: + return {"downloads": self.control.download_list()} + + @extension(BROWSER, op="downloads/get") + async def browser_downloads_get(self, guid: str, write_namespace: Optional[dict] = None) -> dict: + """The download ``guid`` and, completed, its file: uploaded under ``write_namespace``, a grant, when given, + else inline.""" + download, grant = self.control.download(guid), _grant(write_namespace) + home = Path.home().resolve() + landed = Path(download["path"]).resolve() if download["state"] == "completed" and download["path"] else None + parts = await file_parts(home, [landed.relative_to(home).as_posix()], grant) \ + if landed is not None and landed.is_relative_to(home) else [] + return {"download": download, "parts": [part.model_dump(mode="json", exclude_none=True) for part in parts]} + + @extension(BROWSER, op="dialog/answer") + async def browser_dialog_answer(self, accept: bool, text: Optional[str] = None) -> dict: + await self.control.answer_dialog(accept, text) + return {} + + @extension(BROWSER, op="files/choose") + async def browser_files_choose(self, paths: list[str]) -> dict: + await self.control.choose_files(paths) + return {} + + def extension_params(self, uri: str) -> dict: + if uri == BROWSER: + params: dict[str, Any] = {"chromium": self.version, "clock": "real"} + if _alone(): + params["window"] = self.settings.window.model_dump() + return params + return super().extension_params(uri) + + @agent_skill + def browser_skill(self) -> tuple[str, str]: + return "How to use this website browser.", SKILL + + +def _alone() -> bool: + """Whether it runs on a computer with no display, a browser computer's, alone and headless.""" + return computer_app.detect().display == "none" + + +def _parsed(model: type[BaseModel], items: Any, what: str) -> list: + """Each of ``items`` as ``model``; InvalidArgument naming the fields that don't fit and never their values, which + for a cookie are its credentials.""" + if not isinstance(items, list): + raise InvalidArgument(f"{what} is a list") + parsed = [] + for index, item in enumerate(items): + try: + parsed.append(model.model_validate(item)) + except ValidationError as e: + reasons = "; ".join(f"{'.'.join(map(str, error['loc'])) or 'it'}: {error['msg']}" + for error in e.errors(include_input=False, include_url=False)) + raise InvalidArgument(f"{what}[{index}]: {reasons}") + return parsed + + +def _grant(write_namespace: Optional[dict]) -> Optional[WriteNamespaceGrant]: + """``write_namespace`` as a grant; InvalidArgument naming its fields only, since its values carry a signature.""" + if write_namespace is None: + return None + try: + return WriteNamespaceGrant.model_validate(write_namespace) + except ValidationError as e: + fields = ", ".join(".".join(map(str, error["loc"])) or "write_namespace" for error in e.errors()) + raise InvalidArgument(f"write_namespace is not a valid namespace grant ({fields})") + + +def _shm_bytes(path: str = "/dev/shm") -> int: + try: + stats = os.statvfs(path) + except OSError: + return 0 + return stats.f_frsize * stats.f_blocks + + +def _refuse_owned(name: str, launch: Launch) -> None: + if owned := [arg for arg in launch.command if arg.startswith(OWNED_FLAGS)]: + raise ValueError(f"{name}'s launch sets {', '.join(owned)}, which agentenv sets for every website browser") diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/shapes.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/shapes.py new file mode 100644 index 000000000..eff7fced5 --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/shapes.py @@ -0,0 +1,161 @@ +"""The shapes of a website browser's state and of agentenv's record of it: cookies and local storage as Playwright's +``storageState`` has them, so one exported from Playwright loads as is, and agentenv's own records in its names.""" +from __future__ import annotations + +from typing import Literal, Optional +from urllib.parse import urlsplit + +from pydantic import BaseModel, ConfigDict, field_validator, model_validator + +SameSite = Literal["Strict", "Lax", "None"] + + +class PartitionKey(BaseModel): + """CDP's ``Network.CookiePartitionKey``: the top-level site a partitioned cookie was set under.""" + topLevelSite: str + hasCrossSiteAncestor: bool + + +class Cookie(BaseModel): + """A cookie as the browser has it, CDP's ``Network.Cookie`` fields that ``storageState`` keeps: ``expires`` in + seconds since the epoch, -1 for one that ends with the session, and a ``domain`` with a leading dot for one its + subdomains get too; and a partitioned cookie's ``partitionKey``, without which it would be another cookie.""" + name: str + value: str + domain: str + path: str + expires: float + httpOnly: bool + secure: bool + sameSite: SameSite = "Lax" + partitionKey: Optional[PartitionKey] = None + + +class CookieToSet(BaseModel): + """A cookie to set, at ``url``, or at ``domain`` and ``path``, ``/`` by default. ``max_age``, in place of + ``expires``, is seconds from when it's set on the browser's clock; with neither, or ``expires`` -1, it ends with + the session.""" + model_config = ConfigDict(extra="forbid") + + name: str + value: str + url: Optional[str] = None + domain: Optional[str] = None + path: Optional[str] = None + expires: Optional[float] = None + max_age: Optional[float] = None + httpOnly: bool = False + secure: bool = False + sameSite: Optional[SameSite] = None + partitionKey: Optional[PartitionKey] = None + + @model_validator(mode="after") + def _placed_and_timed_once(self) -> CookieToSet: + if not self.url and not self.domain: + raise ValueError(f"cookie {self.name!r} names neither a url nor a domain") + if self.expires is not None and self.max_age is not None: + raise ValueError(f"cookie {self.name!r} has both expires and max_age") + return self + + +class StorageItem(BaseModel): + name: str + value: str + + +class OriginStorage(BaseModel): + """An origin's local storage as ``storageState`` has it.""" + origin: str + localStorage: list[StorageItem] + + @field_validator("origin") + @classmethod + def _an_origin(cls, origin: str) -> str: + return origin_of(origin, strict=True) + + +class OpenTab(BaseModel): + """An open tab, by the browser's target id, with the tab that opened it, a popup's, and whether it's active.""" + id: str + url: str + title: str + opened_at: str + opener: Optional[str] = None + active: bool + + +class ClosedTab(BaseModel): + """A closed tab as of its page's last load, and who closed it: agentenv over its own connection, someone it can't + tell, or the browser's exit or its server's stop.""" + id: str + url: str + title: str + opened_at: str + opener: Optional[str] = None + closed_at: str + closed_by: Literal["agentenv", "unknown", "exit", "stop"] + + +class Download(BaseModel): + """A download into ``~/Downloads``; completed, with where it landed, its size and its sha256.""" + guid: str + url: str + filename: str + state: Literal["inProgress", "completed", "canceled"] + started_at: str + ended_at: Optional[str] = None + path: Optional[str] = None + size: Optional[int] = None + sha256: Optional[str] = None + + +class Dialog(BaseModel): + """A page's JavaScript dialog, its answer and who gave it.""" + tab: Optional[str] + type: str + message: str + default_text: str + url: str + opened_at: str + answer: Optional[Literal["accepted", "dismissed"]] = None + by: Optional[Literal["agentenv", "agent"]] = None + answered_at: Optional[str] = None + input: Optional[str] = None + + +class LaunchEntry(BaseModel): + """A launch, stop or exit of the browser in its app server's log, with a launch's cause and an exit's ``code`` or + the ``signal`` that ended it.""" + at: str + event: Literal["launch", "stop", "exit"] + pid: int + cause: Optional[str] = None + code: Optional[int] = None + signal: Optional[str] = None + + +class BrowserRecord(BaseModel): + """``browser/v1 record``: what agentenv's connection saw ``since`` its server started, and each launch since the + computer booted.""" + since: str + tabs: list[OpenTab] + closed_tabs: list[ClosedTab] + downloads: list[Download] + dialogs: list[Dialog] + launches: list[LaunchEntry] + cdp_connections: list[dict] + + +def origin_of(url: str, *, strict: bool = False) -> Optional[str]: + """``url``'s origin, ``scheme://host[:port]``, for an http or https URL, else None; ``strict``, ValueError for + anything that isn't an origin, a path or query beyond ``/`` included.""" + parts = urlsplit(url) + host = parts.netloc.rsplit("@", 1)[-1].lower() + if parts.scheme not in ("http", "https") or not host: + if strict: + raise ValueError(f"{url!r} isn't an http or https origin") + return None + if strict and (parts.path not in ("", "/") or parts.query or parts.fragment or "@" in parts.netloc): + raise ValueError(f"{url!r} is a URL, not an origin: give only its scheme, host and port") + default = {"http": ":80", "https": ":443"}[parts.scheme] + return f"{parts.scheme}://{host.removesuffix(default)}" diff --git a/packages/agentenv-protocol/tests/test_browser_cdp.py b/packages/agentenv-protocol/tests/test_browser_cdp.py index 9589e689b..2e7e4a698 100644 --- a/packages/agentenv-protocol/tests/test_browser_cdp.py +++ b/packages/agentenv-protocol/tests/test_browser_cdp.py @@ -294,7 +294,7 @@ async def readiness(): return AppReadiness(ready=next(answers, False), problems=[], launched=True, waited_s=0, timeout_s=30) monkeypatch.setattr(browser._computer_app, "readiness", readiness) - monkeypatch.setattr("agentenv_protocol.website_browser.READY_POLL_S", 0.01) + monkeypatch.setattr("agentenv_protocol.website_browser.environment.READY_POLL_S", 0.01) timeout_s = browser._computer_app.timeout_s await browser._until_ready() browser._computer_app.timeout_s = 0.05 diff --git a/packages/agentenv-protocol/tests/test_browser_state.py b/packages/agentenv-protocol/tests/test_browser_state.py new file mode 100644 index 000000000..a6f6e67a9 --- /dev/null +++ b/packages/agentenv-protocol/tests/test_browser_state.py @@ -0,0 +1,314 @@ +"""browser/v1's typed operations on a browser's state: cookies as Playwright's storageState has them, picked for URLs as +Playwright picks them, set with a max_age on this machine's clock and deleted by name, domain and path; local storage +read and written through a page agentenv serves, so the site gets no request, for the origins named or every origin its +pages and frames loaded; downloads with their file; and input refused without its values.""" +from __future__ import annotations + +import asyncio +import base64 +import hashlib + +import pytest + +from agentenv_protocol import website_browser +from agentenv_protocol.computer import Launch, Ready, app, linux +from agentenv_protocol.computer.errors import InvalidArgument +from agentenv_protocol.computer.platform import Platform +from agentenv_protocol.website_browser import (BrowserControl, CookieToSet, DevToolsError, OriginStorage, + WebsiteBrowserEnvironment) +from agentenv_protocol.website_browser.control import ARRIVED, LEAVING, SERVED_PAGE +from agentenv_protocol.website_browser.environment import _parsed + + +@linux(launch=Launch(command=["chromium/chrome"], window={"class": "Chromium"}, ready=Ready(window=True))) +class Browser(WebsiteBrowserEnvironment): + version = "156.0.8078.4" + + +@pytest.fixture(autouse=True) +def browser_home(monkeypatch, tmp_path): + monkeypatch.setattr(app, "detect", lambda: Platform(os="linux")) + monkeypatch.setattr(website_browser.environment, "_shm_bytes", lambda path="/dev/shm": 1 << 30) + monkeypatch.setenv("HOME", str(tmp_path / "home")) + (tmp_path / "home").mkdir() + monkeypatch.setenv("ENVIRONMENT_NAME", "chromium") + return tmp_path / "home" + + +class StubDevTools: + """A browser connection that answers each call from ``answers`` by method, records it, and hands the control the + events a test sends.""" + + def __init__(self, answers=None): + self.calls, self.answers, self.listener = [], answers or {}, None + + def listen(self, listener): + self.listener = listener + + async def call(self, method, params=None, *, session_id=None, timeout_s=None): + self.calls.append((method, params or {}, session_id)) + answer = self.answers.get(method, {}) + return answer(params, session_id) if callable(answer) else answer + + def send(self, method, params, session=None): + self.listener({"method": method, "params": params, **({"sessionId": session} if session else {})}) + + +def _sent(devtools, method): + return [(params, session) for called, params, session in devtools.calls if called == method] + + +CDP_COOKIES = [ + {"name": "sid", "value": "s1", "domain": "shop.test", "path": "/", "expires": -1, "size": 5, "httpOnly": True, + "secure": False, "session": True, "sameSite": "Strict", "priority": "Medium", "sourceScheme": "NonSecure"}, + {"name": "pref", "value": "dark", "domain": ".wiki.test", "path": "/app", "expires": 1900000000.5, "size": 8, + "httpOnly": False, "secure": False, "session": False, "priority": "Medium"}, + {"name": "token", "value": "t1", "domain": "bank.test", "path": "/", "expires": -1, "httpOnly": False, + "secure": True, "session": True, "sameSite": "None"}, + {"name": "token", "value": "t2", "domain": "localhost", "path": "/", "expires": -1, "httpOnly": False, + "secure": True, "session": True, "sameSite": "Lax"}, +] + + +@pytest.mark.asyncio +async def test_its_cookies_are_storage_states_and_a_url_gets_those_a_request_to_it_would_send(): + control = BrowserControl(StubDevTools({"Storage.getCookies": {"cookies": CDP_COOKIES}}), "agent") + + every = await control.cookies() + assert every[0] == {"name": "sid", "value": "s1", "domain": "shop.test", "path": "/", "expires": -1, + "httpOnly": True, "secure": False, "sameSite": "Strict"} + assert every[1]["sameSite"] == "Lax" + assert [cookie["name"] for cookie in await control.cookies(["https://en.wiki.test/app/page"])] == ["pref"] + assert await control.cookies(["https://en.wiki.test/other"]) == [] + assert [c["value"] for c in await control.cookies(["http://bank.test/", "http://localhost:8080/"])] == ["t2"] + assert [c["value"] for c in await control.cookies(["https://bank.test/"])] == ["t1"] + + +@pytest.mark.asyncio +async def test_a_max_age_becomes_an_expiry_on_this_machines_clock_and_a_domain_cookie_gets_its_path(monkeypatch): + devtools = StubDevTools() + control = BrowserControl(devtools, "agent") + monkeypatch.setattr(website_browser.control.time, "time", lambda: 1000.0) + + await control.set_cookies([CookieToSet(name="sid", value="v", domain="shop.test", max_age=60), + CookieToSet(name="s", value="v", url="https://shop.test/cart", expires=-1), + CookieToSet(name="e", value="v", domain=".shop.test", path="/a", expires=5000, + sameSite="None", secure=True)]) + + [(params, _)] = _sent(devtools, "Storage.setCookies") + sent = params["cookies"] + assert sent[0] == {"name": "sid", "value": "v", "domain": "shop.test", "path": "/", "expires": 1060.0, + "httpOnly": False, "secure": False} + assert sent[1] == {"name": "s", "value": "v", "url": "https://shop.test/cart", "httpOnly": False, "secure": False} + assert (sent[2]["path"], sent[2]["expires"], sent[2]["sameSite"]) == ("/a", 5000, "None") + + +@pytest.mark.asyncio +async def test_cookies_are_set_in_one_call_so_one_the_browser_refuses_leaves_none_set(): + def set_cookies(params, session): + if any(cookie["name"] == "bad" for cookie in params["cookies"]): + raise DevToolsError("Storage.setCookies: Invalid cookie fields") + return {} + + devtools = StubDevTools({"Storage.setCookies": set_cookies}) + control = BrowserControl(devtools, "agent") + with pytest.raises(InvalidArgument, match="refused the cookies, so it set none of them: .*Invalid cookie fields") \ + as refused: + await control.set_cookies([CookieToSet(name="ok", value="v", domain="a.test"), + CookieToSet(name="bad", value="v", domain="a.test")]) + assert refused.value.applied == "no" + assert [[c["name"] for c in params["cookies"]] for params, _ in _sent(devtools, "Storage.setCookies")] == [ + ["ok", "bad"]] + + +@pytest.mark.asyncio +async def test_a_partitioned_cookie_keeps_its_key_read_set_and_deleted_beside_an_unpartitioned_one_of_its_name(): + key = {"topLevelSite": "https://top.test", "hasCrossSiteAncestor": True} + cdp = [{"name": "chip", "value": "1", "domain": "embed.test", "path": "/", "expires": -1, "httpOnly": False, + "secure": True, "sameSite": "None", "partitionKey": key}, + {"name": "chip", "value": "2", "domain": "embed.test", "path": "/", "expires": -1, "httpOnly": False, + "secure": True, "sameSite": "None"}] + devtools = StubDevTools({"Storage.getCookies": {"cookies": cdp}, "Target.createTarget": {"targetId": "h"}, + "Target.attachToTarget": {"sessionId": "s-h"}}) + control = BrowserControl(devtools, "agent") + + read = await control.cookies() + assert read[0]["partitionKey"] == key and "partitionKey" not in read[1] + await control.set_cookies([CookieToSet.model_validate(read[0])]) + assert _sent(devtools, "Storage.setCookies")[0][0]["cookies"][0]["partitionKey"] == key + assert await control.delete_cookies(name="chip") == 2 + assert [params.get("partitionKey") for params, _ in _sent(devtools, "Network.deleteCookies")] == [key, None] + + +def test_a_cookie_with_no_place_both_expiries_or_a_field_it_doesnt_take_is_refused_without_its_value(): + for cookie, reason in (({"name": "sid", "value": "SECRET-1"}, "names neither a url nor a domain"), + ({"name": "sid", "value": "SECRET-2", "domain": "a.test", "expires": 5, "max_age": 5}, + "both expires and max_age"), + ({"name": "sid", "value": "SECRET-3", "domain": "a.test", "maxAge": 5}, "maxAge")): + with pytest.raises(InvalidArgument, match=reason) as refused: + _parsed(CookieToSet, [cookie], "cookies") + assert "SECRET" not in str(refused.value) and "cookies[0]" in str(refused.value) + with pytest.raises(InvalidArgument, match="cookies is a list"): + _parsed(CookieToSet, {"name": "sid"}, "cookies") + + +@pytest.mark.asyncio +async def test_cookies_matching_every_field_given_are_deleted_by_their_exact_domain_on_a_page_of_agentenvs_own(): + devtools = StubDevTools({"Storage.getCookies": {"cookies": CDP_COOKIES}, "Target.createTarget": {"targetId": "h"}, + "Target.attachToTarget": {"sessionId": "s-h"}}) + control = BrowserControl(devtools, "agent") + + assert await control.delete_cookies(name="token", domain="bank.test") == 1 + assert _sent(devtools, "Network.deleteCookies") == [({"name": "token", "domain": "bank.test", "path": "/"}, "s-h")] + assert await control.delete_cookies() == 4 + assert await control.delete_cookies(name="nothing") == 0 + + +def _serving_browser(storage, names=None): + """A browser whose pages go where they're sent once agentenv answers their request, each origin named as ``names`` + has it, and whose local storage is ``storage``, by storage key.""" + devtools = StubDevTools() + at, left = {}, {} + + def navigate(params, session): + devtools.send("Fetch.requestPaused", {"requestId": f"r-{params['url']}", "request": {"url": params["url"]}}, + session) + asked = params["url"].rstrip("/") + at[session], left[session] = (names or {}).get(asked, asked), False + return {"frameId": "f"} + + def evaluate(params, session): + if params["expression"] == LEAVING: + left[session] = True + return {"result": {}} + if params["expression"] == ARRIVED: + return {"result": {"value": "" if left.get(session, True) else at[session]}} + return {"result": {"value": ["hidden", False]}} + + devtools.answers.update({ + "Target.createTarget": {"targetId": "h"}, "Target.attachToTarget": {"sessionId": "s-h"}, + "Page.navigate": navigate, "Runtime.evaluate": evaluate, + "DOMStorage.getDOMStorageItems": lambda params, session: { + "entries": [[k, v] for k, v in storage.get(params["storageId"]["storageKey"], {}).items()]}, + "DOMStorage.setDOMStorageItem": lambda params, session: storage.setdefault( + params["storageId"]["storageKey"], {}).update({params["key"]: params["value"]}) or {}, + "DOMStorage.clear": lambda params, session: storage.pop(params["storageId"]["storageKey"], None) and {}}) + return devtools + + +@pytest.mark.asyncio +async def test_local_storage_goes_through_a_page_agentenv_serves_so_the_site_gets_no_request(): + storage = {"https://shop.test/": {"cart": "3"}} + devtools = _serving_browser(storage) + control = BrowserControl(devtools, "agent") + + assert await control.local_storage(["https://shop.test"]) == [ + {"origin": "https://shop.test", "localStorage": [{"name": "cart", "value": "3"}]}] + await control.set_local_storage([OriginStorage.model_validate( + {"origin": "http://wiki.test:8080/", "localStorage": [{"name": "theme", "value": "dark"}]})]) + await asyncio.sleep(0) + + assert storage["http://wiki.test:8080/"] == {"theme": "dark"} + first_calls = [method for method, _, session in devtools.calls if session == "s-h"][:4] + assert first_calls == ["Network.enable", "Network.setBypassServiceWorker", "Fetch.enable", "DOMStorage.enable"] + assert _sent(devtools, "Network.setBypassServiceWorker")[0] == ({"bypass": True}, "s-h") + assert ({"patterns": [{"urlPattern": "*"}]}, "s-h") in _sent(devtools, "Fetch.enable") + fulfilled = _sent(devtools, "Fetch.fulfillRequest") + assert [params["requestId"] for params, _ in fulfilled] == ["r-https://shop.test/", "r-http://wiki.test:8080/"] + assert all(params["body"] == SERVED_PAGE and session == "s-h" for params, session in fulfilled) + assert base64.b64decode(SERVED_PAGE) == b"" + assert ("Target.closeTarget", {"targetId": "h"}, None) in devtools.calls + + devtools.send("Fetch.requestPaused", {"requestId": "an-agents"}, "s-agent") + await asyncio.sleep(0) + assert len(_sent(devtools, "Fetch.fulfillRequest")) == 2 + + +@pytest.mark.asyncio +async def test_with_no_origins_named_it_reads_and_clears_every_origin_its_pages_and_frames_loaded(): + storage = {"https://a.test/": {"k": "1"}, "https://b.test:8443/": {}, "http://frame.test/": {"f": "2"}} + devtools = _serving_browser(storage) + control = BrowserControl(devtools, "agent") + await control.attach() + devtools.send("Target.targetCreated", {"targetInfo": {"targetId": "t1", "type": "page", "url": "https://a.test/x"}}) + devtools.send("Target.targetCreated", {"targetInfo": {"targetId": "t2", "type": "page", "url": "about:blank"}}) + devtools.send("Target.targetInfoChanged", + {"targetInfo": {"targetId": "t2", "type": "page", "url": "https://b.test:8443/q?x=1"}}) + devtools.send("Target.attachedToTarget", + {"sessionId": "s1", "targetInfo": {"targetId": "t1", "type": "page", "url": "https://a.test/x"}}) + devtools.send("Page.frameNavigated", {"frame": {"url": "http://frame.test:80/embed"}}, "s1") + devtools.send("Page.frameNavigated", {"frame": {"url": "data:text/html,hi"}}, "s1") + devtools.send("Page.frameNavigated", {"frame": {"url": "https://unfollowed.test/"}}, "s-not-ours") + await asyncio.sleep(0) + + read = await control.local_storage() + origins = ["http://frame.test", "https://a.test", "https://b.test:8443"] + assert [origin["origin"] for origin in read] == origins + assert read[0]["localStorage"] == [{"name": "f", "value": "2"}] + assert [tab["id"] for tab in await control.tab_list()] == ["t1", "t2"] + assert await control.clear_local_storage() == origins + assert storage == {} + + +@pytest.mark.asyncio +async def test_an_origin_is_kept_as_the_browser_names_it_once_its_page_is_there(): + storage = {"https://xn--bcher-kva.test/": {"k": "v"}} + devtools = _serving_browser(storage, names={"https://bücher.test": "https://xn--bcher-kva.test"}) + control = BrowserControl(devtools, "agent") + + assert await control.local_storage(["https://bücher.test", "https://a.test"]) == [ + {"origin": "https://xn--bcher-kva.test", "localStorage": [{"name": "k", "value": "v"}]}, + {"origin": "https://a.test", "localStorage": []}] + assert control.origins == {"https://xn--bcher-kva.test", "https://a.test"} + + +@pytest.mark.asyncio +async def test_a_url_that_isnt_an_origin_is_refused_and_an_origin_is_kept_as_the_browser_names_it(): + control = BrowserControl(_serving_browser({}), "agent") + for url in ("https://a.test/path", "https://a.test/?q=1", "ftp://a.test", "a.test"): + with pytest.raises(InvalidArgument): + await control.local_storage([url]) + assert OriginStorage(origin="HTTPS://A.test:443/", localStorage=[]).origin == "https://a.test" + assert OriginStorage(origin="http://[::1]:8080", localStorage=[]).origin == "http://[::1]:8080" + + +@pytest.mark.asyncio +async def test_a_download_comes_with_its_file_inline_once_completed_and_a_guid_it_hasnt_seen_is_refused(browser_home): + browser = Browser() + landed = browser_home / "Downloads" / "report.csv" + landed.write_bytes(b"a,b\n1,2\n") + browser.control.downloads["g1"] = {"guid": "g1", "url": "https://shop.test/report", "filename": "report.csv", + "state": "completed", "started_at": "t0", "ended_at": "t1", + "path": str(landed), "size": 8, + "sha256": hashlib.sha256(b"a,b\n1,2\n").hexdigest()} + browser.control.downloads["g2"] = {"guid": "g2", "url": "https://shop.test/big", "filename": "big.zip", + "state": "inProgress", "started_at": "t0"} + + got = await browser.browser_downloads_get("g1") + assert got["download"]["sha256"] == hashlib.sha256(b"a,b\n1,2\n").hexdigest() + [part] = got["parts"] + assert base64.b64decode(part["file"]["bytes"]) == b"a,b\n1,2\n" and part["file"]["name"] == "report.csv" + assert (await browser.browser_downloads_get("g2"))["parts"] == [] + assert [d["guid"] for d in (await browser.browser_downloads_list())["downloads"]] == ["g1", "g2"] + with pytest.raises(InvalidArgument, match="no download nope"): + await browser.browser_downloads_get("nope") + with pytest.raises(InvalidArgument, match=r"write_namespace is not a valid namespace grant \(") as refused: + await browser.browser_downloads_get("g1", write_namespace={"signature": "SIGNED-VALUE"}) + assert "SIGNED-VALUE" not in str(refused.value) + + +@pytest.mark.asyncio +async def test_its_record_and_tab_list_have_the_shapes_and_name_each_tabs_opener(): + devtools = StubDevTools() + control = BrowserControl(devtools, "agent") + await control.attach() + devtools.send("Target.targetCreated", {"targetInfo": {"targetId": "a", "type": "page", "url": "https://a.test/"}}) + devtools.send("Target.targetCreated", + {"targetInfo": {"targetId": "popup", "type": "page", "url": "https://a.test/p", "openerId": "a"}}) + devtools.send("Target.targetDestroyed", {"targetId": "popup"}) + + record = await control.record() + assert set(record["tabs"][0]) == {"id", "url", "title", "opened_at", "opener", "active"} + assert record["tabs"][0]["opener"] is None + assert (record["closed_tabs"][0]["opener"], record["closed_tabs"][0]["closed_by"]) == ("a", "unknown") + assert (await control.tab_list())[0]["id"] == "a" diff --git a/packages/agentenv-protocol/tests/test_website_browser.py b/packages/agentenv-protocol/tests/test_website_browser.py index 80fd9e910..07f906d93 100644 --- a/packages/agentenv-protocol/tests/test_website_browser.py +++ b/packages/agentenv-protocol/tests/test_website_browser.py @@ -76,7 +76,7 @@ def browser_home(monkeypatch, tmp_path): """A browser on Linux, its session user's home under ``tmp_path``, inside a computer with no site proxy and 1 GiB of /dev/shm.""" monkeypatch.setattr(app, "detect", lambda: Platform(os="linux")) - monkeypatch.setattr(website_browser, "_shm_bytes", lambda path="/dev/shm": 1 << 30) + monkeypatch.setattr(website_browser.environment, "_shm_bytes", lambda path="/dev/shm": 1 << 30) monkeypatch.setenv("HOME", str(tmp_path / "home")) (tmp_path / "home").mkdir() monkeypatch.setenv("ENVIRONMENT_NAME", "chromium") @@ -166,7 +166,7 @@ class Widened(WebsiteBrowserEnvironment): @pytest.mark.parametrize("mib, in_files", [(64, True), (1024, False)], ids=["docker-default", "sized"]) def test_it_keeps_its_shared_memory_in_files_only_when_dev_shm_is_under_1_gib(monkeypatch, mib, in_files): - monkeypatch.setattr(website_browser, "_shm_bytes", lambda path="/dev/shm": mib * 2**20) + monkeypatch.setattr(website_browser.environment, "_shm_bytes", lambda path="/dev/shm": mib * 2**20) assert ("--disable-dev-shm-usage" in Browser().launch.command) is in_files diff --git a/tst/integration/computer/browser_computer_test.py b/tst/integration/computer/browser_computer_test.py index 5c095b763..7233d7c6b 100644 --- a/tst/integration/computer/browser_computer_test.py +++ b/tst/integration/computer/browser_computer_test.py @@ -10,7 +10,8 @@ relaunch, and the computer's reset ends it. browser_connect_cdp hands the echo agent its address, with the computer's tools hidden from its role; after reset_env that address fails until the task runs browser_connect_cdp again, whose fresh one its next turn opens the site through. Through an app server and a driver restart and a relaunch, the agent's -address keeps working, and two agents connected at once each drive the browser through their own. Needs Docker and the local registry on +address keeps working, and two agents connected at once each drive the browser through their own. browser/v1 reads +and sets its cookies, local storage and downloads. Needs Docker and the local registry on localhost:5000, and Node 20 or later and npm for the CDP clients.""" import asyncio @@ -45,6 +46,7 @@ from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep from agent_env.task_step.task_steps.reset_env import ResetEnvTaskStep +from tst.integration.computer.browser_state import check_browser_state, check_browser_state_under_chaos from tst.util.a2a_test_agent import put_test_agent pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow] @@ -280,6 +282,44 @@ async def test_an_agent_navigates_moves_in_history_answers_a_dialog_uses_tabs_an assert not report.failed +@pytest.mark.asyncio +async def test_its_cookies_local_storage_and_downloads_are_read_and_set_through_browser_v1(browser_computer): + with namespace_routing(): + deployed = await Env.get(browser_computer.id).deploy() + context = TaskStepContext(deployed_envs=[deployed]) + try: + await _ready(deployed) + card = (await _get(deployed, "/.well-known/agent-env.json")).json() + app = next(child for child in card["children_environments"] if child["name"] == "chromium") + await check_browser_state(deployed, lambda op: deployed.environment_url + + protocol_v1.find_extension_method(app, BROWSER, op)["endpoint"]) + finally: + report = await teardown_run(context) + + assert not report.failed + + +@pytest.mark.asyncio +async def test_its_state_operations_hold_up_at_once_through_a_server_restart_and_fail_at_once_with_the_browser_gone( + browser_computer): + with namespace_routing(): + deployed = await Env.get(browser_computer.id).deploy() + context = TaskStepContext(deployed_envs=[deployed]) + try: + await _ready(deployed) + card = (await _get(deployed, "/.well-known/agent-env.json")).json() + app = next(child for child in card["children_environments"] if child["name"] == "chromium") + kept = await check_browser_state_under_chaos( + deployed, lambda op: deployed.environment_url + protocol_v1.find_extension_method(app, BROWSER, op)[ + "endpoint"], lambda pattern: _kill(deployed, pattern, oldest=pattern.startswith("[-]")), + lambda: _ready(deployed)) + finally: + report = await teardown_run(context) + + print("browser-state-kept", json.dumps(kept)) + assert not report.failed + + async def _shown(session: ClientSession, holds) -> dict: """A screenshot's result once the page it shows is what ``holds`` waits for, within ``WAIT_S``.""" last, deadline = None, time.monotonic() + WAIT_S diff --git a/tst/integration/computer/browser_state.py b/tst/integration/computer/browser_state.py new file mode 100644 index 000000000..c4dc5e01d --- /dev/null +++ b/tst/integration/computer/browser_state.py @@ -0,0 +1,220 @@ +"""browser/v1's typed operations on a deployed chromium, inside a computer or alone: cookies set with a max_age, which +the site gets and the browser expires an hour from the computer's clock, read for the site's URL and deleted by name; +local storage written through a page agentenv serves, so the site sees no request, read for every origin and cleared; +a partitioned cookie read with its key and deleted beside an unpartitioned one of its name; a site's service worker, which answers its pages, kept from agentenv's; +a download with its file; and an operation refused without the control token. Under chaos: calls at once each get +their answer; the pages agentenv serves itself never show in the record; with the browser's server killed and back, +the operations work again; and with the browser killed, they fail at once rather than wait.""" + +import asyncio +import base64 +import json +import time +from typing import Awaitable, Callable + +import httpx + +from agent_env.env.envs.computer.utils import inside + +SITE = "http://127.0.0.1:8779/" +ORIGIN = SITE.rstrip("/") +#: How long the site serves, an effect may take to show, how often it's asked meanwhile, and how far the cookie's +#: expiry may be from an hour after the computer's clock read before it was set. +SERVE_S, WAIT_S, POLL_S, EXPIRY_SLACK_S = 600, 60.0, 0.5, 120 +SERVED = "/tmp/browser-state-served.json" +#: A site on an origin of its own whose service worker answers every navigation with a page that writes to local +#: storage. +SW_SITE = "http://127.0.0.1:8781/" +SW_ORIGIN = SW_SITE.rstrip("/") +# Serves, on loopback for the seconds it's given: on 8779 a page whose title is the Cookie header it got, its +# document.cookie and its local storage, and /download, a page that goes to report.csv, a download; on 8781 a page that +# registers the service worker. Each port and path it's asked for is written to SERVED. +SERVE_STATE = r"""import http.server, json, sys, threading, time +served = [] +WORKER = b"self.addEventListener('install', e => self.skipWaiting()); self.addEventListener('activate', e => e.waitUntil(clients.claim())); self.addEventListener('fetch', e => { if (e.request.mode === 'navigate') e.respondWith(new Response('from-sw', {headers: {'Content-Type': 'text/html'}})); });" +class Page(http.server.BaseHTTPRequestHandler): + def do_GET(self): + port = self.server.server_address[1] + served.append(f"{port}{self.path}") + with open("SERVED_PATH", "w") as file: + json.dump(served, file) + kind = "text/html" + if port == 8781: + body, kind = (WORKER, "text/javascript") if self.path == "/sw.js" else ( + b"registering", kind) + elif self.path == "/report.csv": + body, kind = b"a,b\n1,2\n", "text/csv" + elif self.path == "/download": + body = b"downloading" + else: + body = ("loading" + % json.dumps(self.headers.get("Cookie", ""))).encode() + self.send_response(200) + self.send_header("Content-Type", kind) + self.send_header("Content-Length", str(len(body))) + if self.path == "/report.csv": + self.send_header("Content-Disposition", "attachment; filename=report.csv") + self.end_headers() + self.wfile.write(body) + def log_message(self, *args): + pass +for port in (8779, 8781): + server = http.server.ThreadingHTTPServer(("127.0.0.1", port), Page) + threading.Thread(target=server.serve_forever, daemon=True).start() +time.sleep(float(sys.argv[1])) +""".replace("SERVED_PATH", SERVED) + + +def _client(deployed, endpoint: Callable[[str], str]): + """``call``, an operation's response, and ``ok``, its result once it answered 200.""" + async def call(op: str, params: dict | None = None, *, method: str = "POST", headers=None) -> httpx.Response: + headers = deployed.control_headers if headers is None else headers + async with httpx.AsyncClient(timeout=60) as http: + if method == "GET": + return await http.get(endpoint(op), headers=headers) + return await http.post(endpoint(op), json=params or {}, headers=headers) + + async def ok(op: str, params: dict | None = None, *, method: str = "POST") -> dict: + response = await call(op, params, method=method) + assert response.status_code == 200, f"{op}: {response.status_code} {response.text}" + return response.json() + + return call, ok + + +def _serve(deployed) -> asyncio.Future: + return asyncio.ensure_future(inside.run(deployed, ["python3", "-c", SERVE_STATE, str(SERVE_S)], + timeout_s=SERVE_S + 60)) + + +async def check_browser_state(deployed, endpoint: Callable[[str], str]) -> None: + """Every cookie, local storage and download operation of ``browser/v1`` on the deployed chromium, each at + ``endpoint(op)``, as the module says.""" + call, ok = _client(deployed, endpoint) + + async def page_at(url: str) -> dict: + await ok("navigate", {"url": url}) + last, deadline = None, time.monotonic() + WAIT_S + while time.monotonic() < deadline: + tabs = (await ok("tabs/list", method="GET"))["tabs"] + active = next((tab for tab in tabs if tab["active"]), None) + if active and active["url"] == url and active["title"].startswith("{"): + return json.loads(active["title"]) + last = tabs + await asyncio.sleep(POLL_S) + raise AssertionError(f"the page at {url} didn't load within {WAIT_S:g} s; the tabs were {last!r}") + + async def titled(url: str, title: str) -> dict: + await ok("navigate", {"url": url}) + last, deadline = None, time.monotonic() + WAIT_S + while time.monotonic() < deadline: + tabs = (await ok("tabs/list", method="GET"))["tabs"] + if (active := next((tab for tab in tabs if tab["active"]), None)) and active["title"] == title: + return active + last = tabs + await asyncio.sleep(POLL_S) + raise AssertionError(f"the page at {url} wasn't {title!r} within {WAIT_S:g} s; the tabs were {last!r}") + + async def served() -> list[str]: + out = await inside.run(deployed, ["sh", "-c", f"cat {SERVED} 2>/dev/null || echo []"], timeout_s=30) + return json.loads(out["stdout"]) + + serving = _serve(deployed) + try: + await asyncio.sleep(3) + refused = await call("cookies/get", {}, headers={}) + now = float((await inside.run(deployed, ["date", "+%s"], timeout_s=30))["stdout"]) + await ok("cookies/set", {"cookies": [{"name": "sid", "value": "v1", "url": SITE, "max_age": 3600}, + {"name": "pref", "value": "dark", "domain": "127.0.0.1", + "httpOnly": True}]}) + cookies = {cookie["name"]: cookie for cookie in (await ok("cookies/get", {"urls": [SITE]}))["cookies"]} + await ok("local_storage/set", {"origins": [{"origin": ORIGIN, "localStorage": [{"name": "cart", "value": "3"}]}]}) + before_visit = await served() + signed_in = await page_at(SITE) + every_origin = (await ok("local_storage/get"))["origins"] + deleted = (await ok("cookies/delete", {"name": "sid"}))["deleted"] + left = [cookie["name"] for cookie in (await ok("cookies/get", {"urls": [SITE]}))["cookies"]] + cleared = (await ok("local_storage/clear", {"origins": [ORIGIN]}))["origins"] + after_clear = (await ok("local_storage/get", {"origins": [ORIGIN]}))["origins"] + signed_out = await page_at(f"{SITE}?again") + chip = {"name": "chip", "value": "p", "domain": "embed.test", "path": "/", "secure": True, "sameSite": "None"} + key = {"topLevelSite": "https://top.test", "hasCrossSiteAncestor": True} + await ok("cookies/set", {"cookies": [{**chip, "partitionKey": key}, {**chip, "value": "u"}]}) + chips = [c for c in (await ok("cookies/get"))["cookies"] if c["name"] == "chip"] + chips_deleted = (await ok("cookies/delete", {"name": "chip"}))["deleted"] + chips_left = [c for c in (await ok("cookies/get"))["cookies"] if c["name"] == "chip"] + await titled(f"{SW_SITE}register", "registered") + worker_site_before = [path for path in await served() if path.startswith("8781")] + worker_bypassed = (await ok("local_storage/get", {"origins": [SW_ORIGIN]}))["origins"] + worker_site_after = [path for path in await served() if path.startswith("8781")] + await titled(f"{SW_SITE}any", "from-sw") + worker_answered = (await ok("local_storage/get", {"origins": [SW_ORIGIN]}))["origins"] + await ok("navigate", {"url": f"{SITE}download"}) + download, deadline = None, time.monotonic() + WAIT_S + while time.monotonic() < deadline and not (download and download["sha256"]): + download = next((d for d in (await ok("downloads/list", method="GET"))["downloads"] + if d["filename"] == "report.csv"), None) + await asyncio.sleep(POLL_S) + got = await ok("downloads/get", {"guid": download["guid"]}) if download else None + finally: + serving.cancel() + + assert refused.status_code in (401, 403), (refused.status_code, refused.text) + assert abs(cookies["sid"]["expires"] - (now + 3600)) < EXPIRY_SLACK_S, (cookies["sid"], now) + assert (cookies["pref"]["domain"], cookies["pref"]["httpOnly"], cookies["pref"]["expires"]) == ("127.0.0.1", True, -1) + assert before_visit == [], before_visit + assert "sid=v1" in signed_in["header"] and "pref=dark" in signed_in["header"], signed_in + assert signed_in["cookie"] == "sid=v1" and signed_in["storage"] == {"cart": "3"}, signed_in + assert {"origin": ORIGIN, "localStorage": [{"name": "cart", "value": "3"}]} in every_origin, every_origin + assert (deleted, left, cleared) == (1, ["pref"], [ORIGIN]) + assert after_clear == [{"origin": ORIGIN, "localStorage": []}], after_clear + assert "sid" not in signed_out["header"] and signed_out["storage"] == {}, signed_out + assert sorted((c["value"], c.get("partitionKey")) for c in chips) == sorted([("p", key), ("u", None)]), chips + assert (chips_deleted, chips_left) == (2, []), chips_left + assert worker_bypassed == [{"origin": SW_ORIGIN, "localStorage": []}], worker_bypassed + assert worker_site_after == worker_site_before, (worker_site_before, worker_site_after) + assert {"name": "sw_ran", "value": "yes"} in worker_answered[0]["localStorage"], worker_answered + assert got is not None and got["download"]["state"] == "completed", download + assert base64.b64decode(got["parts"][0]["file"]["bytes"]) == b"a,b\n1,2\n", got + + +async def check_browser_state_under_chaos(deployed, endpoint: Callable[[str], str], kill: Callable[[str], Awaitable], + ready: Callable[[], Awaitable]) -> dict: + """The chaos the module names, ``kill(pattern)`` killing the newest process matching it, or the oldest for the + browser, and ``ready()`` waiting for the computer to be ready. Returns whether the cookie and the item set before + the server was killed were still there after, which depends on what the browser had written to its profile.""" + call, ok = _client(deployed, endpoint) + item = {"origin": ORIGIN, "localStorage": [{"name": "k", "value": "1"}]} + serving = _serve(deployed) + try: + await asyncio.sleep(3) + await ok("cookies/set", {"cookies": [{"name": "kept", "value": "1", "url": SITE, "max_age": 3600}]}) + await ok("local_storage/set", {"origins": [item]}) + at_once = await asyncio.gather(*[ok("local_storage/get", {"origins": [ORIGIN]}) for _ in range(4)], + ok("cookies/get", {"urls": [SITE]}), ok("local_storage/set", {"origins": [ + {"origin": ORIGIN, "localStorage": [{"name": "k2", "value": "2"}]}]}), + ok("local_storage/clear", {"origins": ["http://127.0.0.1:8780"]})) + after = (await ok("local_storage/get", {"origins": [ORIGIN]}))["origins"] + recorded = await ok("record", method="GET") + await kill("[s]erve_app") + await ready() + cookies_back = [c["name"] for c in (await ok("cookies/get", {"urls": [SITE]}))["cookies"]] + storage_back = (await ok("local_storage/get", {"origins": [ORIGIN]}))["origins"][0]["localStorage"] + recorded_back = await ok("record", method="GET") + await kill("[-]-remote-debugging-pipe") + started = time.monotonic() + down = [await call("cookies/get", {"urls": [SITE]}), await call("local_storage/get", {"origins": [ORIGIN]})] + down_s = time.monotonic() - started + finally: + serving.cancel() + + for answer in at_once[:4]: + assert {"name": "k", "value": "1"} in answer["origins"][0]["localStorage"], at_once + assert "kept" in [c["name"] for c in at_once[4]["cookies"]], at_once[4] + assert {"name": "k2", "value": "2"} in after[0]["localStorage"], after + for record in (recorded, recorded_back): + assert [tab["url"] for tab in record["tabs"]] == ["about:blank"] and record["closed_tabs"] == [], record + assert all(answer.status_code >= 400 for answer in down) and down_s < 30, ([a.text for a in down], down_s) + return {"cookie_kept": "kept" in cookies_back, "item_kept": {"name": "k", "value": "1"} in storage_back} diff --git a/tst/integration/computer/chromium_app_test.py b/tst/integration/computer/chromium_app_test.py index aac3fb21f..caac65b07 100644 --- a/tst/integration/computer/chromium_app_test.py +++ b/tst/integration/computer/chromium_app_test.py @@ -7,7 +7,8 @@ launcher, run while it runs, opens another window of it with no second copy, its cache kept and it still ready; and the desktop that gives it French defaults deploys with them, a deploy's time zone over them, each recorded; and the record its control connection keeps lists a tab an agent opened and closed, a download with its size and sha256, a confirm() -accepted for it, and each launch, exit and stop with its cause, across a kill and a restart of its server. Needs Docker +accepted for it, and each launch, exit and stop with its cause, across a kill and a restart of its server; and browser/v1 reads and sets its +cookies, local storage and downloads. Needs Docker and the local registry on localhost:5000.""" import asyncio @@ -24,6 +25,7 @@ from agent_env.env import Env from agentenv_protocol.types import WELL_KNOWN_PATH from agentenv_protocol.website_browser import BROWSER +from tst.integration.computer.browser_state import check_browser_state from tst.integration.computer.website_browser_test import _app, _browser_process, _keys, _ready from agent_env.env.envs.computer.utils import inside from agent_env.store.routing import namespace_routing @@ -304,13 +306,29 @@ async def test_with_dialogs_left_to_the_agent_chromium_shows_one_and_records_the assert not report.failed +@pytest.mark.asyncio +async def test_its_cookies_local_storage_and_downloads_are_read_and_set_through_browser_v1(chromium_desktop): + with namespace_routing(): + deployed = await Env.get(chromium_desktop.id).deploy() + context = TaskStepContext(deployed_envs=[deployed]) + try: + card = await _app(deployed, WELL_KNOWN_PATH, "chromium") + await _ready(deployed, card, "chromium") + await check_browser_state(deployed, lambda op: f"{deployed.environment_url}/svc/chromium" + f"{protocol_v1.find_extension_method(card, BROWSER, op)['endpoint']}") + finally: + report = await teardown_run(context) + + assert not report.failed + + def _report(n: int) -> bytes: """The content SERVE_RECORDED serves at its nth request for report.bin.""" return b"agentenv report %d\n" % n * 64 def _hashed(record: dict) -> list[dict]: - return [download for download in record["downloads"] if "sha256" in download] + return [download for download in record["downloads"] if download["sha256"]] async def _record(deployed, card: dict, until) -> dict: diff --git a/tst/integration/computer/website_browser_test.py b/tst/integration/computer/website_browser_test.py index 5027a9f7b..732a24770 100644 --- a/tst/integration/computer/website_browser_test.py +++ b/tst/integration/computer/website_browser_test.py @@ -131,9 +131,11 @@ async def test_chromium_runs_with_agentenvs_flags_on_its_pipe_and_a_reset_relaun policies["stdout"]).items() params = protocol_v1.extension_params(card, BROWSER) assert (params["chromium"], params["clock"], sorted(params["methods"])) == ( - "156.0.8078.4", "real", ["cdp", "dialog/answer", "files/choose", "history", "navigate", "page/capture", - "page/input", "page/state", "page/viewport", "record", "tabs/activate", "tabs/close", - "tabs/list", "tabs/open"]) + "156.0.8078.4", "real", ["cdp", "cookies/delete", "cookies/get", "cookies/set", "dialog/answer", "downloads/get", + "downloads/list", "files/choose", "history", "local_storage/clear", + "local_storage/get", "local_storage/set", "navigate", "page/capture", "page/input", + "page/state", "page/viewport", "record", "tabs/activate", "tabs/close", "tabs/list", + "tabs/open"]) assert readiness["ready"] is True, readiness assert relaunched["pid"] != browser["pid"] and relaunched["listening"] == 0 assert relaunched_readiness["ready"] is True, relaunched_readiness