2148 lines
105 KiB
Python
2148 lines
105 KiB
Python
"""Restricted JSONL worker for the fast-browser prototype.
|
|
|
|
The worker deliberately delegates browser ownership to browser-harness. It does
|
|
not contain a second CDP client, navigation API, JavaScript execution API, or
|
|
model-facing browser handle. Only the small observe/click/fill surface below
|
|
is exposed to the TypeScript adapter. The session runtime below adds only
|
|
fixed, typed read/assertion, tab, navigation, and bounded evidence operations.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ipaddress
|
|
import signal
|
|
import subprocess
|
|
import json
|
|
import os
|
|
import secrets
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
import base64
|
|
import re
|
|
from dataclasses import dataclass, replace
|
|
from pathlib import Path
|
|
from typing import Any, Iterable, Protocol
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
# Protocol limits are intentionally small. In particular, page text never
|
|
# becomes an unbounded JSONL response.
|
|
# Runtime evidence carries bounded base64 bytes; keep JSONL framing bounded
|
|
# above the published 8 MiB screenshot ceiling while retaining one line/request.
|
|
MAX_LINE_BYTES = 12 * 1024 * 1024
|
|
MAX_OUTPUT_BYTES = 12 * 1024 * 1024
|
|
MAX_CANDIDATES = 60
|
|
MAX_METADATA_BATCH = 16
|
|
MAX_OBSERVATIONS = 8
|
|
MAX_STRING = 4_000
|
|
MAX_NAME = 240
|
|
MAX_FILL_TEXT = 8_000
|
|
MAX_READ_TEXT = 16_000
|
|
# JSONL transport remains bounded below the existing bridge line ceiling. The
|
|
# parent runtime writes these bytes to its own PrivateArtifactStore.
|
|
MAX_SCREENSHOT_BYTES = 8 * 1024 * 1024
|
|
MAX_DIAGNOSTICS_BYTES = 256 * 1024
|
|
MAX_EVENTS = 256
|
|
MAX_CONSOLE_FAILURES = 64
|
|
MAX_NETWORK_FAILURES = 64
|
|
MAX_EVENT_MESSAGE = 1_000
|
|
ACTION_TIMEOUT_SECONDS = 3.0
|
|
READ_ASSERTION_TIMEOUT_SECONDS = 3.0
|
|
NAVIGATION_TIMEOUT_SECONDS = 10.0
|
|
# Leave a small IPC margin below the published 5s popup budget so a normal
|
|
# timeout returns a structured result instead of racing the parent deadline.
|
|
POPUP_WAIT_SECONDS = 4.75
|
|
EXPECTED_BROWSER_HARNESS_VERSION = "0.1.13"
|
|
_INTERNAL_TARGET_PREFIXES = (
|
|
"chrome://", "chrome-untrusted://", "devtools://", "chrome-extension://", "about:",
|
|
)
|
|
_TEXT_INPUT_TYPES = {"", "text", "search", "email", "url", "tel", "number", "password"}
|
|
|
|
_FILL_ROLES = {"textbox", "searchbox", "spinbutton"}
|
|
_CLICK_ROLES = {
|
|
"button", "link", "checkbox", "radio", "switch", "tab", "menuitem",
|
|
"option", "combobox", "listbox", "treeitem",
|
|
}
|
|
|
|
|
|
class WorkerError(RuntimeError):
|
|
"""A safe, non-sensitive error suitable for the JSONL response."""
|
|
|
|
|
|
class BrowserOperationError(WorkerError):
|
|
"""Structured operation failure with no raw browser exception attached."""
|
|
|
|
def __init__(self, details: dict[str, Any]):
|
|
self.details = details
|
|
super().__init__(str(details.get("message", "browser operation failed")))
|
|
|
|
|
|
def _operation_error(
|
|
operation: str,
|
|
code: str,
|
|
phase: str,
|
|
message: str,
|
|
*,
|
|
mutation_state: str = "not_attempted",
|
|
retry: str = "never",
|
|
) -> BrowserOperationError:
|
|
# Keep this shape aligned with BrowserErrorDetails. Messages are selected
|
|
# from fixed, non-sensitive strings by callers; browser exception text is
|
|
# deliberately never inserted here.
|
|
return BrowserOperationError({
|
|
"code": code,
|
|
"operation": operation,
|
|
"phase": phase,
|
|
"message": message[:MAX_STRING],
|
|
"mutationState": mutation_state,
|
|
"retry": retry,
|
|
})
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ElementMetadata:
|
|
backend_id: int
|
|
tag: str
|
|
input_type: str
|
|
role: str
|
|
name: str
|
|
visible: bool
|
|
disabled: bool
|
|
connected: bool = True
|
|
main_frame: bool = True
|
|
contenteditable: bool = False
|
|
ax_role: str = ""
|
|
ax_name: str = ""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CandidateRecord:
|
|
id: str
|
|
backend_id: int
|
|
role: str
|
|
name: str
|
|
actions: tuple[str, ...]
|
|
metadata: ElementMetadata
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ObservationRecord:
|
|
observation_id: str
|
|
target_id: str
|
|
frame_id: str
|
|
loader_id: str
|
|
url: str
|
|
title: str
|
|
candidates: tuple[CandidateRecord, ...]
|
|
|
|
|
|
def _safe_event_message(value: Any) -> str:
|
|
if not isinstance(value, str):
|
|
return "browser event"
|
|
value = " ".join(value.split())
|
|
# Console data is untrusted page content. It remains local evidence, but
|
|
# public shaping must not carry URL query/fragment tokens or obvious
|
|
# credential-bearing key/value diagnostics.
|
|
value = re.sub(r"https?://[^\s?#]+(?:[?#][^\s]*)?", lambda match: (_origin(match.group(0)) or "[url]"), value)
|
|
value = re.sub(r"(?i)(password|passwd|secret|token|authorization|cookie)(\s*[:=]\s*)[^\s,;]+", r"\1\2[redacted]", value)
|
|
value = "".join(ch for ch in value if ord(ch) >= 0x20 or ch in "\t\n")
|
|
return value[:MAX_EVENT_MESSAGE] or "browser event"
|
|
|
|
|
|
def _policy_event_message(value: str, policy: OriginPolicy) -> str:
|
|
return re.sub(
|
|
r"https?://[^\s?#]+(?:[?#][^\s]*)?",
|
|
lambda match: (_origin(match.group(0)) if _approved_origin(_origin(match.group(0)), policy) else "[url]"),
|
|
value,
|
|
)
|
|
|
|
|
|
def _safe_component(value: Any, label: str) -> str:
|
|
if not isinstance(value, str) or not value or len(value) > MAX_STRING or "\x00" in value:
|
|
raise WorkerError(f"invalid {label}")
|
|
# IDs may be opaque but must not become path traversal or separator input.
|
|
if value in {".", ".."} or "/" in value or "\\" in value:
|
|
raise WorkerError(f"invalid {label}")
|
|
return value
|
|
|
|
|
|
class BrowserBackend(Protocol):
|
|
"""The deliberately tiny seam used by offline tests and the real adapter."""
|
|
|
|
def current_tab(self) -> dict[str, Any]: ...
|
|
def document_identity(self) -> tuple[str, str]: ...
|
|
def accessibility_nodes(self) -> Iterable[dict[str, Any]]: ...
|
|
def metadata(self, backend_id: int) -> ElementMetadata: ...
|
|
def click(self, backend_id: int) -> None: ...
|
|
def fill(self, backend_id: int, text: str) -> None: ...
|
|
def value_matches(self, backend_id: int, text: str) -> bool: ...
|
|
def visible_text(self, text: str) -> bool: ...
|
|
def close(self) -> None: ...
|
|
|
|
|
|
def _origin(url: str) -> str | None:
|
|
"""Return a normalized web origin, rejecting non-web and malformed URLs."""
|
|
try:
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme.lower() not in {"http", "https"} or not parsed.hostname or "*" in parsed.hostname:
|
|
return None
|
|
if parsed.username is not None or parsed.password is not None:
|
|
return None
|
|
port = parsed.port
|
|
host = parsed.hostname.lower()
|
|
if "*" in host:
|
|
return None
|
|
try:
|
|
address = ipaddress.ip_address(host)
|
|
host = f"[{host}]" if address.version == 6 else host
|
|
except ValueError:
|
|
pass
|
|
if (parsed.scheme.lower() == "http" and port == 80) or (
|
|
parsed.scheme.lower() == "https" and port == 443
|
|
):
|
|
port = None
|
|
netloc = host if port is None else f"{host}:{port}"
|
|
return urlunsplit((parsed.scheme.lower(), netloc, "", "", ""))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
OriginPolicy = str | frozenset[str]
|
|
|
|
|
|
def normalize_allowed_origins(origins: str | Iterable[str]) -> OriginPolicy:
|
|
if origins == "*":
|
|
return "*"
|
|
if isinstance(origins, str):
|
|
raise ValueError("allowed origins must be '*' or a JSON list")
|
|
normalized: set[str] = set()
|
|
for raw in origins:
|
|
if not isinstance(raw, str) or not raw or len(raw) > MAX_STRING:
|
|
raise ValueError("allowed origins must be short origin strings")
|
|
value = _origin(raw)
|
|
if value is None or raw not in {value, value + "/"}:
|
|
# Requiring origin-only input avoids silently broadening an
|
|
# allowlist containing a path, query, or credential.
|
|
raise ValueError(f"allowed origin is not an exact web origin: {raw!r}")
|
|
normalized.add(value)
|
|
if not normalized:
|
|
raise ValueError("at least one allowed origin is required")
|
|
return frozenset(normalized)
|
|
|
|
|
|
def _approved_origin(origin: str | None, policy: OriginPolicy) -> bool:
|
|
return origin is not None and (policy == "*" or origin in policy)
|
|
|
|
|
|
def validate_cdp_url(value: str) -> str:
|
|
if not isinstance(value, str) or len(value) > MAX_STRING:
|
|
raise ValueError("cdpUrl must be a short URL")
|
|
try:
|
|
parsed = urlsplit(value)
|
|
host = (parsed.hostname or "").lower()
|
|
try:
|
|
address = ipaddress.ip_address(host) if host else None
|
|
except ValueError:
|
|
address = None
|
|
loopback = host == "localhost" or (address is not None and address.is_loopback)
|
|
if parsed.scheme.lower() != "http" or not loopback or parsed.username is not None or parsed.password is not None:
|
|
raise ValueError
|
|
if parsed.port is None or not (1 <= parsed.port <= 65535):
|
|
raise ValueError
|
|
if parsed.fragment or parsed.query or parsed.path not in {"", "/"}:
|
|
raise ValueError
|
|
except (ValueError, TypeError):
|
|
raise ValueError("cdpUrl must be an http loopback endpoint") from None
|
|
return value
|
|
|
|
|
|
def _short(value: Any, limit: int = MAX_STRING) -> str:
|
|
if not isinstance(value, str):
|
|
raise WorkerError("expected string")
|
|
if len(value) > limit:
|
|
raise WorkerError("string exceeds protocol limit")
|
|
return value
|
|
|
|
|
|
def _ax_value(value: Any) -> str:
|
|
if isinstance(value, dict):
|
|
value = value.get("value", "")
|
|
return value if isinstance(value, str) else str(value or "")
|
|
|
|
|
|
def _safe_name(value: str, limit: int = MAX_NAME) -> str:
|
|
# The worker never reads or returns an input's current value. Strip
|
|
# controls so an AX label cannot forge a JSONL/log line in a downstream UI.
|
|
value = " ".join(value.split())
|
|
return value[:limit] or "(unnamed)"
|
|
|
|
|
|
def _candidate_actions(meta: ElementMetadata, ax_role: str) -> tuple[str, ...]:
|
|
role = (ax_role or meta.role or "").lower()
|
|
tag = meta.tag.lower()
|
|
input_type = meta.input_type.lower()
|
|
if input_type in {"hidden", "file"} or not meta.visible or meta.disabled or not meta.main_frame or meta.contenteditable:
|
|
return ()
|
|
actions: list[str] = []
|
|
if (tag == "textarea" and role not in {"combobox"}) or (tag == "input" and input_type in _TEXT_INPUT_TYPES):
|
|
actions.append("fill")
|
|
if role in _CLICK_ROLES or tag in {"button", "a", "summary"} or input_type in {"button", "submit", "reset", "checkbox", "radio"}:
|
|
actions.append("click")
|
|
# A combobox is exposed for filling only unless its AX role explicitly
|
|
# describes a clickable widget. Keep the list stable and duplicate-free.
|
|
return tuple(dict.fromkeys(actions))
|
|
|
|
|
|
def _is_visible_text(value: str) -> bool:
|
|
return bool(value and len(value) <= MAX_STRING)
|
|
|
|
|
|
class BrowserWorker:
|
|
"""Stateful, serialized implementation of the restricted bridge protocol."""
|
|
|
|
def __init__(self, backend: BrowserBackend, allowed_origins: OriginPolicy):
|
|
self.backend = backend
|
|
self.allowed_origins = normalize_allowed_origins(allowed_origins)
|
|
self._counter = 0
|
|
self._observations: dict[str, ObservationRecord] = {}
|
|
self._closed = False
|
|
self._blocked = False
|
|
# Explicit mutation boundary consumed by the session runtime. It is
|
|
# set before invoking the backend and never inferred from exception
|
|
# wording, which may be arbitrary browser text.
|
|
self._mutation_dispatched = False
|
|
self._mutation_verified = False
|
|
|
|
def _require_open(self) -> None:
|
|
if self._closed:
|
|
raise WorkerError("worker is closed")
|
|
if self._blocked:
|
|
raise WorkerError("browser left the approved origin; bridge stopped")
|
|
|
|
def _approved_url(self, url: str) -> str:
|
|
origin = _origin(url)
|
|
if not _approved_origin(origin, self.allowed_origins):
|
|
self._blocked = True
|
|
raise WorkerError("browser URL is outside the approved origins")
|
|
return url
|
|
|
|
def _state(self) -> tuple[dict[str, Any], str, str]:
|
|
tab = self.backend.current_tab()
|
|
if not isinstance(tab, dict):
|
|
raise WorkerError("invalid browser tab state")
|
|
target_id = tab.get("targetId") or tab.get("target_id")
|
|
url = tab.get("url")
|
|
title = tab.get("title") or ""
|
|
if not isinstance(target_id, str) or not target_id or not isinstance(url, str):
|
|
raise WorkerError("invalid browser tab identity")
|
|
self._approved_url(url)
|
|
frame_id, loader_id = self.backend.document_identity()
|
|
if not isinstance(frame_id, str) or not isinstance(loader_id, str):
|
|
raise WorkerError("invalid main-frame identity")
|
|
return {"target_id": target_id, "url": url, "title": _short(title, MAX_STRING)}, frame_id, loader_id
|
|
|
|
def observe(self) -> dict[str, Any]:
|
|
self._require_open()
|
|
tab, frame_id, loader_id = self._state()
|
|
self._counter += 1
|
|
observation_id = f"obs-{self._counter}-{secrets.token_hex(4)}"
|
|
records: list[CandidateRecord] = []
|
|
seen: set[int] = set()
|
|
batch_metadata = getattr(self.backend, "metadata_many", None)
|
|
pending: list[tuple[int, str, str]] = []
|
|
|
|
def consume(items: list[tuple[int, str, str]]) -> None:
|
|
if not items:
|
|
return
|
|
metadata_by_id: dict[int, ElementMetadata] = {}
|
|
if callable(batch_metadata):
|
|
try:
|
|
result = batch_metadata([backend_id for backend_id, _role, _name in items])
|
|
except Exception:
|
|
result = None
|
|
if isinstance(result, dict):
|
|
for backend_id, meta in result.items():
|
|
if type(backend_id) is int and isinstance(meta, ElementMetadata):
|
|
metadata_by_id[backend_id] = meta
|
|
for backend_id, ax_role, ax_name in items:
|
|
meta = metadata_by_id.get(backend_id)
|
|
if meta is None or meta.backend_id != backend_id:
|
|
try:
|
|
meta = self.backend.metadata(backend_id)
|
|
except Exception:
|
|
continue
|
|
if not meta.connected or meta.backend_id != backend_id:
|
|
continue
|
|
# Full AX discovery is authoritative for observation roles and
|
|
# click names. DOM metadata remains the only source for fill
|
|
# labels, so a field's current value cannot enter an observation.
|
|
observed_meta = replace(meta, ax_role=ax_role, ax_name=ax_name)
|
|
candidate_role = (ax_role or observed_meta.role).lower()
|
|
actions = _candidate_actions(observed_meta, candidate_role)
|
|
if not actions:
|
|
continue
|
|
name = _safe_name(
|
|
observed_meta.name if "fill" in actions
|
|
else (ax_name or observed_meta.ax_name or observed_meta.name)
|
|
)
|
|
candidate_id = f"c-{len(records)}"
|
|
records.append(CandidateRecord(
|
|
candidate_id, backend_id, candidate_role, name, actions, observed_meta
|
|
))
|
|
if len(records) >= MAX_CANDIDATES:
|
|
break
|
|
|
|
for node in self.backend.accessibility_nodes():
|
|
if len(records) >= MAX_CANDIDATES:
|
|
break
|
|
if not isinstance(node, dict) or node.get("ignored") is True:
|
|
continue
|
|
backend_id = node.get("backendDOMNodeId")
|
|
if type(backend_id) is not int or backend_id <= 0 or backend_id in seen:
|
|
continue
|
|
seen.add(backend_id)
|
|
ax_role = _ax_value(node.get("role", "")).lower()
|
|
ax_name = _ax_value(node.get("name", ""))
|
|
# Do not spend several CDP round trips on static text/layout nodes.
|
|
if ax_role not in _FILL_ROLES | _CLICK_ROLES:
|
|
continue
|
|
pending.append((backend_id, ax_role, ax_name))
|
|
if len(pending) >= MAX_METADATA_BATCH:
|
|
consume(pending)
|
|
pending.clear()
|
|
consume(pending)
|
|
snapshot = ObservationRecord(
|
|
observation_id, tab["target_id"], frame_id, loader_id, tab["url"], tab["title"], tuple(records)
|
|
)
|
|
self._observations[observation_id] = snapshot
|
|
while len(self._observations) > MAX_OBSERVATIONS:
|
|
self._observations.pop(next(iter(self._observations)))
|
|
return {
|
|
"observationId": observation_id,
|
|
"url": tab["url"],
|
|
"title": tab["title"],
|
|
"candidates": [
|
|
{"id": c.id, "role": c.role, "name": c.name, "actions": list(c.actions)}
|
|
for c in records
|
|
],
|
|
}
|
|
|
|
def _find_snapshot(self, request: dict[str, Any]) -> tuple[ObservationRecord, CandidateRecord]:
|
|
observation_id = request.get("observationId")
|
|
candidate_id = request.get("candidateId")
|
|
if not isinstance(observation_id, str) or not isinstance(candidate_id, str):
|
|
raise WorkerError("invalid action target")
|
|
snapshot = self._observations.pop(observation_id, None)
|
|
# Every attempted mutation invalidates the entire prior observation set,
|
|
# not only the particular handle chosen for this action.
|
|
self._observations.clear()
|
|
if snapshot is None:
|
|
raise WorkerError("observation is missing or already consumed")
|
|
candidate = next((item for item in snapshot.candidates if item.id == candidate_id), None)
|
|
if candidate is None:
|
|
raise WorkerError("candidate is not in the observation")
|
|
return snapshot, candidate
|
|
|
|
def _revalidate(self, snapshot: ObservationRecord, candidate: CandidateRecord, action: str) -> None:
|
|
tab, frame_id, loader_id = self._state()
|
|
if (
|
|
tab["target_id"] != snapshot.target_id
|
|
or tab["url"] != snapshot.url
|
|
or frame_id != snapshot.frame_id
|
|
or loader_id != snapshot.loader_id
|
|
):
|
|
raise WorkerError("observation is stale")
|
|
if action not in candidate.actions:
|
|
raise WorkerError("action is not allowed for candidate")
|
|
current = self.backend.metadata(candidate.backend_id)
|
|
current_role = (current.ax_role or current.role).lower()
|
|
current_name = current.name if action == "fill" else (current.ax_name or current.name)
|
|
current_actions = _candidate_actions(current, current_role)
|
|
if (
|
|
not current.connected
|
|
or not current.visible
|
|
or current.disabled
|
|
or current.tag.lower() != candidate.metadata.tag.lower()
|
|
or current.input_type.lower() != candidate.metadata.input_type.lower()
|
|
or current.main_frame != candidate.metadata.main_frame
|
|
or current.contenteditable != candidate.metadata.contenteditable
|
|
or current_role != candidate.role.lower()
|
|
or action not in current_actions
|
|
or _safe_name(current_name) != candidate.name
|
|
):
|
|
raise WorkerError("candidate changed or is unavailable")
|
|
|
|
def _post_action_state(self, snapshot: ObservationRecord, same_document: bool = False) -> dict[str, Any]:
|
|
tab, frame_id, loader_id = self._state()
|
|
if tab["target_id"] != snapshot.target_id:
|
|
raise WorkerError("browser tab changed during action")
|
|
if same_document and (frame_id != snapshot.frame_id or loader_id != snapshot.loader_id):
|
|
raise WorkerError("document changed during fill")
|
|
return tab
|
|
|
|
def act(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
self._require_open()
|
|
self._mutation_dispatched = False
|
|
self._mutation_verified = False
|
|
if not isinstance(request, dict):
|
|
raise WorkerError("invalid action request")
|
|
action = request.get("action")
|
|
if action not in {"click", "fill"}:
|
|
raise WorkerError("unsupported action")
|
|
snapshot, candidate = self._find_snapshot(request)
|
|
# Popping the snapshot happens before any revalidation or mutation:
|
|
# attempted mutations can never be replayed with the same observation.
|
|
if action == "click":
|
|
if set(request) - {"observationId", "candidateId", "action", "postcondition"}:
|
|
raise WorkerError("malformed click request")
|
|
postcondition = request.get("postcondition")
|
|
if not isinstance(postcondition, dict) or set(postcondition) != {"kind", "value"}:
|
|
raise WorkerError("click requires one postcondition")
|
|
kind = postcondition.get("kind")
|
|
value = postcondition.get("value")
|
|
if kind not in {"url", "text"} or not isinstance(value, str) or not value or len(value) > MAX_STRING:
|
|
raise WorkerError("invalid click postcondition")
|
|
if kind == "url" and not _approved_origin(_origin(value), self.allowed_origins):
|
|
raise WorkerError("postcondition URL is outside the approved origins")
|
|
else:
|
|
if set(request) - {"observationId", "candidateId", "action", "text", "postcondition"}:
|
|
raise WorkerError("malformed fill request")
|
|
text = request.get("text")
|
|
if not isinstance(text, str) or len(text) > MAX_FILL_TEXT:
|
|
raise WorkerError("fill text exceeds protocol limit")
|
|
postcondition = request.get("postcondition")
|
|
if postcondition is not None:
|
|
if not isinstance(postcondition, dict) or set(postcondition) != {"kind", "value"}:
|
|
raise WorkerError("invalid fill postcondition")
|
|
if postcondition.get("kind") not in {"url", "text"} or not isinstance(postcondition.get("value"), str):
|
|
raise WorkerError("invalid fill postcondition")
|
|
if not postcondition["value"] or len(postcondition["value"]) > MAX_STRING:
|
|
raise WorkerError("invalid fill postcondition")
|
|
if postcondition["kind"] == "url" and not _approved_origin(_origin(postcondition["value"]), self.allowed_origins):
|
|
raise WorkerError("fill postcondition URL is outside the approved origins")
|
|
self._revalidate(snapshot, candidate, action)
|
|
if action == "fill":
|
|
self._mutation_dispatched = True
|
|
self.backend.fill(candidate.backend_id, text)
|
|
deadline = time.monotonic() + ACTION_TIMEOUT_SECONDS
|
|
while time.monotonic() < deadline:
|
|
tab = self._post_action_state(snapshot, same_document=True)
|
|
value_verified = self.backend.value_matches(candidate.backend_id, text)
|
|
condition_verified = True
|
|
if postcondition is not None:
|
|
if postcondition["kind"] == "url":
|
|
condition_verified = tab.get("url") == postcondition["value"]
|
|
else:
|
|
condition_verified = self.backend.visible_text(postcondition["value"])
|
|
if value_verified and condition_verified:
|
|
self._mutation_verified = True
|
|
return {"verified": True, "evidence": "field value and postcondition verified" if postcondition else "field value verified"}
|
|
time.sleep(0.025)
|
|
return {"verified": False, "evidence": "field value or postcondition could not be verified"}
|
|
|
|
# A postcondition that already holds cannot prove that this click did
|
|
# anything. Refuse it before the one and only mutation.
|
|
tab = self._post_action_state(snapshot)
|
|
if postcondition["kind"] == "url":
|
|
already_satisfied = tab["url"] == postcondition["value"]
|
|
else:
|
|
already_satisfied = _is_visible_text(postcondition["value"]) and self.backend.visible_text(postcondition["value"])
|
|
if already_satisfied:
|
|
raise WorkerError("click postcondition was already satisfied")
|
|
|
|
# Exactly one click is ever issued. Verification polling below never
|
|
# retries the mutation, and checks origin before examining each result.
|
|
self._mutation_dispatched = True
|
|
self.backend.click(candidate.backend_id)
|
|
deadline = time.monotonic() + ACTION_TIMEOUT_SECONDS
|
|
while time.monotonic() < deadline:
|
|
tab = self._post_action_state(snapshot)
|
|
if postcondition["kind"] == "url":
|
|
if tab["url"] == postcondition["value"]:
|
|
self._mutation_verified = True
|
|
return {"verified": True, "evidence": "URL postcondition verified"}
|
|
elif _is_visible_text(postcondition["value"]) and self.backend.visible_text(postcondition["value"]):
|
|
self._mutation_verified = True
|
|
return {"verified": True, "evidence": "visible-text postcondition verified"}
|
|
time.sleep(0.025)
|
|
# _post_action_state performs the final approved-origin check too.
|
|
self._post_action_state(snapshot)
|
|
return {"verified": False, "evidence": "click postcondition was not verified"}
|
|
|
|
def close(self) -> None:
|
|
if self._closed:
|
|
return
|
|
self._closed = True
|
|
self._observations.clear()
|
|
self.backend.close()
|
|
|
|
|
|
_METADATA_FUNCTION = """function () {
|
|
const e = this;
|
|
const style = getComputedStyle(e);
|
|
const type = (e.getAttribute('type') || '').toLowerCase();
|
|
const text = node => node ? String(node.innerText || node.textContent || '').trim() : '';
|
|
const labelledBy = (e.getAttribute('aria-labelledby') || '').split(/\\s+/)
|
|
.map(id => id ? text(document.getElementById(id)) : '').filter(Boolean).join(' ');
|
|
const labels = e.labels ? [...e.labels].map(text).filter(Boolean).join(' ') : '';
|
|
const visibleControlText = ['BUTTON', 'A', 'SUMMARY'].includes(e.tagName) ? text(e) : '';
|
|
// Never read .value: for fill candidates this is user data.
|
|
const label = e.getAttribute('aria-label') || labelledBy || labels ||
|
|
(visibleControlText || e.getAttribute('placeholder') ||
|
|
e.getAttribute('name') || e.getAttribute('id') || '');
|
|
return {
|
|
tag: e.tagName.toLowerCase(), type, role: e.getAttribute('role') || '',
|
|
name: String(label).slice(0, 240), connected: !!e.isConnected,
|
|
mainFrame: window.top === window,
|
|
contenteditable: e.isContentEditable,
|
|
visible: !!(e.getClientRects().length && style.display !== 'none' &&
|
|
style.visibility !== 'hidden' && style.opacity !== '0' &&
|
|
e.getAttribute('aria-hidden') !== 'true'),
|
|
disabled: !!(e.disabled || e.getAttribute('aria-disabled') === 'true' ||
|
|
e.matches(':disabled'))
|
|
};
|
|
}"""
|
|
_METADATA_MANY_FUNCTION = (
|
|
"function (...nodes) {\n"
|
|
f" const read = {_METADATA_FUNCTION};\n"
|
|
" return nodes.map(node => node && typeof node === 'object' ? read.call(node) : null);\n"
|
|
"}"
|
|
)
|
|
|
|
|
|
class RealBrowserBackend:
|
|
"""Adapter over browser-harness; no CDP implementation is duplicated here."""
|
|
|
|
def __init__(self, cdp_url: str, allowed_origins: OriginPolicy):
|
|
validate_cdp_url(cdp_url)
|
|
self.allowed_origins = normalize_allowed_origins(allowed_origins)
|
|
self._sandbox = Path(tempfile.mkdtemp(prefix="fast-browser-worker-"))
|
|
self._screenshot_dir = self._sandbox / "screenshots"
|
|
self._screenshot_dir.mkdir(mode=0o700)
|
|
try:
|
|
os.chmod(self._screenshot_dir, 0o700)
|
|
except OSError:
|
|
pass
|
|
self._daemon_ready = False
|
|
self._process: subprocess.Popen | None = None
|
|
self._attached_target: str | None = None
|
|
self._select_all_modifier: int | None = None
|
|
self._configure_environment(cdp_url, self._sandbox)
|
|
try:
|
|
self._load_harness()
|
|
self._start_daemon()
|
|
self._attach_initial_tab()
|
|
except BaseException:
|
|
self.close()
|
|
raise
|
|
|
|
def _start_daemon(self) -> None:
|
|
# Own the actual child handle. admin.ensure_daemon detaches/retries its
|
|
# child and can launch Chrome; neither behavior belongs in this adapter.
|
|
self._process = subprocess.Popen(
|
|
[sys.executable, "-m", "browser_harness.daemon"],
|
|
env=dict(os.environ), stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
start_new_session=True,
|
|
)
|
|
deadline = time.monotonic() + 5.0
|
|
while time.monotonic() < deadline:
|
|
if self._process.poll() is not None:
|
|
raise WorkerError("browser-harness daemon failed to connect to the configured endpoint")
|
|
try:
|
|
self._helpers.cdp("Target.getTargets", _response_timeout=0.2)
|
|
self._daemon_ready = True
|
|
return
|
|
except Exception:
|
|
time.sleep(0.025)
|
|
raise WorkerError("configured browser endpoint was not ready within five seconds")
|
|
|
|
@staticmethod
|
|
def _configure_environment(cdp_url: str, root: Path) -> None:
|
|
# The directory is retained by the instance through the environment;
|
|
# setting it before importing browser-harness prevents its .env and
|
|
# agent-helper discovery from reaching the user's workspace.
|
|
os.environ.update({
|
|
"BU_CDP_URL": cdp_url,
|
|
"BU_NAME": f"fast-{uuid.uuid4().hex[:20]}",
|
|
"BH_HOME": str(root / "home"),
|
|
"BH_CONFIG_DIR": str(root / "config"),
|
|
"BH_RUNTIME_DIR": str(root / "runtime"),
|
|
"BH_RUNTIME_DIR_SHARED": "0",
|
|
"BH_TMP_DIR": str(root / "tmp"),
|
|
"BH_TMP_DIR_SHARED": "0",
|
|
"BH_AGENT_WORKSPACE": str(root / "workspace"),
|
|
"BH_TELEMETRY": "0",
|
|
"BH_UPDATE_CHECK": "0",
|
|
"BH_RECORD": "0",
|
|
"BH_TAB_MARKER": "0",
|
|
"PYTHONNOUSERSITE": "1",
|
|
# Empty, rather than absent: browser-harness's .env loader uses
|
|
# setdefault and must not reintroduce cloud credentials or spawn.
|
|
"BROWSER_USE_API_KEY": "",
|
|
"BU_CDP_WS": "",
|
|
"BU_AUTOSPAWN": "",
|
|
"BU_BROWSER_ID": "",
|
|
"BU_PROFILE_ID": "",
|
|
"BU_PROXY_URL": "",
|
|
"BU_PROXY_USERNAME": "",
|
|
"BU_PROXY_PASSWORD": "",
|
|
})
|
|
|
|
def _load_harness(self) -> None:
|
|
# browser-harness must come from the configured Python environment's
|
|
# installed package. In particular, never import a temporary checkout
|
|
# via sys.path: that would make the runtime transport supply-chain
|
|
# mutable and would bypass the pinned environment chosen by the caller.
|
|
from importlib.metadata import PackageNotFoundError, version
|
|
try:
|
|
installed = version("browser-harness")
|
|
except PackageNotFoundError:
|
|
raise WorkerError(
|
|
"browser-harness is not installed in this Python environment; "
|
|
"install the pinned 0.1.13 package or set FAST_BROWSER_PYTHON"
|
|
) from None
|
|
if installed != EXPECTED_BROWSER_HARNESS_VERSION:
|
|
raise WorkerError(
|
|
f"browser-harness {EXPECTED_BROWSER_HARNESS_VERSION} is required, found {installed}; "
|
|
"use a dedicated pinned Python environment (no auto-upgrade is attempted)"
|
|
)
|
|
from browser_harness import helpers # type: ignore
|
|
self._helpers = helpers
|
|
|
|
def _attach_initial_tab(self) -> None:
|
|
"""Attach one eligible target when unambiguous, otherwise stay idle.
|
|
|
|
Runtime startup must remain usable with multiple tabs so callers can
|
|
list and explicitly select a target. A zero-tab startup still fails
|
|
closed because no operation can be served until a target exists.
|
|
"""
|
|
tabs = self._eligible_tabs()
|
|
if not tabs:
|
|
raise WorkerError("no approved-origin browser tab is available")
|
|
if len(tabs) > 1:
|
|
self._attached_target = None
|
|
return
|
|
target = tabs[0].get("targetId") or tabs[0].get("target_id")
|
|
if not isinstance(target, str) or not target:
|
|
raise WorkerError("approved browser tab has no target identity")
|
|
self.switch_tab(target)
|
|
|
|
def _eligible_tabs(self) -> list[dict[str, Any]]:
|
|
tabs = self._helpers.list_tabs(include_chrome=False)
|
|
return [
|
|
tab for tab in tabs
|
|
if isinstance(tab, dict)
|
|
and isinstance(tab.get("url"), str)
|
|
and _approved_origin(_origin(tab["url"]), self.allowed_origins)
|
|
and isinstance(tab.get("targetId") or tab.get("target_id"), str)
|
|
]
|
|
|
|
def _attach_approved_tab(self) -> None:
|
|
# A named browser-harness daemon owns a blank dedicated tab. It is
|
|
# intentionally not navigated: attach only to the one already-open
|
|
# approved tab and leave the user's selected tab/visibility untouched.
|
|
tabs = self._helpers.list_tabs(include_chrome=False)
|
|
approved = [
|
|
tab for tab in tabs
|
|
if isinstance(tab, dict)
|
|
and isinstance(tab.get("url"), str)
|
|
and _approved_origin(_origin(tab["url"]), self.allowed_origins)
|
|
]
|
|
if len(approved) != 1:
|
|
# Diagnose redirects without exposing titles, paths, query tokens,
|
|
# field values, or any content from outside the allowed origins.
|
|
web_origins = sorted({
|
|
origin for tab in tabs if isinstance(tab, dict)
|
|
and isinstance(tab.get("url"), str)
|
|
if (origin := _origin(tab["url"])) is not None
|
|
})
|
|
summary = json.dumps([origin[:240] for origin in web_origins[:10]])
|
|
raise WorkerError(
|
|
"expected exactly one existing approved-origin browser tab; "
|
|
f"found {len(approved)}; open web origins: {summary}"
|
|
)
|
|
target = approved[0].get("targetId") or approved[0].get("target_id")
|
|
if not isinstance(target, str) or not target:
|
|
raise WorkerError("approved browser tab has no target identity")
|
|
self._helpers.switch_tab(target, activate=False)
|
|
|
|
def switch_tab(self, target_id: str) -> None:
|
|
if not isinstance(target_id, str) or not target_id:
|
|
raise WorkerError("tab identity is unavailable")
|
|
# browser-harness 0.1.13 enables Page/DOM/Runtime/Network as part of
|
|
# its switch_tab set_session path. Re-enabling them here duplicates
|
|
# four CDP calls and can race the daemon's session setup.
|
|
self._session_id = self._helpers.switch_tab(target_id, activate=False)
|
|
self._attached_target = target_id
|
|
|
|
def list_tabs(self) -> list[dict[str, Any]]:
|
|
# helpers.list_tabs() and this opener lookup used to issue two identical
|
|
# Target.getTargets requests. Keep its page/internal filtering locally
|
|
# while deriving opener metadata from the same target inventory.
|
|
raw = self._helpers.cdp("Target.getTargets")
|
|
target_infos = raw.get("targetInfos", []) if isinstance(raw, dict) else []
|
|
if not isinstance(target_infos, list):
|
|
return []
|
|
result: list[dict[str, Any]] = []
|
|
for info in target_infos:
|
|
if not isinstance(info, dict) or info.get("type") != "page":
|
|
continue
|
|
url = info.get("url", "")
|
|
title = info.get("title", "")
|
|
if not isinstance(url, str) or not isinstance(title, str):
|
|
continue
|
|
# Match browser-harness helpers.list_tabs(include_chrome=False),
|
|
# including its startup-placeholder suppression.
|
|
if title.startswith("Starting agent ") and (
|
|
url in ("", "about:blank") or url.startswith("about:blank#")
|
|
):
|
|
continue
|
|
if url.startswith(_INTERNAL_TARGET_PREFIXES):
|
|
continue
|
|
target = info.get("targetId")
|
|
if not isinstance(target, str) or not target:
|
|
continue
|
|
opener = info.get("openerId")
|
|
result.append({
|
|
"targetId": target,
|
|
"target_id": target,
|
|
"url": url,
|
|
"title": title,
|
|
"openerTargetId": opener if isinstance(opener, str) and opener else None,
|
|
})
|
|
return result
|
|
|
|
def navigate(self, url: str) -> dict[str, Any]:
|
|
result = self._helpers.goto_url(url)
|
|
if isinstance(result, dict):
|
|
return result
|
|
return {}
|
|
|
|
def reload(self) -> dict[str, Any]:
|
|
return self._helpers.cdp("Page.reload")
|
|
|
|
def wait_ready(self, timeout: float = NAVIGATION_TIMEOUT_SECONDS) -> bool:
|
|
# Keep the harness' exact readyState query, but avoid its coarse 300ms
|
|
# sleep. This is only a readiness read; it never dispatches navigation.
|
|
if timeout <= 0:
|
|
return False
|
|
deadline = time.monotonic() + timeout
|
|
while True:
|
|
if self._helpers.js("document.readyState") == "complete":
|
|
return True
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
return False
|
|
time.sleep(min(0.025, remaining))
|
|
|
|
def visible_text_content(self) -> str:
|
|
# This is a fixed, worker-owned read operation. No caller expression,
|
|
# selector, HTML, form value, frame, or script output is accepted.
|
|
expression = """(() => {
|
|
const root = document.body;
|
|
if (!root) return '';
|
|
const blocked = new Set(['SCRIPT','STYLE','NOSCRIPT','TEMPLATE',
|
|
'INPUT','TEXTAREA','SELECT','OPTION','DATALIST','OUTPUT']);
|
|
const visible = node => {
|
|
for (let e = node.nodeType === Node.ELEMENT_NODE ? node : node.parentElement;
|
|
e && e !== root.parentElement; e = e.parentElement) {
|
|
const s = getComputedStyle(e);
|
|
if (e.getAttribute('aria-hidden') === 'true' ||
|
|
s.display === 'none' || s.visibility === 'hidden' ||
|
|
s.opacity === '0') return false;
|
|
}
|
|
return true;
|
|
};
|
|
const walker = document.createTreeWalker(root, NodeFilter.SHOW_TEXT);
|
|
const parts = [];
|
|
let size = 0;
|
|
let node;
|
|
while ((node = walker.nextNode())) {
|
|
const parent = node.parentElement;
|
|
if (!parent || blocked.has(parent.tagName) || !visible(parent)) continue;
|
|
const text = String(node.nodeValue || '').replace(/[\\t\\n\\r ]+/g, ' ').trim();
|
|
if (text) {
|
|
parts.push(text);
|
|
size += text.length + 1;
|
|
if (size > 16001) break;
|
|
}
|
|
}
|
|
return parts.join(' ').replace(/ +/g, ' ').trim().slice(0, 16001);
|
|
})()"""
|
|
result = self._helpers.cdp("Runtime.evaluate", expression=expression,
|
|
returnByValue=True, awaitPromise=False)
|
|
remote = result.get("result") or {}
|
|
value = remote.get("value")
|
|
if not isinstance(value, str):
|
|
raise WorkerError("visible page text is unavailable")
|
|
return value
|
|
|
|
def drain_events(self) -> list[dict[str, Any]]:
|
|
try:
|
|
events = self._helpers.drain_events()
|
|
except Exception:
|
|
return []
|
|
return events if isinstance(events, list) else []
|
|
|
|
def screenshot_bytes(self) -> bytes:
|
|
# NamedTemporaryFile creates the file with O_EXCL inside the daemon's
|
|
# private 0700 sandbox; the helper only receives this worker-owned path.
|
|
handle = tempfile.NamedTemporaryFile(
|
|
prefix="shot-", suffix=".png", dir=self._screenshot_dir,
|
|
mode="w+b", delete=False,
|
|
)
|
|
path = Path(handle.name)
|
|
handle.close()
|
|
try:
|
|
os.chmod(path, 0o600)
|
|
self._helpers.capture_screenshot(str(path), full=False, max_dim=2000)
|
|
with open(path, "rb") as stream:
|
|
return stream.read(MAX_SCREENSHOT_BYTES + 1)
|
|
finally:
|
|
try:
|
|
path.unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
def close_tab(self, target_id: str) -> None:
|
|
self._helpers.close_tab(target_id)
|
|
|
|
def current_tab(self) -> dict[str, Any]:
|
|
return self._helpers.current_tab()
|
|
|
|
def document_identity(self) -> tuple[str, str]:
|
|
tree = self._helpers.cdp("Page.getFrameTree")
|
|
frame = (tree.get("frameTree") or {}).get("frame") or {}
|
|
frame_id = frame.get("id")
|
|
loader_id = frame.get("loaderId") or "no-loader"
|
|
if not isinstance(frame_id, str):
|
|
raise WorkerError("main frame is unavailable")
|
|
return frame_id, str(loader_id)
|
|
|
|
def accessibility_nodes(self) -> Iterable[dict[str, Any]]:
|
|
tree = self._helpers.cdp("Accessibility.getFullAXTree", depth=10)
|
|
nodes = tree.get("nodes")
|
|
return nodes if isinstance(nodes, list) else ()
|
|
|
|
def _call(self, backend_id: int, function: str, arguments: list[dict[str, Any]] | None = None) -> Any:
|
|
resolved = self._helpers.cdp("DOM.resolveNode", backendNodeId=backend_id)
|
|
object_id = (resolved.get("object") or {}).get("objectId")
|
|
if not isinstance(object_id, str):
|
|
raise WorkerError("element is unavailable")
|
|
try:
|
|
result = self._helpers.cdp(
|
|
"Runtime.callFunctionOn", objectId=object_id, functionDeclaration=function,
|
|
arguments=arguments or [], returnByValue=True, awaitPromise=False,
|
|
)
|
|
details = result.get("exceptionDetails")
|
|
remote = result.get("result") or {}
|
|
if details or remote.get("subtype") == "error" or "value" not in remote:
|
|
raise WorkerError("browser evaluation failed")
|
|
value = remote.get("value")
|
|
if not isinstance(value, (dict, bool, str, int, float, type(None))):
|
|
raise WorkerError("invalid browser metadata")
|
|
return value
|
|
finally:
|
|
try:
|
|
self._helpers.cdp("Runtime.releaseObject", objectId=object_id)
|
|
except Exception:
|
|
pass
|
|
|
|
@staticmethod
|
|
def _metadata_from_dom_value(backend_id: int, value: Any) -> ElementMetadata:
|
|
if not isinstance(value, dict):
|
|
raise WorkerError("element metadata unavailable")
|
|
return ElementMetadata(
|
|
backend_id, str(value.get("tag", "")), str(value.get("type", "")),
|
|
str(value.get("role", "")), " ".join(str(value.get("name", "")).split())[:MAX_NAME],
|
|
bool(value.get("visible")), bool(value.get("disabled")), bool(value.get("connected")),
|
|
bool(value.get("mainFrame", True)), bool(value.get("contenteditable", False)),
|
|
)
|
|
|
|
def metadata(self, backend_id: int) -> ElementMetadata:
|
|
value = self._call(backend_id, _METADATA_FUNCTION)
|
|
metadata = self._metadata_from_dom_value(backend_id, value)
|
|
ax_role, ax_name = "", ""
|
|
try:
|
|
partial = self._helpers.cdp(
|
|
"Accessibility.getPartialAXTree", backendNodeId=backend_id, fetchRelatives=False
|
|
)
|
|
for node in partial.get("nodes", []) if isinstance(partial, dict) else []:
|
|
if not isinstance(node, dict):
|
|
continue
|
|
ax_role = _ax_value(node.get("role", "")).lower()
|
|
ax_name = _ax_value(node.get("name", ""))
|
|
if ax_role or ax_name:
|
|
break
|
|
except Exception:
|
|
# DOM metadata still provides safe labels; accessibility is a
|
|
# refinement and not a reason to expose a broad CDP surface.
|
|
pass
|
|
return replace(metadata, ax_role=ax_role, ax_name=ax_name)
|
|
|
|
def metadata_many(self, backend_ids: Iterable[int]) -> dict[int, ElementMetadata]:
|
|
"""Read value-free DOM metadata for one bounded observation chunk.
|
|
|
|
The full AX tree is already the observation's source for role/name. This
|
|
seam only batches the DOM extraction; action revalidation continues to
|
|
use metadata(), including its fresh partial AX read.
|
|
"""
|
|
ids = [backend_id for backend_id in backend_ids if type(backend_id) is int and backend_id > 0][:MAX_METADATA_BATCH]
|
|
if not ids:
|
|
return {}
|
|
object_group = f"fast-browser-metadata-{uuid.uuid4().hex}"
|
|
object_ids: list[str | None] = []
|
|
try:
|
|
for backend_id in ids:
|
|
try:
|
|
resolved = self._helpers.cdp(
|
|
"DOM.resolveNode", backendNodeId=backend_id, objectGroup=object_group
|
|
)
|
|
except Exception:
|
|
object_ids.append(None)
|
|
continue
|
|
object_id = (resolved.get("object") or {}).get("objectId") if isinstance(resolved, dict) else None
|
|
object_ids.append(object_id if isinstance(object_id, str) else None)
|
|
|
|
target_object = next((object_id for object_id in object_ids if object_id is not None), None)
|
|
if target_object is None:
|
|
return {}
|
|
arguments = [
|
|
{"objectId": object_id} if object_id is not None else {"value": None}
|
|
for object_id in object_ids
|
|
]
|
|
try:
|
|
response = self._helpers.cdp(
|
|
"Runtime.callFunctionOn", objectId=target_object,
|
|
functionDeclaration=_METADATA_MANY_FUNCTION, arguments=arguments,
|
|
returnByValue=True, awaitPromise=False, objectGroup=object_group,
|
|
)
|
|
except Exception:
|
|
return {}
|
|
details = response.get("exceptionDetails") if isinstance(response, dict) else None
|
|
remote = response.get("result") if isinstance(response, dict) else None
|
|
values = remote.get("value") if isinstance(remote, dict) else None
|
|
if details or not isinstance(remote, dict) or remote.get("subtype") == "error" or not isinstance(values, list):
|
|
return {}
|
|
result: dict[int, ElementMetadata] = {}
|
|
for backend_id, value in zip(ids, values):
|
|
try:
|
|
result[backend_id] = self._metadata_from_dom_value(backend_id, value)
|
|
except WorkerError:
|
|
# Missing or malformed entries use the safe legacy path in
|
|
# BrowserWorker.observe rather than being treated as absent.
|
|
continue
|
|
return result
|
|
finally:
|
|
try:
|
|
self._helpers.cdp("Runtime.releaseObjectGroup", objectGroup=object_group)
|
|
except Exception:
|
|
pass
|
|
|
|
def click(self, backend_id: int) -> None:
|
|
box = self._helpers.cdp("DOM.getBoxModel", backendNodeId=backend_id)
|
|
content = (box.get("model") or {}).get("content")
|
|
if not isinstance(content, list) or len(content) < 8:
|
|
raise WorkerError("candidate has no clickable box")
|
|
xs = [float(content[i]) for i in range(0, len(content), 2)]
|
|
ys = [float(content[i]) for i in range(1, len(content), 2)]
|
|
x, y = sum(xs) / len(xs), sum(ys) / len(ys)
|
|
hit = self._call(backend_id, """function (x, y) {
|
|
const hit = document.elementFromPoint(x, y);
|
|
return !!hit && (hit === this || this.contains(hit));
|
|
}""", [{"value": x}, {"value": y}])
|
|
if hit is not True:
|
|
raise WorkerError("candidate is covered or no longer at its observed position")
|
|
self._helpers.cdp("Input.dispatchMouseEvent", type="mousePressed", x=x, y=y, button="left", clickCount=1)
|
|
self._helpers.cdp("Input.dispatchMouseEvent", type="mouseReleased", x=x, y=y, button="left", clickCount=1)
|
|
|
|
def fill(self, backend_id: int, text: str) -> None:
|
|
self._helpers.cdp("DOM.focus", backendNodeId=backend_id)
|
|
focused = self._call(backend_id, """function () {
|
|
return document.activeElement === this;
|
|
}""")
|
|
if focused is not True:
|
|
raise WorkerError("candidate could not be focused")
|
|
# Select all without an arbitrary page script. Input.insertText is a
|
|
# single bounded browser operation and is substantially faster than
|
|
# dispatching up to 8,000 individual key events. Browser.getVersion is
|
|
# stable for this backend, so cache the platform modifier for its life.
|
|
if getattr(self, "_select_all_modifier", None) is None:
|
|
user_agent = self._helpers.cdp("Browser.getVersion").get("userAgent", "")
|
|
self._select_all_modifier = 4 if "Mac OS X" in user_agent or "Macintosh" in user_agent else 2
|
|
modifiers = self._select_all_modifier
|
|
select_all = {"key": "a", "code": "KeyA", "modifiers": modifiers,
|
|
"windowsVirtualKeyCode": 65, "nativeVirtualKeyCode": 65,
|
|
"commands": ["SelectAll"]}
|
|
self._helpers.cdp("Input.dispatchKeyEvent", type="rawKeyDown", **select_all)
|
|
self._helpers.cdp("Input.dispatchKeyEvent", type="keyUp", **{k: v for k, v in select_all.items() if k != "commands"})
|
|
self._helpers.press_key("Backspace")
|
|
if text:
|
|
self._helpers.cdp("Input.insertText", text=text)
|
|
self._call(backend_id, """function () {
|
|
this.dispatchEvent(new Event('input', {bubbles: true}));
|
|
this.dispatchEvent(new Event('change', {bubbles: true}));
|
|
return true;
|
|
}""")
|
|
|
|
def value_matches(self, backend_id: int, text: str) -> bool:
|
|
value = self._call(backend_id, """function () {
|
|
return 'value' in this ? String(this.value) : String(this.textContent || '');
|
|
}""")
|
|
return isinstance(value, str) and value == text
|
|
|
|
def visible_text(self, text: str) -> bool:
|
|
# The literal is JSON-encoded, never concatenated as executable code.
|
|
result = self._helpers.cdp(
|
|
"Runtime.evaluate", expression="""(() => {
|
|
const wanted = __TEXT__;
|
|
const visible = e => {
|
|
const s = getComputedStyle(e);
|
|
return !!(e.getClientRects().length && s.display !== 'none' &&
|
|
s.visibility !== 'hidden' && s.opacity !== '0');
|
|
};
|
|
return [...document.querySelectorAll('body *')].some(e =>
|
|
visible(e) && (e.innerText || '').includes(wanted));
|
|
})()""".replace("__TEXT__", json.dumps(text)), returnByValue=True,
|
|
)
|
|
return (result.get("result") or {}).get("value") is True
|
|
|
|
def close(self) -> None:
|
|
if getattr(self, "_closed", False):
|
|
return
|
|
self._closed = True
|
|
if getattr(self, "_daemon_ready", False):
|
|
try:
|
|
# This is the browser-harness daemon's ownership-aware
|
|
# shutdown; it closes only this named daemon's dedicated tab.
|
|
self._helpers._send({"meta": "shutdown"}, response_timeout=0.3)
|
|
except Exception:
|
|
pass
|
|
self._daemon_ready = False
|
|
process = getattr(self, "_process", None)
|
|
if process is not None:
|
|
try:
|
|
process.wait(timeout=0.3)
|
|
except subprocess.TimeoutExpired:
|
|
# asyncio's SIGINT handling runs the harness ownership cleanup.
|
|
try:
|
|
process.send_signal(signal.SIGINT)
|
|
except ProcessLookupError:
|
|
pass
|
|
try:
|
|
process.wait(timeout=0.3)
|
|
except subprocess.TimeoutExpired:
|
|
try:
|
|
process.kill()
|
|
except ProcessLookupError:
|
|
pass
|
|
try:
|
|
process.wait(timeout=0.3)
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
self._cleanup_sandbox()
|
|
|
|
def _cleanup_sandbox(self) -> None:
|
|
import shutil
|
|
shutil.rmtree(self._sandbox, ignore_errors=True)
|
|
|
|
|
|
@dataclass
|
|
class _RuntimeTab:
|
|
tab_id: str
|
|
target_id: str
|
|
kind: str
|
|
opener_target_id: str | None
|
|
url: str
|
|
title: str
|
|
active: bool = True
|
|
|
|
|
|
class BrowserRuntimeWorker(BrowserWorker):
|
|
"""Persistent session runtime for the published M2/M3/M4 operations."""
|
|
|
|
def __init__(self, backend: BrowserBackend, allowed_origins: OriginPolicy,
|
|
*, session_id: str | None = None):
|
|
super().__init__(backend, allowed_origins)
|
|
self.session_id = _safe_component(session_id or f"session-{uuid.uuid4().hex}", "session id")
|
|
self._base_policy = self.allowed_origins
|
|
self._tabs: dict[str, _RuntimeTab] = {}
|
|
self._target_to_tab: dict[str, str] = {}
|
|
self._selected_tab_id: str | None = None
|
|
self._tab_counter = 0
|
|
self._event_buffer: list[dict[str, Any]] = []
|
|
self._popup_candidates: set[str] = set()
|
|
self._claimed_popups: set[str] = set()
|
|
self._initializing = True
|
|
self._closed_runtime = False
|
|
self._refresh_tabs()
|
|
# Existing popup/page targets are baseline state, not newly-owned
|
|
# popup candidates for a later waitPopup call.
|
|
self._popup_candidates.clear()
|
|
self._initializing = False
|
|
# Exactly one initial target is safe to select. With multiple tabs no
|
|
# arbitrary page is selected; listTabs remains available.
|
|
if len(self._tabs) == 1:
|
|
self._selected_tab_id = next(iter(self._tabs))
|
|
self._attach(self._tabs[self._selected_tab_id].target_id, "selectTab")
|
|
|
|
def _fail(self, operation: str, code: str, phase: str, message: str,
|
|
*, mutation_state: str = "not_attempted", retry: str = "never") -> BrowserOperationError:
|
|
return _operation_error(operation, code, phase, message,
|
|
mutation_state=mutation_state, retry=retry)
|
|
|
|
def _backend_tabs(self) -> list[dict[str, Any]]:
|
|
listing = getattr(self.backend, "list_tabs", None)
|
|
if callable(listing):
|
|
try:
|
|
value = listing()
|
|
except Exception:
|
|
raise self._fail("listTabs", "worker_failure", "list", "browser tabs are unavailable")
|
|
return value if isinstance(value, list) else []
|
|
try:
|
|
current = self.backend.current_tab()
|
|
except Exception:
|
|
return []
|
|
return [current] if isinstance(current, dict) else []
|
|
|
|
def _refresh_tabs(self) -> None:
|
|
seen: set[str] = set()
|
|
raw_tabs = self._backend_tabs()
|
|
# Build the eligible target index before classifying any row. CDP target
|
|
# ordering is not an ownership signal: a popup may precede its opener.
|
|
entries: list[tuple[dict[str, Any], str, str]] = []
|
|
target_index: set[str] = set()
|
|
for raw in raw_tabs:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
target = raw.get("targetId") or raw.get("target_id")
|
|
url = raw.get("url")
|
|
if not isinstance(target, str) or not target or not isinstance(url, str):
|
|
continue
|
|
origin = _origin(url)
|
|
# Discovery is bounded by the immutable session base. A per-call
|
|
# context only filters/authorizes an operation; it must not make a
|
|
# tab handle disappear when the next call restores the base.
|
|
if not _approved_origin(origin, self._base_policy):
|
|
continue
|
|
entries.append((raw, target, url))
|
|
target_index.add(target)
|
|
for raw, target, url in entries:
|
|
seen.add(target)
|
|
tab_id = self._target_to_tab.get(target)
|
|
opener = raw.get("openerTargetId") or raw.get("openerId") or raw.get("opener_target_id")
|
|
opener_target = opener if isinstance(opener, str) and opener else None
|
|
owned_opener = opener_target if opener_target in target_index or opener_target in self._target_to_tab else None
|
|
if tab_id is None:
|
|
self._tab_counter += 1
|
|
tab_id = f"tab-{self._tab_counter}-{secrets.token_hex(6)}"
|
|
self._target_to_tab[target] = tab_id
|
|
kind = "popup" if owned_opener else "page"
|
|
self._tabs[tab_id] = _RuntimeTab(tab_id, target, kind, owned_opener,
|
|
url, _short(raw.get("title") or ""))
|
|
if kind == "popup" and not self._initializing:
|
|
self._popup_candidates.add(target)
|
|
else:
|
|
tab = self._tabs[tab_id]
|
|
tab.url = url
|
|
tab.title = _short(raw.get("title") or "")
|
|
if owned_opener:
|
|
was_popup = tab.kind == "popup"
|
|
tab.opener_target_id = owned_opener
|
|
tab.kind = "popup"
|
|
if not was_popup and not self._initializing:
|
|
self._popup_candidates.add(target)
|
|
# Handles to closed tabs are invalidated. Do not invent a replacement
|
|
# selection when a selected tab disappears.
|
|
for target, tab_id in list(self._target_to_tab.items()):
|
|
if target not in seen:
|
|
self._target_to_tab.pop(target, None)
|
|
self._tabs.pop(tab_id, None)
|
|
self._popup_candidates.discard(target)
|
|
self._claimed_popups.discard(target)
|
|
if self._selected_tab_id == tab_id:
|
|
self._selected_tab_id = None
|
|
|
|
def _attach(self, target_id: str, operation: str = "selectTab") -> None:
|
|
attached = getattr(self.backend, "_attached_target", None)
|
|
if attached is not None and attached == target_id:
|
|
return
|
|
switch = getattr(self.backend, "switch_tab", None)
|
|
if callable(switch):
|
|
try:
|
|
switch(target_id)
|
|
except Exception:
|
|
raise self._fail(operation, "no_tab", "attach", "tab is unavailable")
|
|
return
|
|
try:
|
|
current = self.backend.current_tab()
|
|
except Exception:
|
|
raise self._fail(operation, "no_tab", "attach", "tab is unavailable")
|
|
if current.get("targetId") != target_id and current.get("target_id") != target_id:
|
|
raise self._fail(operation, "no_tab", "attach", "tab is unavailable")
|
|
|
|
def _resolve_tab(self, request: dict[str, Any] | None, operation: str) -> _RuntimeTab:
|
|
self._refresh_tabs()
|
|
tab_id = request.get("tabId") if isinstance(request, dict) else None
|
|
if tab_id is not None:
|
|
if not isinstance(tab_id, str) or len(tab_id) > 200:
|
|
raise self._fail(operation, "invalid_request", "target", "tab handle is invalid")
|
|
tab = self._tabs.get(tab_id)
|
|
if tab is None:
|
|
raise self._fail(operation, "no_tab", "target", "tab handle is stale or unavailable")
|
|
else:
|
|
if self._selected_tab_id is None:
|
|
if not self._tabs:
|
|
raise self._fail(operation, "no_tab", "target", "no approved tab is available")
|
|
raise self._fail(operation, "ambiguous_tab", "target", "select an approved tab explicitly")
|
|
tab = self._tabs.get(self._selected_tab_id)
|
|
if tab is None:
|
|
self._selected_tab_id = None
|
|
raise self._fail(operation, "no_tab", "target", "selected tab is unavailable")
|
|
self._attach(tab.target_id, operation)
|
|
# The inherited worker uses the current backend target and current
|
|
# policy. Verify it once more without returning raw browser details.
|
|
try:
|
|
current = self.backend.current_tab()
|
|
url = current.get("url") if isinstance(current, dict) else None
|
|
except Exception:
|
|
raise self._fail(operation, "no_tab", "target", "tab is unavailable")
|
|
if not isinstance(url, str) or not _approved_origin(_origin(url), self.allowed_origins):
|
|
raise self._fail(operation, "policy_denied", "policy", "tab is outside the approved origins")
|
|
return tab
|
|
|
|
def _document_id(self, operation: str) -> str:
|
|
try:
|
|
frame, loader = self.backend.document_identity()
|
|
except Exception:
|
|
raise self._fail(operation, "worker_failure", "document", "document identity is unavailable")
|
|
if not isinstance(frame, str) or not isinstance(loader, str):
|
|
raise self._fail(operation, "worker_failure", "document", "document identity is unavailable")
|
|
return f"{frame}:{loader}"[:MAX_STRING]
|
|
|
|
def _public_tab(self, tab: _RuntimeTab) -> dict[str, Any]:
|
|
opener_tab = self._target_to_tab.get(tab.opener_target_id or "")
|
|
return {
|
|
"tabId": tab.tab_id,
|
|
"kind": "popup" if tab.kind == "popup" else "page",
|
|
**({"openerTabId": opener_tab} if opener_tab else {}),
|
|
"selected": tab.tab_id == self._selected_tab_id,
|
|
"owned": True,
|
|
"url": tab.url,
|
|
"origin": _origin(tab.url) or "",
|
|
"title": tab.title,
|
|
}
|
|
|
|
def _collect_events(self) -> None:
|
|
drain = getattr(self.backend, "drain_events", None)
|
|
if not callable(drain):
|
|
return
|
|
try:
|
|
events = drain()
|
|
except Exception:
|
|
return
|
|
if isinstance(events, list):
|
|
active_session = getattr(self.backend, "_session_id", None)
|
|
supported = {"Runtime.consoleAPICalled", "Network.requestWillBeSent",
|
|
"Network.loadingFailed", "Network.responseReceived"}
|
|
for event in events[:MAX_EVENTS]:
|
|
if not isinstance(event, dict) or event.get("method") not in supported:
|
|
continue
|
|
event_session = event.get("session_id")
|
|
# The pinned daemon includes the CDP session on every tapped
|
|
# event. Unattributed events are fail-closed: retaining them
|
|
# could mix a prior tab's private diagnostics into this target.
|
|
if not active_session or not isinstance(event_session, str) or event_session != active_session:
|
|
continue
|
|
method = event.get("method")
|
|
params = event.get("params") if isinstance(event.get("params"), dict) else {}
|
|
safe_params: dict[str, Any] = {}
|
|
if method == "Runtime.consoleAPICalled":
|
|
level = params.get("type")
|
|
if level not in {"error", "warning", "warn"}:
|
|
continue
|
|
safe_args: list[dict[str, str]] = []
|
|
args = params.get("args")
|
|
if isinstance(args, list):
|
|
for arg in args[:8]:
|
|
if not isinstance(arg, dict):
|
|
continue
|
|
value = arg.get("value")
|
|
description = arg.get("description")
|
|
if isinstance(value, (str, int, float, bool)):
|
|
safe_args.append({"value": _safe_event_message(str(value))})
|
|
elif isinstance(description, str):
|
|
safe_args.append({"value": _safe_event_message(description)})
|
|
safe_params = {"type": level, "args": safe_args}
|
|
elif method == "Network.requestWillBeSent":
|
|
request_id = params.get("requestId")
|
|
request = params.get("request") if isinstance(params.get("request"), dict) else {}
|
|
url = request.get("url")
|
|
origin = _origin(url) if isinstance(url, str) else None
|
|
if not isinstance(request_id, str) or not origin:
|
|
continue
|
|
safe_params = {"requestId": request_id,
|
|
"request": {"url": origin, "method": str(request.get("method") or "GET")[:32]}}
|
|
elif method == "Network.loadingFailed":
|
|
url = params.get("url")
|
|
origin = _origin(url) if isinstance(url, str) else None
|
|
safe_params = {"requestId": str(params.get("requestId") or "")[:200],
|
|
"url": origin or "", "blockedReason": bool(params.get("blockedReason"))}
|
|
else:
|
|
response = params.get("response") if isinstance(params.get("response"), dict) else {}
|
|
safe_params = {"requestId": str(params.get("requestId") or "")[:200],
|
|
"response": {"status": response.get("status")}}
|
|
# Preserve provenance with the sanitized event. The daemon's
|
|
# drain buffer is persistent across target switches; without
|
|
# retaining this session ID, evidence on a later tab could
|
|
# replay private diagnostics collected from an earlier tab.
|
|
self._event_buffer.append({
|
|
"method": method, "params": safe_params,
|
|
"session_id": active_session,
|
|
})
|
|
if len(self._event_buffer) > MAX_EVENTS:
|
|
del self._event_buffer[:-MAX_EVENTS]
|
|
|
|
def _read_text(self) -> tuple[str, bool]:
|
|
getter = getattr(self.backend, "visible_text_content", None)
|
|
if not callable(getter):
|
|
# Offline backends may expose the safe content under this explicit
|
|
# seam; never fall back to a caller-provided predicate or script.
|
|
getter = getattr(self.backend, "read_visible_text", None)
|
|
if not callable(getter):
|
|
raise self._fail("read", "unsupported", "visibleText", "visible page text is unavailable")
|
|
try:
|
|
value = getter()
|
|
except BrowserOperationError:
|
|
raise
|
|
except Exception:
|
|
raise self._fail("read", "worker_failure", "visibleText", "visible page text is unavailable")
|
|
if not isinstance(value, str):
|
|
raise self._fail("read", "worker_failure", "visibleText", "visible page text is unavailable")
|
|
# Remove control characters that can forge protocol lines, preserving
|
|
# ordinary Unicode and the rendered line structure.
|
|
clean = "".join(ch for ch in value if ch in "\t\n" or ord(ch) >= 0x20 and ch != "\x7f")
|
|
truncated = len(clean) > MAX_READ_TEXT
|
|
return clean[:MAX_READ_TEXT], truncated
|
|
|
|
def list_tabs(self) -> list[dict[str, Any]]:
|
|
self._refresh_tabs()
|
|
# Sort by safe public fields; raw target IDs never determine ordering.
|
|
return [self._public_tab(tab) for tab in sorted(
|
|
(tab for tab in self._tabs.values() if _approved_origin(_origin(tab.url), self.allowed_origins)),
|
|
key=lambda item: (item.kind, _origin(item.url) or "", item.url, item.title)
|
|
)]
|
|
|
|
def apply_context(self, context: dict[str, Any] | None, operation: str = "narrowOriginPolicy") -> None:
|
|
"""Set one effective policy for a single serialized dispatch."""
|
|
if context is None:
|
|
self.allowed_origins = self._base_policy
|
|
return
|
|
if not isinstance(context, dict) or set(context) != {"allowedOrigins"}:
|
|
raise self._fail(operation, "invalid_request", "policy", "invalid operation context")
|
|
try:
|
|
policy = normalize_allowed_origins(context["allowedOrigins"])
|
|
except (ValueError, TypeError):
|
|
raise self._fail(operation, "invalid_request", "policy", "origin policy is invalid")
|
|
if self._base_policy != "*" and (policy == "*" or not policy.issubset(self._base_policy)):
|
|
raise self._fail(operation, "policy_denied", "policy", "origin policy cannot broaden the session policy")
|
|
self.allowed_origins = policy
|
|
|
|
def narrow_origin_policy(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(request, dict) or set(request) != {"allowedOrigins"}:
|
|
raise self._fail("narrowOriginPolicy", "invalid_request", "policy", "invalid origin policy request")
|
|
try:
|
|
requested = normalize_allowed_origins(request["allowedOrigins"])
|
|
except (ValueError, TypeError):
|
|
raise self._fail("narrowOriginPolicy", "invalid_request", "policy", "origin policy is invalid")
|
|
if self._base_policy != "*" and (requested == "*" or not requested.issubset(self._base_policy)):
|
|
raise self._fail("narrowOriginPolicy", "policy_denied", "policy", "origin policy cannot broaden the session policy")
|
|
# If this request itself is context-scoped, the returned context must
|
|
# remain within that effective policy too. It is never persisted.
|
|
if self.allowed_origins != "*" and (requested == "*" or not requested.issubset(self.allowed_origins)):
|
|
raise self._fail("narrowOriginPolicy", "policy_denied", "policy", "origin policy cannot broaden the operation policy")
|
|
return {"allowedOrigins": "*" if requested == "*" else sorted(requested)}
|
|
|
|
def observe(self) -> dict[str, Any]:
|
|
self._resolve_tab({}, "observe")
|
|
try:
|
|
return super().observe()
|
|
except WorkerError as exc:
|
|
if isinstance(exc, BrowserOperationError):
|
|
raise
|
|
message = str(exc).lower()
|
|
code = "stale" if "stale" in message else "policy_denied" if "outside" in message else "worker_failure"
|
|
raise self._fail("observe", code, "observe", "browser observation failed")
|
|
|
|
def act(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(request, dict):
|
|
raise self._fail("act", "invalid_request", "request", "invalid action request")
|
|
self._resolve_tab({}, "act")
|
|
try:
|
|
return super().act(request)
|
|
except BrowserOperationError:
|
|
raise
|
|
except WorkerError as exc:
|
|
# The compatibility worker's policy fuse is scoped to its current
|
|
# effective policy. A narrower context must not poison the
|
|
# immutable base when a page redirects outside that context.
|
|
if self._blocked:
|
|
try:
|
|
current = self.backend.current_tab()
|
|
current_url = current.get("url") if isinstance(current, dict) else None
|
|
if _approved_origin(_origin(current_url), self._base_policy):
|
|
self._blocked = False
|
|
except Exception:
|
|
pass
|
|
# The explicit backend boundary, not arbitrary exception text,
|
|
# decides whether a dispatched mutation may have executed.
|
|
message = str(exc)
|
|
lowered = message.lower()
|
|
state = "unknown" if self._mutation_dispatched and not self._mutation_verified else "not_attempted"
|
|
code = "stale" if "stale" in lowered and not self._mutation_dispatched else "policy_denied" if "outside" in lowered else "worker_failure"
|
|
raise self._fail("act", code, "act", "browser action failed",
|
|
mutation_state=state, retry="never")
|
|
except Exception:
|
|
# Backend exceptions are opaque by design; the dispatch flag still
|
|
# preserves the conservative unknown-mutation classification.
|
|
state = "unknown" if self._mutation_dispatched and not self._mutation_verified else "not_attempted"
|
|
raise self._fail("act", "worker_failure", "act", "browser action failed",
|
|
mutation_state=state, retry="never")
|
|
except BaseException as exc:
|
|
# Cancellation can be raised as a BaseException by an embedding
|
|
# host. Preserve process signals, but classify post-dispatch
|
|
# cancellation exactly like any other uncertain mutation.
|
|
if isinstance(exc, (KeyboardInterrupt, SystemExit)):
|
|
raise
|
|
state = "unknown" if self._mutation_dispatched and not self._mutation_verified else "not_attempted"
|
|
raise self._fail("act", "aborted", "act", "browser action was cancelled",
|
|
mutation_state=state, retry="never")
|
|
|
|
def read(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(request, dict) or set(request) - {"tabId"}:
|
|
raise self._fail("read", "invalid_request", "request", "invalid read request")
|
|
tab = self._resolve_tab(request, "read")
|
|
self._collect_events()
|
|
try:
|
|
current = self.backend.current_tab()
|
|
url = current.get("url")
|
|
title = current.get("title") or ""
|
|
except Exception:
|
|
raise self._fail("read", "worker_failure", "state", "page state is unavailable")
|
|
if not isinstance(url, str) or not _approved_origin(_origin(url), self.allowed_origins):
|
|
raise self._fail("read", "policy_denied", "policy", "page is outside the approved origins")
|
|
text, truncated = self._read_text()
|
|
return {"tab": self._public_tab(tab), "documentId": self._document_id("read"),
|
|
"url": url, "title": _short(title), "visibleText": text,
|
|
"visibleTextTruncated": truncated}
|
|
|
|
def assert_page(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(request, dict) or set(request) - {"tabId", "assertion", "wait"} or "assertion" not in request or request.get("wait") not in {"none", "bounded"}:
|
|
raise self._fail("assert", "invalid_request", "request", "invalid assertion request")
|
|
assertion = request["assertion"]
|
|
if not isinstance(assertion, dict) or set(assertion) != {"kind", "match", "value"}:
|
|
raise self._fail("assert", "invalid_request", "assertion", "invalid assertion")
|
|
kind, match, value = assertion.get("kind"), assertion.get("match"), assertion.get("value")
|
|
valid = ((kind == "url" and match in {"exact", "origin"}) or
|
|
(kind == "title" and match in {"exact", "contains"}) or
|
|
(kind == "visibleText" and match == "contains"))
|
|
if not valid or not isinstance(value, str) or not value or len(value) > MAX_STRING:
|
|
raise self._fail("assert", "invalid_request", "assertion", "invalid assertion")
|
|
if kind == "url" and (_origin(value) is None or (match == "exact" and not _approved_origin(_origin(value), self.allowed_origins))):
|
|
raise self._fail("assert", "policy_denied", "assertion", "assertion URL is outside the approved origins")
|
|
deadline = time.monotonic() + (READ_ASSERTION_TIMEOUT_SECONDS if request["wait"] == "bounded" else 0)
|
|
last: dict[str, Any] | None = None
|
|
while True:
|
|
last = self.read({"tabId": request.get("tabId")} if "tabId" in request else {})
|
|
actual = last["url"] if kind == "url" else last["title"] if kind == "title" else last["visibleText"]
|
|
if kind == "url" and match == "origin":
|
|
satisfied = _origin(actual) == _origin(value)
|
|
elif match == "contains":
|
|
satisfied = value in actual
|
|
else:
|
|
satisfied = actual == value
|
|
if satisfied:
|
|
status = "satisfied"
|
|
break
|
|
if request["wait"] == "none":
|
|
status = "not_satisfied"
|
|
break
|
|
if time.monotonic() >= deadline:
|
|
status = "timed_out"
|
|
break
|
|
time.sleep(0.025)
|
|
return {"status": status, "readOnly": True, "tab": last["tab"], "documentId": last["documentId"]}
|
|
|
|
def _wait_navigation(self, operation: str) -> None:
|
|
wait_ready = getattr(self.backend, "wait_ready", None)
|
|
if callable(wait_ready):
|
|
try:
|
|
ready = wait_ready(NAVIGATION_TIMEOUT_SECONDS)
|
|
except Exception:
|
|
raise self._fail(operation, "worker_failure", "wait", "navigation readiness is unavailable", mutation_state="unknown")
|
|
except BaseException as exc:
|
|
if isinstance(exc, (KeyboardInterrupt, SystemExit)):
|
|
raise
|
|
raise self._fail(operation, "aborted", "wait", "navigation readiness was cancelled", mutation_state="unknown")
|
|
if ready is False:
|
|
raise self._fail(operation, "timeout", "wait", "navigation did not complete", mutation_state="unknown")
|
|
|
|
def navigate(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(request, dict) or set(request) - {"tabId", "url"} or not isinstance(request.get("url"), str):
|
|
raise self._fail("navigate", "invalid_request", "request", "invalid navigation request")
|
|
url = request["url"]
|
|
if len(url) > MAX_STRING or _origin(url) is None or not _approved_origin(_origin(url), self.allowed_origins):
|
|
raise self._fail("navigate", "policy_denied", "policy", "navigation URL is outside the approved origins")
|
|
tab = self._resolve_tab(request, "navigate")
|
|
nav = getattr(self.backend, "navigate", None)
|
|
if not callable(nav):
|
|
raise self._fail("navigate", "unsupported", "navigate", "navigation is unavailable")
|
|
self._mutation_dispatched = False
|
|
self._mutation_verified = False
|
|
try:
|
|
self._mutation_dispatched = True
|
|
nav(url)
|
|
except Exception:
|
|
raise self._fail("navigate", "worker_failure", "navigate", "navigation failed",
|
|
mutation_state="unknown" if self._mutation_dispatched else "not_attempted")
|
|
except BaseException as exc:
|
|
if isinstance(exc, (KeyboardInterrupt, SystemExit)):
|
|
raise
|
|
raise self._fail("navigate", "aborted", "navigate", "navigation was cancelled",
|
|
mutation_state="unknown" if self._mutation_dispatched else "not_attempted")
|
|
self._wait_navigation("navigate")
|
|
self._refresh_tabs()
|
|
current = self._tabs.get(tab.tab_id)
|
|
if current is None or not _approved_origin(_origin(current.url), self.allowed_origins):
|
|
raise self._fail("navigate", "policy_denied", "redirect", "navigation left the approved origins", mutation_state="unknown")
|
|
self._attach(tab.target_id, "navigate")
|
|
self._mutation_verified = True
|
|
return {"status": "committed", "mutationState": "verified", "tab": self._public_tab(current),
|
|
"documentId": self._document_id("navigate")}
|
|
|
|
def reload(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(request, dict) or set(request) - {"tabId"}:
|
|
raise self._fail("reload", "invalid_request", "request", "invalid reload request")
|
|
tab = self._resolve_tab(request, "reload")
|
|
reload_page = getattr(self.backend, "reload", None)
|
|
if not callable(reload_page):
|
|
raise self._fail("reload", "unsupported", "reload", "reload is unavailable")
|
|
self._mutation_dispatched = False
|
|
self._mutation_verified = False
|
|
try:
|
|
self._mutation_dispatched = True
|
|
reload_page()
|
|
except Exception:
|
|
raise self._fail("reload", "worker_failure", "reload", "reload failed",
|
|
mutation_state="unknown" if self._mutation_dispatched else "not_attempted")
|
|
except BaseException as exc:
|
|
if isinstance(exc, (KeyboardInterrupt, SystemExit)):
|
|
raise
|
|
raise self._fail("reload", "aborted", "reload", "reload was cancelled",
|
|
mutation_state="unknown" if self._mutation_dispatched else "not_attempted")
|
|
self._wait_navigation("reload")
|
|
self._refresh_tabs()
|
|
current = self._tabs.get(tab.tab_id)
|
|
if current is None:
|
|
raise self._fail("reload", "no_tab", "reload", "tab closed during reload", mutation_state="unknown")
|
|
self._attach(tab.target_id, "reload")
|
|
self._mutation_verified = True
|
|
return {"status": "committed", "mutationState": "verified", "tab": self._public_tab(current),
|
|
"documentId": self._document_id("reload")}
|
|
|
|
def select_tab(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(request, dict) or set(request) != {"tabId"} or not isinstance(request.get("tabId"), str):
|
|
raise self._fail("selectTab", "invalid_request", "request", "invalid tab selection request")
|
|
self._refresh_tabs()
|
|
tab = self._tabs.get(request["tabId"])
|
|
if tab is None:
|
|
raise self._fail("selectTab", "no_tab", "target", "tab handle is stale or unavailable")
|
|
if not _approved_origin(_origin(tab.url), self.allowed_origins):
|
|
raise self._fail("selectTab", "policy_denied", "policy", "tab is outside the approved origins")
|
|
self._attach(tab.target_id, "selectTab")
|
|
self._selected_tab_id = tab.tab_id
|
|
return {"selected": True, "tab": self._public_tab(tab)}
|
|
|
|
def wait_popup(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(request, dict) or set(request) != {"openerTabId"} or not isinstance(request.get("openerTabId"), str):
|
|
raise self._fail("waitPopup", "invalid_request", "request", "invalid popup request")
|
|
# Refresh first: an immediate popup may already be in the target
|
|
# inventory before waitPopup starts. Newly discovered popup candidates
|
|
# are tracked separately from baseline/previously claimed targets.
|
|
self._refresh_tabs()
|
|
opener = self._tabs.get(request["openerTabId"])
|
|
if opener is None:
|
|
raise self._fail("waitPopup", "no_tab", "target", "opener tab is stale or unavailable")
|
|
if not _approved_origin(_origin(opener.url), self.allowed_origins):
|
|
raise self._fail("waitPopup", "policy_denied", "policy", "opener tab is outside the approved origins")
|
|
deadline = time.monotonic() + POPUP_WAIT_SECONDS
|
|
while True:
|
|
self._refresh_tabs()
|
|
for target in list(self._popup_candidates):
|
|
tab_id = self._target_to_tab.get(target)
|
|
tab = self._tabs.get(tab_id) if tab_id else None
|
|
if (tab is not None and target not in self._claimed_popups
|
|
and tab.opener_target_id == opener.target_id
|
|
and _approved_origin(_origin(tab.url), self.allowed_origins)):
|
|
self._popup_candidates.discard(target)
|
|
self._claimed_popups.add(target)
|
|
return {"status": "found", "tab": self._public_tab(tab), "autoSwitched": False}
|
|
if time.monotonic() >= deadline:
|
|
return {"status": "timed_out", "autoSwitched": False}
|
|
time.sleep(0.025)
|
|
|
|
def close_tab(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(request, dict) or set(request) != {"tabId"} or not isinstance(request.get("tabId"), str):
|
|
raise self._fail("closeTab", "invalid_request", "request", "invalid tab close request")
|
|
self._refresh_tabs()
|
|
tab = self._tabs.get(request["tabId"])
|
|
if tab is None:
|
|
raise self._fail("closeTab", "no_tab", "target", "tab handle is stale or unavailable")
|
|
if not _approved_origin(_origin(tab.url), self.allowed_origins):
|
|
raise self._fail("closeTab", "policy_denied", "policy", "tab is outside the approved origins")
|
|
close = getattr(self.backend, "close_tab", None)
|
|
if not callable(close):
|
|
raise self._fail("closeTab", "unsupported", "close", "tab closing is unavailable")
|
|
self._mutation_dispatched = False
|
|
self._mutation_verified = False
|
|
try:
|
|
self._mutation_dispatched = True
|
|
close(tab.target_id)
|
|
except Exception:
|
|
raise self._fail("closeTab", "worker_failure", "close", "tab close failed",
|
|
mutation_state="unknown" if self._mutation_dispatched else "not_attempted")
|
|
except BaseException as exc:
|
|
if isinstance(exc, (KeyboardInterrupt, SystemExit)):
|
|
raise
|
|
raise self._fail("closeTab", "aborted", "close", "tab close was cancelled",
|
|
mutation_state="unknown" if self._mutation_dispatched else "not_attempted")
|
|
# Do not assume a close acknowledgement means the target disappeared;
|
|
# verify inventories until the target is gone or the existing action
|
|
# budget expires. The close mutation itself is dispatched exactly once.
|
|
deadline = time.monotonic() + ACTION_TIMEOUT_SECONDS
|
|
while True:
|
|
try:
|
|
self._refresh_tabs()
|
|
except BrowserOperationError:
|
|
raise self._fail("closeTab", "worker_failure", "verify", "tab close could not be verified", mutation_state="unknown")
|
|
if tab.target_id not in self._target_to_tab:
|
|
break
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise self._fail("closeTab", "worker_failure", "verify", "tab close could not be verified", mutation_state="unknown")
|
|
time.sleep(min(0.025, remaining))
|
|
self._target_to_tab.pop(tab.target_id, None)
|
|
self._tabs.pop(tab.tab_id, None)
|
|
if self._selected_tab_id == tab.tab_id:
|
|
self._selected_tab_id = None
|
|
self._mutation_verified = True
|
|
return {"closed": True, "tabId": tab.tab_id, "mutationState": "verified"}
|
|
|
|
def _summarize_events(self) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
|
consoles: list[dict[str, Any]] = []
|
|
failures: list[dict[str, Any]] = []
|
|
request_by_id: dict[str, tuple[str, str]] = {}
|
|
active_session = getattr(self.backend, "_session_id", None)
|
|
if not isinstance(active_session, str) or not active_session:
|
|
return consoles, failures
|
|
for event in self._event_buffer[-MAX_EVENTS:]:
|
|
if not isinstance(event, dict) or event.get("session_id") != active_session:
|
|
continue
|
|
method = event.get("method") if isinstance(event, dict) else None
|
|
params = event.get("params") if isinstance(event, dict) else None
|
|
if not isinstance(params, dict):
|
|
continue
|
|
if method == "Runtime.consoleAPICalled":
|
|
level = params.get("type")
|
|
if level not in {"error", "warning", "warn"}:
|
|
continue
|
|
args = params.get("args")
|
|
pieces: list[str] = []
|
|
if isinstance(args, list):
|
|
for arg in args[:8]:
|
|
if isinstance(arg, dict):
|
|
value = arg.get("value")
|
|
if isinstance(value, (str, int, float, bool)):
|
|
pieces.append(str(value))
|
|
elif isinstance(arg.get("description"), str):
|
|
pieces.append(arg["description"])
|
|
message = _policy_event_message(
|
|
_safe_event_message(" ".join(pieces) or "console event"),
|
|
self.allowed_origins,
|
|
)
|
|
existing = next((item for item in consoles if item["level"] == ("warning" if level == "warn" else level) and item["message"] == message), None)
|
|
if existing:
|
|
existing["count"] += 1
|
|
elif len(consoles) < MAX_CONSOLE_FAILURES:
|
|
consoles.append({"level": "warning" if level == "warn" else level,
|
|
"message": message, "count": 1, "truncated": len(message) >= MAX_EVENT_MESSAGE})
|
|
elif method == "Network.requestWillBeSent":
|
|
request_id = params.get("requestId")
|
|
request = params.get("request")
|
|
if isinstance(request_id, str) and isinstance(request, dict):
|
|
url = request.get("url")
|
|
origin = _origin(url) if isinstance(url, str) else None
|
|
if origin and _approved_origin(origin, self.allowed_origins):
|
|
request_by_id[request_id] = (str(request.get("method") or "GET")[:32], origin)
|
|
elif method == "Network.loadingFailed":
|
|
request_id = params.get("requestId")
|
|
method_origin = request_by_id.get(request_id) if isinstance(request_id, str) else None
|
|
if method_origin is None and isinstance(params.get("url"), str):
|
|
origin = _origin(params["url"])
|
|
if origin and _approved_origin(origin, self.allowed_origins):
|
|
method_origin = ("GET", origin)
|
|
if method_origin and len(failures) < MAX_NETWORK_FAILURES:
|
|
failures.append({"method": method_origin[0], "origin": method_origin[1],
|
|
"category": "blocked" if params.get("blockedReason") else "network"})
|
|
elif method == "Network.responseReceived":
|
|
request_id = params.get("requestId")
|
|
response = params.get("response")
|
|
method_origin = request_by_id.get(request_id) if isinstance(request_id, str) else None
|
|
if method_origin and isinstance(response, dict):
|
|
status = response.get("status")
|
|
if isinstance(status, (int, float)) and status >= 400 and len(failures) < MAX_NETWORK_FAILURES:
|
|
failures.append({"method": method_origin[0], "origin": method_origin[1],
|
|
"status": int(status), "category": "http"})
|
|
return consoles, failures
|
|
|
|
def capture_evidence(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(request, dict) or set(request) - {"runId", "reason", "operation"} or not isinstance(request.get("runId"), str) or request.get("reason") not in {"failure", "requested"}:
|
|
raise self._fail("captureEvidence", "invalid_request", "request", "invalid evidence request")
|
|
run_id = request["runId"]
|
|
try:
|
|
_safe_component(run_id, "run id")
|
|
except WorkerError:
|
|
raise self._fail("captureEvidence", "invalid_request", "request", "invalid evidence request")
|
|
operations = {"observe", "act", "read", "assert", "navigate", "reload", "listTabs", "selectTab", "waitPopup", "closeTab", "captureEvidence", "narrowOriginPolicy", "close"}
|
|
if request.get("operation") is not None and request.get("operation") not in operations:
|
|
raise self._fail("captureEvidence", "invalid_request", "request", "invalid evidence operation")
|
|
self._collect_events()
|
|
consoles, failures = self._summarize_events()
|
|
screenshot_payload: dict[str, Any] | None = None
|
|
diagnostics_payload: dict[str, Any] | None = None
|
|
status = "complete"
|
|
target_available = self._selected_tab_id is not None
|
|
if target_available:
|
|
try:
|
|
self._resolve_tab({}, "captureEvidence")
|
|
except BrowserOperationError:
|
|
target_available = False
|
|
else:
|
|
status = "partial"
|
|
shot = getattr(self.backend, "screenshot_bytes", None)
|
|
if callable(shot) and target_available:
|
|
try:
|
|
data = shot()
|
|
if not isinstance(data, (bytes, bytearray)):
|
|
raise WorkerError("screenshot is unavailable")
|
|
raw = bytes(data)
|
|
clipped = raw[:MAX_SCREENSHOT_BYTES]
|
|
screenshot_payload = {
|
|
"kind": "screenshot", "contentType": "image/png",
|
|
"encoding": "base64", "dataBase64": base64.b64encode(clipped).decode("ascii"),
|
|
"byteLength": len(clipped), "truncated": len(raw) > len(clipped),
|
|
}
|
|
if len(raw) > len(clipped):
|
|
status = "partial"
|
|
except Exception:
|
|
status = "partial"
|
|
else:
|
|
status = "partial"
|
|
try:
|
|
payload = json.dumps({"console": consoles, "failedRequests": failures},
|
|
ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
# Keep the diagnostics artifact valid JSON while enforcing its
|
|
# transport ceiling; event summaries are evidence, never a reason
|
|
# to fail the original browser operation.
|
|
while len(payload) > MAX_DIAGNOSTICS_BYTES and (consoles or failures):
|
|
if len(consoles) >= len(failures) and consoles:
|
|
consoles.pop()
|
|
elif failures:
|
|
failures.pop()
|
|
payload = json.dumps({"console": consoles, "failedRequests": failures},
|
|
ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
if len(payload) > MAX_DIAGNOSTICS_BYTES:
|
|
raise WorkerError("diagnostics exceed their size limit")
|
|
diagnostics_payload = {
|
|
"kind": "diagnostics", "contentType": "application/json",
|
|
"encoding": "base64", "dataBase64": base64.b64encode(payload).decode("ascii"),
|
|
"byteLength": len(payload), "truncated": False,
|
|
}
|
|
except Exception:
|
|
status = "partial" if status == "complete" else status
|
|
if screenshot_payload is None and diagnostics_payload is None:
|
|
status = "unavailable"
|
|
return {"status": status,
|
|
**({"screenshot": screenshot_payload} if screenshot_payload else {}),
|
|
**({"diagnostics": diagnostics_payload} if diagnostics_payload else {}),
|
|
"console": consoles, "failedRequests": failures}
|
|
|
|
def close(self) -> None:
|
|
if self._closed_runtime:
|
|
return
|
|
self._closed_runtime = True
|
|
super().close()
|
|
|
|
|
|
# Descriptive aliases for adapters that refer to the published runtime rather
|
|
# than the historical observe/act worker name.
|
|
BrowserRuntime = BrowserRuntimeWorker
|
|
RuntimeWorker = BrowserRuntimeWorker
|
|
|
|
|
|
def _response(request_id: Any, *, result: Any = None, error: str | None = None,
|
|
details: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
if type(request_id) is not int or request_id < 0:
|
|
raise WorkerError("invalid request id")
|
|
if error is not None:
|
|
response = {"id": request_id, "ok": False, "error": str(error)[:MAX_STRING]}
|
|
if details is not None:
|
|
response["details"] = details
|
|
return response
|
|
return {"id": request_id, "ok": True, "result": result}
|
|
|
|
|
|
def process_request(worker: BrowserWorker, request: Any) -> dict[str, Any]:
|
|
runtime_worker = isinstance(worker, BrowserRuntimeWorker)
|
|
startup_worker = isinstance(worker, StartupFailureWorker)
|
|
if not isinstance(request, dict) or type(request.get("id")) is not int:
|
|
raise WorkerError("malformed request")
|
|
if runtime_worker or (startup_worker and "operation" in request):
|
|
if "operation" not in request:
|
|
raise WorkerError("malformed transport request")
|
|
op = request.get("operation", {}).get("op") if isinstance(request.get("operation"), dict) else None
|
|
else:
|
|
if "op" not in request:
|
|
raise WorkerError("malformed request")
|
|
op = request.get("op")
|
|
request_id = request["id"]
|
|
if isinstance(worker, BrowserRuntimeWorker):
|
|
# Published transport shape: operation is the typed BrowserOperation
|
|
# union and context is an optional one-dispatch policy narrowing.
|
|
if (set(request) - {"id", "operation", "context"} or "operation" not in request
|
|
or not isinstance(request["operation"], dict)
|
|
or ("context" in request and request["context"] is None)):
|
|
raise WorkerError("malformed transport request")
|
|
operation = request["operation"]
|
|
if set(operation) - {"op", "request"} or not isinstance(operation.get("op"), str):
|
|
raise WorkerError("malformed transport operation")
|
|
operation_name = operation["op"]
|
|
if operation_name in {"observe", "listTabs", "close"} and set(operation) != {"op"}:
|
|
raise WorkerError("malformed transport operation")
|
|
if operation_name not in {"observe", "listTabs", "close"} and (set(operation) != {"op", "request"} or not isinstance(operation.get("request"), dict)):
|
|
raise WorkerError("malformed transport operation")
|
|
worker.apply_context(request.get("context"), operation_name)
|
|
try:
|
|
body = operation.get("request")
|
|
if operation_name == "observe": result = worker.observe()
|
|
elif operation_name == "act": result = worker.act(body)
|
|
elif operation_name == "read": result = worker.read(body)
|
|
elif operation_name == "assert": result = worker.assert_page(body)
|
|
elif operation_name == "navigate": result = worker.navigate(body)
|
|
elif operation_name == "reload": result = worker.reload(body)
|
|
elif operation_name == "listTabs": result = worker.list_tabs()
|
|
elif operation_name == "selectTab": result = worker.select_tab(body)
|
|
elif operation_name == "waitPopup": result = worker.wait_popup(body)
|
|
elif operation_name == "closeTab": result = worker.close_tab(body)
|
|
elif operation_name == "captureEvidence": result = worker.capture_evidence(body)
|
|
elif operation_name == "narrowOriginPolicy": result = worker.narrow_origin_policy(body)
|
|
elif operation_name == "close":
|
|
worker.close()
|
|
result = {"closed": True}
|
|
else:
|
|
raise WorkerError("unsupported transport operation")
|
|
finally:
|
|
# An omitted context always restores the immutable base, and a
|
|
# supplied context never leaks into the next JSONL request.
|
|
if not worker._closed_runtime:
|
|
worker.apply_context(None)
|
|
return _response(request_id, result={"op": operation_name, "result": result})
|
|
if isinstance(worker, StartupFailureWorker) and "operation" in request:
|
|
if set(request) - {"id", "operation", "context"}:
|
|
raise WorkerError("malformed transport request")
|
|
worker.observe() # raises the bounded startup detail
|
|
if op == "observe":
|
|
if set(request) != {"id", "op"}:
|
|
raise WorkerError("malformed observe request")
|
|
return _response(request_id, result=worker.observe())
|
|
if op == "act":
|
|
if set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
|
|
raise WorkerError("malformed act request")
|
|
return _response(request_id, result=worker.act(request["request"]))
|
|
if op == "read":
|
|
if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
|
|
raise WorkerError("malformed read request")
|
|
return _response(request_id, result=worker.read(request["request"]))
|
|
if op == "assert":
|
|
if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
|
|
raise WorkerError("malformed assert request")
|
|
return _response(request_id, result=worker.assert_page(request["request"]))
|
|
if op == "navigate":
|
|
if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
|
|
raise WorkerError("malformed navigate request")
|
|
return _response(request_id, result=worker.navigate(request["request"]))
|
|
if op == "reload":
|
|
if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
|
|
raise WorkerError("malformed reload request")
|
|
return _response(request_id, result=worker.reload(request["request"]))
|
|
if op == "listTabs":
|
|
if not runtime_worker or set(request) != {"id", "op"}:
|
|
raise WorkerError("malformed listTabs request")
|
|
return _response(request_id, result=worker.list_tabs())
|
|
if op == "selectTab":
|
|
if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
|
|
raise WorkerError("malformed selectTab request")
|
|
return _response(request_id, result=worker.select_tab(request["request"]))
|
|
if op == "waitPopup":
|
|
if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
|
|
raise WorkerError("malformed waitPopup request")
|
|
return _response(request_id, result=worker.wait_popup(request["request"]))
|
|
if op == "closeTab":
|
|
if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
|
|
raise WorkerError("malformed closeTab request")
|
|
return _response(request_id, result=worker.close_tab(request["request"]))
|
|
if op == "captureEvidence":
|
|
if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
|
|
raise WorkerError("malformed captureEvidence request")
|
|
return _response(request_id, result=worker.capture_evidence(request["request"]))
|
|
if op == "narrowOriginPolicy":
|
|
if not runtime_worker or set(request) != {"id", "op", "request"} or not isinstance(request.get("request"), dict):
|
|
raise WorkerError("malformed narrowOriginPolicy request")
|
|
return _response(request_id, result=worker.narrow_origin_policy(request["request"]))
|
|
if op == "close":
|
|
if set(request) != {"id", "op"}:
|
|
raise WorkerError("malformed close request")
|
|
worker.close()
|
|
return _response(request_id, result={"closed": True})
|
|
raise WorkerError("unsupported operation")
|
|
|
|
|
|
def _json_line(value: dict[str, Any]) -> bytes:
|
|
raw = json.dumps(value, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode("utf-8") + b"\n"
|
|
if len(raw) > MAX_OUTPUT_BYTES:
|
|
raise WorkerError("response exceeds protocol limit")
|
|
return raw
|
|
|
|
|
|
def run_stream(worker: BrowserWorker, reader: Any = None, writer: Any = None) -> int:
|
|
reader = reader or sys.stdin.buffer
|
|
writer = writer or sys.stdout.buffer
|
|
while True:
|
|
raw = reader.readline(MAX_LINE_BYTES + 1)
|
|
if not raw:
|
|
break
|
|
request: Any = None
|
|
if len(raw) > MAX_LINE_BYTES:
|
|
response = _response(0, error="request exceeds protocol limit")
|
|
writer.write(_json_line(response)); writer.flush()
|
|
return 1
|
|
try:
|
|
request = json.loads(raw.decode("utf-8"))
|
|
response = process_request(worker, request)
|
|
except Exception as exc:
|
|
request_id = request.get("id", 0) if isinstance(request, dict) else 0
|
|
safe_error = str(exc) if isinstance(exc, WorkerError) else "browser operation failed; no action was retried"
|
|
response = _response(max(0, request_id) if type(request_id) is int else 0,
|
|
error=safe_error[:MAX_STRING],
|
|
details=exc.details if isinstance(exc, BrowserOperationError) else None)
|
|
writer.write(_json_line(response)); writer.flush()
|
|
if isinstance(request, dict) and (request.get("op") == "close" or
|
|
isinstance(request.get("operation"), dict) and request["operation"].get("op") == "close"):
|
|
break
|
|
return 0
|
|
|
|
|
|
def _parse_args(argv: list[str]) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(add_help=False)
|
|
parser.add_argument("--cdp-url", required=True)
|
|
parser.add_argument("--allowed-origins", required=True)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
class StartupFailureWorker:
|
|
"""Keep startup failures inside bounded IPC; never expose raw stderr."""
|
|
|
|
def __init__(self, message: str):
|
|
self.message = message
|
|
|
|
def observe(self) -> dict[str, Any]:
|
|
raise WorkerError(self.message)
|
|
|
|
def act(self, request: Any) -> dict[str, Any]:
|
|
raise WorkerError(self.message)
|
|
|
|
def read(self, request: Any) -> dict[str, Any]:
|
|
raise WorkerError(self.message)
|
|
|
|
def assert_page(self, request: Any) -> dict[str, Any]:
|
|
raise WorkerError(self.message)
|
|
|
|
def navigate(self, request: Any) -> dict[str, Any]:
|
|
raise WorkerError(self.message)
|
|
|
|
def reload(self, request: Any) -> dict[str, Any]:
|
|
raise WorkerError(self.message)
|
|
|
|
def list_tabs(self) -> list[dict[str, Any]]:
|
|
raise WorkerError(self.message)
|
|
|
|
def select_tab(self, request: Any) -> dict[str, Any]:
|
|
raise WorkerError(self.message)
|
|
|
|
def wait_popup(self, request: Any) -> dict[str, Any]:
|
|
raise WorkerError(self.message)
|
|
|
|
def close_tab(self, request: Any) -> dict[str, Any]:
|
|
raise WorkerError(self.message)
|
|
|
|
def capture_evidence(self, request: Any) -> dict[str, Any]:
|
|
raise WorkerError(self.message)
|
|
|
|
def narrow_origin_policy(self, request: Any) -> Any:
|
|
raise WorkerError(self.message)
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = _parse_args(argv or sys.argv[1:])
|
|
worker: BrowserWorker | None = None
|
|
|
|
def shutdown(signum: int, _frame: Any) -> None:
|
|
# Also interrupts startup: RealBrowserBackend catches BaseException and
|
|
# tears down its owned daemon before this SystemExit can escape.
|
|
raise SystemExit(128 + signum)
|
|
|
|
signal.signal(signal.SIGTERM, shutdown)
|
|
signal.signal(signal.SIGINT, shutdown)
|
|
try:
|
|
cdp_url = validate_cdp_url(args.cdp_url)
|
|
origins = json.loads(args.allowed_origins)
|
|
normalized_origins = normalize_allowed_origins(origins)
|
|
session_id = os.environ.get("FAST_BROWSER_SESSION_ID") or f"session-{uuid.uuid4().hex}"
|
|
# Evidence bytes cross the bounded worker transport; the parent
|
|
# runtime commits them to its caller-owned PrivateArtifactStore.
|
|
worker = BrowserRuntimeWorker(
|
|
RealBrowserBackend(cdp_url, normalized_origins), normalized_origins,
|
|
session_id=session_id,
|
|
)
|
|
|
|
return run_stream(worker)
|
|
except Exception as exc:
|
|
detail = str(exc)[:MAX_STRING] if isinstance(exc, WorkerError) else "browser worker initialization failed"
|
|
if isinstance(exc, ModuleNotFoundError) and getattr(exc, "name", "") == "browser_harness":
|
|
detail = "install browser-harness in this Python environment or set FAST_BROWSER_PYTHON to its executable"
|
|
# The bridge deliberately discards raw stderr. Reply to the queued
|
|
# request with a safe failure instead of exiting before IPC starts.
|
|
return run_stream(StartupFailureWorker(detail))
|
|
finally:
|
|
if worker is not None:
|
|
worker.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|