diff --git a/fable_engine/browser.py b/fable_engine/browser.py new file mode 100644 index 0000000..52e766a --- /dev/null +++ b/fable_engine/browser.py @@ -0,0 +1,824 @@ +""" +Stealth Agent Browser Engine for Fable Engine. +A lightweight, agent-first browser engine designed for low memory footprint (<= 20 MB RSS target), +persistent local logins/profiles (~/.fable/browser-profile/), localhost support across any language server, +element indexing, navigation, and layered PNG screenshot generation. +""" + +from __future__ import annotations + +import base64 +import http.cookiejar +from html.parser import HTMLParser +import ipaddress +import json +import logging +import math +import os +import pathlib +import re +import struct +import sys +import threading +import time +import urllib.parse +import urllib.request +import zlib +from typing import Any, Dict, List, Optional, Tuple + +logger = logging.getLogger("fable-engine.browser") + +DEFAULT_PROFILE_DIR = pathlib.Path.home() / ".fable" / "browser-profile" +DEFAULT_VIEWPORT_WIDTH = 1280 +DEFAULT_VIEWPORT_HEIGHT = 800 +MAX_BROWSER_SESSIONS = 32 +MAX_RETAINED_PAGE_HTML_BYTES = 16 * 1024 * 1024 +DEFAULT_MAX_RESPONSE_BYTES = MAX_RETAINED_PAGE_HTML_BYTES // MAX_BROWSER_SESSIONS +DEFAULT_BROWSER_OPEN_TIMEOUT_SECONDS = 15.0 +MAX_BROWSER_OPEN_TIMEOUT_SECONDS = 30.0 +MAX_SCREENSHOT_LAYERS = 10 +MAX_DOM_NODES = 10_000 +MAX_HISTORY_ENTRIES = 100 + + +def bound_browser_timeout(timeout: float) -> float: + """Return a finite, non-negative browser timeout capped at the server limit.""" + value = float(timeout) + if not math.isfinite(value): + raise ValueError("Browser timeout must be finite") + if value < 0: + raise ValueError("Browser timeout must be non-negative") + return min(value, MAX_BROWSER_OPEN_TIMEOUT_SECONDS) + + +class DOMElement: + """Represents a node in the rendered DOM tree with stable element ID and layout box.""" + + def __init__( + self, + element_id: str, + tag: str, + attrs: Dict[str, str], + text: str = "", + children: Optional[List[DOMElement]] = None, + ): + self.element_id = element_id + self.tag = tag.lower() + self.attrs = attrs + self.text = text + self.children = children or [] + self.parent: Optional[DOMElement] = None + self.x = 0 + self.y = 0 + self.width = 0 + self.height = 0 + + def to_dict(self) -> Dict[str, Any]: + attrs = self.attrs + if self.attrs.get("type", "").lower() == "password" and "value" in self.attrs: + attrs = dict(self.attrs) + attrs["value"] = "[REDACTED]" + return { + "element_id": self.element_id, + "tag": self.tag, + "attrs": attrs, + "text": self.text.strip(), + "bbox": [self.x, self.y, self.width, self.height], + "children_count": len(self.children), + } + + +class BrowserDocumentTooLargeError(ValueError): + """Raised when a page exceeds the browser's bounded DOM size.""" + + +class NoHTTPSDowngradeRedirectHandler(urllib.request.HTTPRedirectHandler): + """Follow normal HTTP redirects except HTTPS-to-HTTP downgrades.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): + source_scheme = urllib.parse.urlsplit(req.full_url).scheme.lower() + target_scheme = urllib.parse.urlsplit( + urllib.parse.urljoin(req.full_url, newurl) + ).scheme.lower() + if source_scheme == "https" and target_scheme == "http": + return None + redirected = super().redirect_request(req, fp, code, msg, headers, newurl) + if redirected is not None: + for attr in ("_fable_navigation_deadline", "_fable_navigation_timeout"): + if hasattr(req, attr): + setattr(redirected, attr, getattr(req, attr)) + return redirected + + def http_error_302(self, req, fp, code, msg, headers): + deadline = getattr(req, "_fable_navigation_deadline", None) + if deadline is not None: + timeout = getattr(req, "_fable_navigation_timeout", req.timeout) + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError( + f"Browser navigation timed out after {timeout:g} seconds" + ) + req.timeout = remaining + return super().http_error_302(req, fp, code, msg, headers) + + http_error_301 = http_error_303 = http_error_307 = http_error_308 = http_error_302 + + +def _is_loopback_target(url: str) -> bool: + """Return whether a URL or scheme-less target names the local machine.""" + try: + parsed = urllib.parse.urlsplit(url) + if parsed.scheme.lower() in ("http", "https"): + hostname = parsed.hostname + else: + hostname = urllib.parse.urlsplit(f"//{url}").hostname + except ValueError: + return False + if not hostname: + return False + hostname = hostname.rstrip(".").lower() + if hostname == "localhost": + return True + try: + return ipaddress.ip_address(hostname).is_loopback + except ValueError: + return False + + +class NoRemoteHTTPCookiePolicy(http.cookiejar.DefaultCookiePolicy): + """Never return persistent cookies over remote cleartext HTTP.""" + + def return_ok(self, cookie, request): + parsed = urllib.parse.urlsplit(request.full_url) + if parsed.scheme.lower() == "http" and not _is_loopback_target(request.full_url): + return False + return super().return_ok(cookie, request) + + +class SimpleDOMParser(HTMLParser): + """HTML parser that constructs a clean DOM tree with stable element IDs.""" + + def __init__(self, max_nodes: int = MAX_DOM_NODES): + super().__init__() + self.root = DOMElement("elem_0", "root", {}) + self.stack: List[DOMElement] = [self.root] + self.counter = 1 + self.assigned_ids = {self.root.element_id} + self.max_nodes = max(1, min(int(max_nodes), MAX_DOM_NODES)) + self.node_count = 0 + + def handle_starttag(self, tag: str, attrs: List[Tuple[str, Optional[str]]]): + if self.node_count >= self.max_nodes: + raise BrowserDocumentTooLargeError( + f"Document exceeds the {self.max_nodes} DOM node limit" + ) + attr_dict = {k: (v or "") for k, v in attrs} + base_id = attr_dict.get("id") or f"elem_{self.counter}" + self.counter += 1 + elem_id = base_id + suffix = 2 + while elem_id in self.assigned_ids: + elem_id = f"{base_id}_{suffix}" + suffix += 1 + self.assigned_ids.add(elem_id) + elem = DOMElement(elem_id, tag, attr_dict) + self.node_count += 1 + elem.parent = self.stack[-1] + self.stack[-1].children.append(elem) + if tag.lower() not in ("img", "br", "hr", "input", "meta", "link"): + self.stack.append(elem) + + def handle_endtag(self, tag: str): + if len(self.stack) > 1 and self.stack[-1].tag == tag.lower(): + self.stack.pop() + + def handle_data(self, data: str): + if self.stack: + clean_text = data.strip() + if clean_text: + if self.stack[-1].text: + self.stack[-1].text += " " + clean_text + else: + self.stack[-1].text = clean_text + + +def generate_minimal_png(width: int, height: int, elements: List[DOMElement], bg_color: Tuple[int, int, int] = (245, 247, 250)) -> bytes: + """ + Generates a valid 24-bit RGB PNG image buffer without external image dependencies (pure stdlib struct + zlib). + Renders background canvas and simple color-coded layout boxes for DOM elements. + """ + # Clamp maximum raster dimensions to avoid excessive memory during snapshot + w = min(max(width, 100), 1280) + h = min(max(height, 100), 2400) + + # 3 bytes per pixel RGB + row_bytes = w * 3 + bg_r, bg_g, bg_b = bg_color + # Prefix each scanline with filter type 0, then repeat the RGB background. + scanline = b"\x00" + bytes((bg_r, bg_g, bg_b)) * w + raw_data = bytearray(scanline * h) + + # Simple element box rasterization + for elem in elements: + if elem.width <= 0 or elem.height <= 0: + continue + # Color coding by tag + if elem.tag in ("a", "button", "input"): + box_r, box_g, box_b = (59, 130, 246) # Blue + elif elem.tag in ("h1", "h2", "h3", "header"): + box_r, box_g, box_b = (30, 41, 59) # Dark Slate + else: + box_r, box_g, box_b = (203, 213, 225) # Light Slate Border + + x1 = max(0, min(elem.x, w - 1)) + y1 = max(0, min(elem.y, h - 1)) + x2 = max(0, min(elem.x + elem.width, w - 1)) + y2 = max(0, min(elem.y + elem.height, h - 1)) + + # Fill/Border + for py in range(y1, y2 + 1): + row_offset = py * (row_bytes + 1) + for px_x in range(x1, x2 + 1): + is_border = (px_x == x1 or px_x == x2 or py == y1 or py == y2) + if is_border or elem.tag in ("button", "input"): + idx = row_offset + 1 + px_x * 3 + raw_data[idx] = box_r + raw_data[idx + 1] = box_g + raw_data[idx + 2] = box_b + + compressed = zlib.compress(bytes(raw_data), level=6) + + # Build PNG chunks + ihdr_data = struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0) + ihdr_crc = zlib.crc32(b"IHDR" + ihdr_data) + ihdr_chunk = struct.pack(">I", 13) + b"IHDR" + ihdr_data + struct.pack(">I", ihdr_crc) + + idat_len = len(compressed) + idat_crc = zlib.crc32(b"IDAT" + compressed) + idat_chunk = struct.pack(">I", idat_len) + b"IDAT" + compressed + struct.pack(">I", idat_crc) + + iend_crc = zlib.crc32(b"IEND") + iend_chunk = struct.pack(">I", 0) + b"IEND" + struct.pack(">I", iend_crc) + + png_header = b"\x89PNG\r\n\x1a\n" + return png_header + ihdr_chunk + idat_chunk + iend_chunk + + +class ProfileManager: + """Manages persistent cookies, session data, and login state on local user machine.""" + + def __init__(self, profile_dir: Optional[pathlib.Path | str] = None): + if profile_dir is None: + self.profile_dir = DEFAULT_PROFILE_DIR + else: + self.profile_dir = pathlib.Path(profile_dir) + self.profile_dir.mkdir(mode=0o700, parents=True, exist_ok=True) + self.profile_dir.chmod(0o700) + self.cookie_file = self.profile_dir / "cookies.txt" + self.storage_file = self.profile_dir / "local_storage.json" + self.cookies = http.cookiejar.MozillaCookieJar( + str(self.cookie_file), policy=NoRemoteHTTPCookiePolicy() + ) + self.local_storage: Dict[str, Any] = {} + self._load() + + @staticmethod + def _ensure_private_file(path: pathlib.Path) -> None: + fd = os.open(path, os.O_WRONLY | os.O_CREAT, 0o600) + try: + if hasattr(os, "fchmod"): + os.fchmod(fd, 0o600) + finally: + os.close(fd) + + def _load(self): + if self.cookie_file.exists(): + try: + try: + self.cookie_file.chmod(0o600) + except OSError: + pass + self.cookies.load(ignore_discard=True, ignore_expires=True) + except Exception as e: + logger.warning(f"Could not load cookies from {self.cookie_file}: {e}") + self.cookies.clear() + if self.storage_file.exists(): + try: + try: + self.storage_file.chmod(0o600) + except OSError: + pass + with open(self.storage_file, "r", encoding="utf-8") as f: + self.local_storage = json.load(f) + except Exception as e: + logger.warning(f"Could not load local_storage from {self.storage_file}: {e}") + self.local_storage = {} + + def save(self): + try: + self._ensure_private_file(self.cookie_file) + self.cookies.save(ignore_discard=True, ignore_expires=True) + try: + self.cookie_file.chmod(0o600) + except OSError: + pass + storage_fd = os.open(self.storage_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + if hasattr(os, "fchmod"): + os.fchmod(storage_fd, 0o600) + with os.fdopen(storage_fd, "w", encoding="utf-8") as f: + json.dump(self.local_storage, f, indent=2) + try: + self.storage_file.chmod(0o600) + except OSError: + pass + except Exception as e: + logger.warning(f"Failed to save profile state: {e}") + + +class StealthBrowserSession: + """ + Active browser tab session. + Parses HTML, layouts elements into coordinates, handles inputs, and captures layered PNG screenshots. + """ + + def __init__( + self, + session_id: str, + profile_manager: ProfileManager, + viewport_width: int = DEFAULT_VIEWPORT_WIDTH, + viewport_height: int = DEFAULT_VIEWPORT_HEIGHT, + max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES, + ): + self.session_id = session_id + self.profile_manager = profile_manager + self.viewport_width = viewport_width + self.viewport_height = viewport_height + self.max_response_bytes = max(1, int(max_response_bytes)) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(self.profile_manager.cookies), + NoHTTPSDowngradeRedirectHandler(), + ) + self.credential_free_opener = urllib.request.build_opener( + NoHTTPSDowngradeRedirectHandler() + ) + self.url = "about:blank" + self.history: List[str] = [] + self.history_index = -1 + self.dom_root: Optional[DOMElement] = None + self.elements_by_id: Dict[str, DOMElement] = {} + self.scroll_y = 0 + self.document_height = DEFAULT_VIEWPORT_HEIGHT + self.page_title = "" + self.page_html = "" + self.active_element_id: Optional[str] = None + + def open( + self, + url: str, + timeout: float = DEFAULT_BROWSER_OPEN_TIMEOUT_SECONDS, + *, + record_history: bool = True, + ) -> Dict[str, Any]: + """Opens a URL (including localhost) using urllib with persistent cookies and standard headers.""" + timeout = bound_browser_timeout(timeout) + scheme = urllib.parse.urlsplit(url).scheme.lower() + if scheme not in ("http", "https", "about"): + default_scheme = "http" if self._is_loopback_target(url) else "https" + url = f"{default_scheme}://{url}" + scheme = default_scheme + + if scheme == "about": + page_html = "Blank Page

Blank Page

" + self._commit_navigation(url, page_html, record_history) + return self._build_status() + + req = urllib.request.Request(url) + req.add_header("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36") + req.add_header("Accept", "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8") + req.add_header("Accept-Language", "en-US,en;q=0.9") + + deadline = time.monotonic() + timeout + req._fable_navigation_deadline = deadline + req._fable_navigation_timeout = timeout + try: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError(f"Browser navigation timed out after {timeout:g} seconds") + opener = self.opener + if scheme == "http" and not self._is_loopback_target(url): + opener = self.credential_free_opener + with opener.open(req, timeout=remaining) as resp: + body_bytes = self._read_response_body(resp, deadline, timeout) + self.profile_manager.save() + if len(body_bytes) > self.max_response_bytes: + return { + "status": "error", + "error": f"Response body exceeds {self.max_response_bytes} byte limit", + "url": url, + } + body = body_bytes.decode("utf-8", errors="replace") + response_url = resp.geturl() + self._commit_navigation(response_url, body, record_history) + return self._build_status() + except Exception as e: + logger.error(f"Browser navigation error to {url}: {e}") + return {"status": "error", "error": str(e), "url": url} + + @staticmethod + def _is_loopback_target(url: str) -> bool: + return _is_loopback_target(url) + + def _commit_navigation( + self, + url: str, + page_html: str, + record_history: bool, + ) -> None: + """Parse a candidate document before publishing any navigation state.""" + dom_root, elements_by_id, document_height, page_title = self._layout_document( + url, page_html + ) + self.url = url + self.page_html = page_html + self.dom_root = dom_root + self.elements_by_id = elements_by_id + self.document_height = document_height + self.page_title = page_title + self.active_element_id = None + self.scroll_y = 0 + if record_history: + self._record_history(url) + + def _read_response_body(self, response: Any, deadline: float, timeout: float) -> bytes: + """Read a bounded response while enforcing the navigation's monotonic deadline.""" + body = bytearray() + read_chunk = getattr(response, "read1", response.read) + while len(body) <= self.max_response_bytes: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError(f"Browser navigation timed out after {timeout:g} seconds") + self._set_response_socket_timeout(response, remaining) + chunk = read_chunk(min(64 * 1024, self.max_response_bytes + 1 - len(body))) + if not chunk: + break + body.extend(chunk) + return bytes(body) + + @staticmethod + def _set_response_socket_timeout(response: Any, timeout: float) -> None: + """Apply the remaining deadline to urllib's underlying response socket.""" + stream = response + for _ in range(4): + sock = getattr(stream, "_sock", None) + if sock is not None and hasattr(sock, "settimeout"): + sock.settimeout(timeout) + return + stream = getattr(stream, "raw", None) or getattr(stream, "fp", None) + if stream is None: + return + + def _record_history(self, url: str): + if self.history_index >= 0 and self.history_index < len(self.history): + self.history = self.history[: self.history_index + 1] + if self.history[self.history_index] == url: + return + self.history.append(url) + self.history_index = len(self.history) - 1 + overflow = len(self.history) - MAX_HISTORY_ENTRIES + if overflow > 0: + del self.history[:overflow] + self.history_index -= overflow + + def back(self) -> Dict[str, Any]: + if self.history_index > 0: + target_index = self.history_index - 1 + result = self.open(self.history[target_index], record_history=False) + if result.get("status") != "error": + self.history_index = target_index + return result + return self._build_status() + + def forward(self) -> Dict[str, Any]: + if self.history_index < len(self.history) - 1: + target_index = self.history_index + 1 + result = self.open(self.history[target_index], record_history=False) + if result.get("status") != "error": + self.history_index = target_index + return result + return self._build_status() + + def reload(self) -> Dict[str, Any]: + if self.url and not self.url.startswith("about:"): + return self.open(self.url, record_history=False) + return self._build_status() + + def _parse_and_layout(self): + dom_root, elements_by_id, document_height, page_title = self._layout_document( + self.url, self.page_html + ) + self.dom_root = dom_root + self.elements_by_id = elements_by_id + self.document_height = document_height + self.page_title = page_title + self.active_element_id = None + + def _layout_document( + self, + url: str, + page_html: str, + ) -> Tuple[DOMElement, Dict[str, DOMElement], int, str]: + parser = SimpleDOMParser() + try: + parser.feed(page_html) + except BrowserDocumentTooLargeError: + raise + except Exception: + pass + + dom_root = parser.root + elements_by_id: Dict[str, DOMElement] = {} + + # Extract title + title_match = re.search(r"(.*?)", page_html, re.IGNORECASE | re.DOTALL) + page_title = title_match.group(1).strip() if title_match else url + + # Compute layout boxes + current_y = 20 + all_elems: List[DOMElement] = [] + + def traverse(root: DOMElement): + nonlocal current_y + stack = [root] + while stack: + node = stack.pop() + elements_by_id[node.element_id] = node + all_elems.append(node) + + if node.tag in ("h1", "h2", "h3", "p", "div", "button", "input", "a", "section"): + node.x = 40 + node.y = current_y + node.width = self.viewport_width - 80 + node.height = 36 if node.tag in ("button", "input") else 24 + current_y += node.height + 12 + + stack.extend(reversed(node.children)) + + traverse(dom_root) + document_height = max(self.viewport_height, current_y + 40) + return dom_root, elements_by_id, document_height, page_title + + def scroll(self, delta_y: int) -> Dict[str, Any]: + self.scroll_y = max(0, min(self.scroll_y + delta_y, self.document_height - self.viewport_height)) + return self._build_status() + + def click(self, element_id: str) -> Dict[str, Any]: + """Follow an element's link target; non-link controls are unsupported.""" + elem = self.elements_by_id.get(element_id) + if not elem: + return {"error": f"Element '{element_id}' not found"} + href = elem.attrs.get("href") + if href is not None: + target_url = urllib.parse.urljoin(self.url, href) + return self.open(target_url) + return { + "status": "error", + "error": f"Click is unsupported for <{elem.tag}> elements without a link target", + "element_id": element_id, + } + + def type_text(self, element_id: str, text: str) -> Dict[str, Any]: + elem = self.elements_by_id.get(element_id) + if not elem: + return {"error": f"Element '{element_id}' not found"} + input_type = elem.attrs.get("type", "text").lower() + editable_input_types = { + "text", "search", "email", "url", "tel", "password", "number", + "date", "datetime-local", "month", "time", "week", + } + if elem.tag != "textarea" and not ( + elem.tag == "input" and input_type in editable_input_types + ): + return { + "status": "error", + "error": f"Typing is unsupported for non-editable <{elem.tag}> elements", + "element_id": element_id, + } + elem.attrs["value"] = text + return {"status": "typed", "element_id": element_id, "text": text} + + def close(self) -> None: + """Release per-session resources and discard retained page state.""" + try: + self.opener.close() + finally: + try: + self.credential_free_opener.close() + finally: + self.history.clear() + self.history_index = -1 + self.dom_root = None + self.elements_by_id.clear() + self.page_html = "" + self.page_title = "" + self.active_element_id = None + + def press_key(self, key: str, element_id: Optional[str] = None) -> Dict[str, Any]: + if not key: + return {"status": "error", "error": "Keyboard key must not be empty"} + + if element_id is not None: + if element_id not in self.elements_by_id: + return {"status": "error", "error": f"Element '{element_id}' not found"} + self.active_element_id = element_id + + if key == "Tab": + focusable = [ + elem.element_id for elem in self.elements_by_id.values() + if elem.tag in ("a", "button", "input", "textarea") + ] + if not focusable: + return { + "status": "error", + "error": "Tab is unsupported because the page has no focusable elements", + } + try: + current_index = focusable.index(self.active_element_id) + except ValueError: + current_index = -1 + self.active_element_id = focusable[(current_index + 1) % len(focusable)] + return { + "status": "pressed", + "key": key, + "element_id": self.active_element_id, + } + + target_id = element_id or self.active_element_id + target = self.elements_by_id.get(target_id) if target_id else None + if target is not None: + input_type = target.attrs.get("type", "text").lower() + is_editable = target.tag == "textarea" or ( + target.tag == "input" + and input_type in { + "text", "search", "email", "url", "tel", "password", "number", + "date", "datetime-local", "month", "time", "week", + } + ) + if is_editable: + current_value = target.attrs.get("value", "") + updated_value: Optional[str] = None + if key == "Backspace": + updated_value = current_value[:-1] + elif key == "Space": + updated_value = current_value + " " + elif key == "Enter" and target.tag == "textarea": + updated_value = current_value + "\n" + elif len(key) == 1: + updated_value = current_value + key + if updated_value is not None and updated_value != current_value: + target.attrs["value"] = updated_value + return {"status": "pressed", "key": key, "element_id": target_id} + elif key == "Enter" and target.attrs.get("href") is not None: + return self.click(target.element_id) + + target_description = f"<{target.tag}>" if target is not None else "the active window" + return { + "status": "error", + "error": f"Key '{key}' is unsupported for {target_description}", + "element_id": target_id, + } + + def snapshot_layers(self, max_layers: int = 3) -> Dict[str, Any]: + """Captures N viewport-sized sequential screenshots (1 layer = 1 viewport height).""" + requested_layers = min(max(1, max_layers), MAX_SCREENSHOT_LAYERS) + document_layers = (self.document_height + self.viewport_height - 1) // self.viewport_height + total_layers = max(1, min(requested_layers, document_layers)) + + return self._capture_layers(total_layers, start_top=0) + + def snapshot_viewport(self) -> Dict[str, Any]: + """Captures one viewport at the session's current scroll offset.""" + return self._capture_layers(1, start_top=self.scroll_y) + + def _capture_layers(self, total_layers: int, start_top: int) -> Dict[str, Any]: + layers = [] + all_elements = list(self.elements_by_id.values()) + + for layer_idx in range(total_layers): + layer_top = start_top + layer_idx * self.viewport_height + + visible_elements = [ + e for e in all_elements + if e.y + e.height >= layer_top and e.y <= layer_top + self.viewport_height + ] + + layer_elements = [] + for e in visible_elements: + adjusted = DOMElement(e.element_id, e.tag, e.attrs, e.text) + adjusted.x = e.x + adjusted.y = e.y - layer_top + adjusted.width = e.width + adjusted.height = e.height + layer_elements.append(adjusted) + + png_bytes = generate_minimal_png(self.viewport_width, self.viewport_height, layer_elements) + b64_png = base64.b64encode(png_bytes).decode("ascii") + + layers.append({ + "layer_index": layer_idx + 1, + "viewport": [self.viewport_width, self.viewport_height], + "scroll_top": layer_top, + "b64_png": b64_png, + "elements_count": len(visible_elements), + }) + + return { + "session_id": self.session_id, + "url": self.url, + "title": self.page_title, + "total_layers": len(layers), + "layers": layers, + } + + def _build_status(self) -> Dict[str, Any]: + actionable_tags = ("a", "button", "input", "textarea") + informational_tags = ("h1", "h2", "h3", "form") + actionable_elements = [ + elem.to_dict() for elem in self.elements_by_id.values() + if elem.tag in actionable_tags + ] + informational_elements = [ + elem.to_dict() for elem in self.elements_by_id.values() + if elem.tag in informational_tags + ] + return { + "session_id": self.session_id, + "url": self.url, + "title": self.page_title, + "viewport": [self.viewport_width, self.viewport_height], + "scroll_y": self.scroll_y, + "document_height": self.document_height, + "interactive_elements": (actionable_elements + informational_elements)[:50], + } + + +class StealthBrowserEngine: + """Master Stealth Browser Manager managing active tab sessions and profiles.""" + + def __init__( + self, + profile_dir: Optional[pathlib.Path] = None, + max_sessions: int = MAX_BROWSER_SESSIONS, + ): + self.profile_manager = ProfileManager(profile_dir) + self.sessions: Dict[str, StealthBrowserSession] = {} + self.active_session_id: Optional[str] = None + self.max_sessions = max(1, min(int(max_sessions), MAX_BROWSER_SESSIONS)) + + def get_or_create_session(self, session_id: Optional[str] = None) -> StealthBrowserSession: + sid = session_id or self.active_session_id or "default_session" + if sid not in self.sessions: + if len(self.sessions) >= self.max_sessions: + oldest_sid = next(iter(self.sessions)) + self.close_session(oldest_sid) + self.sessions[sid] = StealthBrowserSession(sid, self.profile_manager) + self.active_session_id = sid + return self.sessions[sid] + + def close_session(self, session_id: Optional[str] = None) -> Dict[str, Any]: + sid = session_id or self.active_session_id + if sid and sid in self.sessions: + session = self.sessions.pop(sid) + session.close() + if self.active_session_id == sid: + self.active_session_id = next(iter(self.sessions.keys())) if self.sessions else None + return {"status": "closed", "session_id": sid} + return {"status": "not_found", "session_id": sid} + + +class _LazyBrowserEngine: + """Compatibility proxy that defers profile creation until first use.""" + + def __init__(self): + self._engine: Optional[StealthBrowserEngine] = None + self._lock = threading.Lock() + + def _get(self) -> StealthBrowserEngine: + if self._engine is None: + with self._lock: + if self._engine is None: + self._engine = StealthBrowserEngine() + return self._engine + + def get_or_create_session(self, session_id: Optional[str] = None) -> StealthBrowserSession: + return self._get().get_or_create_session(session_id) + + def close_session(self, session_id: Optional[str] = None) -> Dict[str, Any]: + if self._engine is None: + return {"status": "not_found", "session_id": session_id} + return self._engine.close_session(session_id) + + def __getattr__(self, name: str) -> Any: + return getattr(self._get(), name) + + +GLOBAL_BROWSER_ENGINE = _LazyBrowserEngine() diff --git a/fable_engine/schema.py b/fable_engine/schema.py index eaf55d9..43f87ef 100644 --- a/fable_engine/schema.py +++ b/fable_engine/schema.py @@ -5,6 +5,10 @@ from __future__ import annotations +from fable_engine.browser import ( + DEFAULT_BROWSER_OPEN_TIMEOUT_SECONDS, + MAX_BROWSER_OPEN_TIMEOUT_SECONDS, +) from fable_engine.session import PHASES TOOL_SCHEMA = { @@ -544,3 +548,167 @@ "required": ["action"] } } + +BROWSER_TOOL_SCHEMAS = [ + { + "name": "browser_open", + "description": "Opens a URL in the stealth agent browser using persistent local logins/profile (<=20 MB RAM ceiling). Supports local dev servers across all languages.", + "inputSchema": { + "type": "object", + "properties": { + "url": {"type": "string", "description": "Target URL or localhost address to open."}, + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."}, + "timeout": { + "type": "number", + "minimum": 0, + "maximum": MAX_BROWSER_OPEN_TIMEOUT_SECONDS, + "default": DEFAULT_BROWSER_OPEN_TIMEOUT_SECONDS, + "description": "Timeout in seconds for page load (maximum 30 seconds).", + } + }, + "required": ["url"] + } + }, + { + "name": "browser_navigate", + "description": "Navigates the browser session to a new URL.", + "inputSchema": { + "type": "object", + "properties": { + "url": {"type": "string", "description": "Target URL to navigate to."}, + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."}, + "timeout": { + "type": "number", + "minimum": 0, + "maximum": MAX_BROWSER_OPEN_TIMEOUT_SECONDS, + "default": DEFAULT_BROWSER_OPEN_TIMEOUT_SECONDS, + "description": "Timeout in seconds for page load (maximum 30 seconds).", + } + }, + "required": ["url"] + } + }, + { + "name": "browser_click", + "description": "Navigates to the href of a link element by its stable element ID. Elements without an href are unsupported.", + "inputSchema": { + "type": "object", + "properties": { + "element_id": {"type": "string", "description": "Stable element ID of an href-bearing link."}, + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."} + }, + "required": ["element_id"] + } + }, + { + "name": "browser_type", + "description": "Types text into an input field identified by its stable element ID.", + "inputSchema": { + "type": "object", + "properties": { + "element_id": {"type": "string", "description": "Stable element ID to type into."}, + "text": {"type": "string", "description": "Text content to enter into element."}, + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."} + }, + "required": ["element_id", "text"] + } + }, + { + "name": "browser_scroll", + "description": "Scrolls the page vertically by delta pixels.", + "inputSchema": { + "type": "object", + "properties": { + "delta_y": {"type": "integer", "description": "Vertical pixel scroll offset."}, + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."} + }, + "required": ["delta_y"] + } + }, + { + "name": "browser_snapshot_layers", + "description": "Captures N sequential screen-sized viewport PNG snapshots (1 layer = 1 PC viewport height, 1280x800 px).", + "inputSchema": { + "type": "object", + "properties": { + "max_layers": {"type": "integer", "minimum": 1, "maximum": 10, "description": "Maximum number of screen-sized viewport layers to capture (default 3)."}, + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."} + } + } + }, + { + "name": "browser_screenshot", + "description": "Captures a single viewport PNG snapshot of the current scroll position.", + "inputSchema": { + "type": "object", + "properties": { + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."} + } + } + }, + { + "name": "browser_close", + "description": "Closes an active browser tab session.", + "inputSchema": { + "type": "object", + "properties": { + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."} + } + } + }, + { + "name": "browser_back", + "description": "Navigates back in history for the browser session.", + "inputSchema": { + "type": "object", + "properties": { + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."} + } + } + }, + { + "name": "browser_forward", + "description": "Navigates forward in history for the browser session.", + "inputSchema": { + "type": "object", + "properties": { + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."} + } + } + }, + { + "name": "browser_wait", + "description": "Waits for a specified duration in seconds.", + "inputSchema": { + "type": "object", + "properties": { + "seconds": {"type": "number", "minimum": 0, "maximum": 10, "description": "Seconds to wait."}, + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."} + }, + "required": ["seconds"] + } + }, + { + "name": "browser_press", + "description": "Edits supported text controls, moves focus with Tab, or activates links with Enter.", + "inputSchema": { + "type": "object", + "properties": { + "key": {"type": "string", "description": "Keyboard key to press (e.g. 'Enter', 'Tab')."}, + "element_id": {"type": "string", "description": "Optional element ID target."}, + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."} + }, + "required": ["key"] + } + }, + { + "name": "browser_reload", + "description": "Reloads the current page in the browser session.", + "inputSchema": { + "type": "object", + "properties": { + "session_id": {"type": "string", "description": "Optional browser tab/session identifier."} + } + } + } +] diff --git a/fable_engine/server.py b/fable_engine/server.py index a5355b6..a61de26 100644 --- a/fable_engine/server.py +++ b/fable_engine/server.py @@ -68,7 +68,12 @@ EpistemicEvidenceValidator, ModelVelocityProfiler, ) -from fable_engine.schema import TOOL_SCHEMA +from fable_engine.browser import ( + DEFAULT_BROWSER_OPEN_TIMEOUT_SECONDS, + GLOBAL_BROWSER_ENGINE, + bound_browser_timeout, +) +from fable_engine.schema import BROWSER_TOOL_SCHEMAS, TOOL_SCHEMA from fable_engine.session import ( ACTIVE_SESSIONS, FORCE_UNLOCK_ENV, @@ -110,6 +115,7 @@ def __getattr__(name: str) -> Any: MAX_RPC_LINE_BYTES = 1 * 1024 * 1024 MAX_RPC_RESPONSE_BYTES = 2 * 1024 * 1024 +BROWSER_TOOL_NAMES = frozenset(tool["name"] for tool in BROWSER_TOOL_SCHEMAS) def send_response(response_dict: Dict[str, Any]): @@ -172,7 +178,13 @@ def _bounded_lines(stream, limit: int): def main(): logger.info("Starting Fable-Engine MCP Server on stdio...") - if AutoUpdater is not None: + if ( + AutoUpdater is not None + and not os.environ.get("FABLE_DISABLE_AUTO_UPDATE") + and not os.environ.get("PYTEST_CURRENT_TEST") + and not os.environ.get("GITHUB_ACTIONS") + and not os.environ.get("CI") + ): try: AutoUpdater().trigger_silent_background_update() except Exception as e: @@ -245,7 +257,7 @@ def main(): "jsonrpc": "2.0", "id": msg_id, "result": { - "tools": [TOOL_SCHEMA] + "tools": [TOOL_SCHEMA] + BROWSER_TOOL_SCHEMAS } }) @@ -288,6 +300,94 @@ def main(): "isError": True } }) + elif tool_name in BROWSER_TOOL_NAMES: + try: + sid = arguments.get("session_id") + res: Any = None + + if tool_name in ("browser_open", "browser_navigate"): + timeout = bound_browser_timeout( + arguments.get("timeout", DEFAULT_BROWSER_OPEN_TIMEOUT_SECONDS) + ) + + if tool_name == "browser_close": + res = GLOBAL_BROWSER_ENGINE.close_session(sid) + elif tool_name == "browser_wait": + sec = float(arguments.get("seconds", 1.0)) + import time + bounded_sec = min(max(sec, 0), 10) + time.sleep(bounded_sec) + res = {"status": "waited", "seconds": bounded_sec} + else: + session = GLOBAL_BROWSER_ENGINE.get_or_create_session(sid) + + if tool_name in ("browser_open", "browser_navigate"): + url = str(arguments.get("url", "")) + res = session.open(url, timeout=timeout) + elif tool_name == "browser_click": + elem_id = str(arguments.get("element_id", "")) + res = session.click(elem_id) + elif tool_name == "browser_type": + elem_id = str(arguments.get("element_id", "")) + txt = str(arguments.get("text", "")) + res = session.type_text(elem_id, txt) + elif tool_name == "browser_scroll": + delta_y = int(arguments.get("delta_y", 0)) + res = session.scroll(delta_y) + elif tool_name == "browser_snapshot_layers": + max_l = int(arguments.get("max_layers", 3)) + res = session.snapshot_layers(max_layers=max_l) + elif tool_name == "browser_screenshot": + res = session.snapshot_viewport() + elif tool_name == "browser_close": + pass + elif tool_name == "browser_back": + res = session.back() + elif tool_name == "browser_forward": + res = session.forward() + elif tool_name == "browser_reload": + res = session.reload() + elif tool_name == "browser_wait": + pass + elif tool_name == "browser_press": + key = str(arguments.get("key", "")) + elem_id = arguments.get("element_id") + res = session.press_key(key, elem_id) + else: + raise ValueError(f"Unknown browser tool action: {tool_name}") + + is_error = ( + isinstance(res, dict) + and (res.get("status") == "error" or "error" in res) + ) + send_response({ + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "content": [ + { + "type": "text", + "text": json.dumps(res, indent=2) + } + ], + "isError": is_error + } + }) + except Exception as ex: + logger.error(f"Error handling {tool_name}: {ex}", exc_info=True) + send_response({ + "jsonrpc": "2.0", + "id": msg_id, + "result": { + "content": [ + { + "type": "text", + "text": f"Stealth Browser Error: {str(ex)}" + } + ], + "isError": True + } + }) else: send_response({ "jsonrpc": "2.0", diff --git a/fable_engine/session.py b/fable_engine/session.py index 69c1fda..a2e63ff 100644 --- a/fable_engine/session.py +++ b/fable_engine/session.py @@ -275,6 +275,8 @@ def __init__( AutoUpdater is not None and not os.environ.get("FABLE_DISABLE_AUTO_UPDATE") and not os.environ.get("PYTEST_CURRENT_TEST") + and not os.environ.get("GITHUB_ACTIONS") + and not os.environ.get("CI") ): try: AutoUpdater().trigger_silent_background_update() diff --git a/fable_engine/test_server.py b/fable_engine/test_server.py index 12d6ef4..589a220 100644 --- a/fable_engine/test_server.py +++ b/fable_engine/test_server.py @@ -780,8 +780,10 @@ def test_mcp_handshake_and_tool_call(self): # 2. Tools List resp = self._rpc_call("tools/list") tools = resp["result"]["tools"] - self.assertEqual(len(tools), 1) - self.assertEqual(tools[0]["name"], "fable_session") + self.assertEqual(len(tools), 14) + tool_names = [t["name"] for t in tools] + self.assertIn("fable_session", tool_names) + self.assertIn("browser_open", tool_names) self.assertIn("log_refinement_cycle", tools[0]["inputSchema"]["properties"]["action"]["enum"]) # 3. Call fable_session: create_session diff --git a/fable_mode/manifest.py b/fable_mode/manifest.py index d95040e..f21309e 100644 --- a/fable_mode/manifest.py +++ b/fable_mode/manifest.py @@ -17,6 +17,7 @@ "fable_mode/safety.py", "fable_mode/resources.json", "fable_engine/__init__.py", + "fable_engine/browser.py", "fable_engine/actions/__init__.py", "fable_engine/actions/cas.py", "fable_engine/actions/deliberation.py", diff --git a/fable_mode/resources.json b/fable_mode/resources.json index 102ef08..f2e0d5b 100644 --- a/fable_mode/resources.json +++ b/fable_mode/resources.json @@ -10,6 +10,7 @@ "fable_mode/safety.py", "fable_mode/resources.json", "fable_engine/__init__.py", + "fable_engine/browser.py", "fable_engine/actions/__init__.py", "fable_engine/actions/cas.py", "fable_engine/actions/deliberation.py", diff --git a/run_tests.py b/run_tests.py index 1183ef9..d973f40 100644 --- a/run_tests.py +++ b/run_tests.py @@ -7,11 +7,14 @@ import argparse import ast +import os import sys import subprocess import time from pathlib import Path +os.environ["FABLE_DISABLE_AUTO_UPDATE"] = "1" + # Explicitly documented allowlist for legitimate dynamic evaluation ALLOWED_EVAL_FILES = { diff --git a/tests/test_execution_broker.py b/tests/test_execution_broker.py index 485e52f..8fd250c 100644 --- a/tests/test_execution_broker.py +++ b/tests/test_execution_broker.py @@ -49,6 +49,7 @@ def test_command_is_allowlisted_and_runs_after_authorization(self): def test_inspect_files_is_implemented_and_bounded(self): target = self.workspace / "input.txt" target.write_text("hello world", encoding="utf-8") + target.chmod(0o600) result = self.broker.handle({"action": "inspect_files", "path": "input.txt"}) self.assertEqual(result["content"], "hello world") self.assertFalse(result["truncated"]) diff --git a/tests/test_stealth_browser.py b/tests/test_stealth_browser.py new file mode 100644 index 0000000..37817cd --- /dev/null +++ b/tests/test_stealth_browser.py @@ -0,0 +1,970 @@ +""" +Unit tests for Stealth Agent Browser Engine and top-level MCP browser tools. +Verifies memory footprint target, local persistent profiles (~/.fable/browser-profile/), +navigation history, element clicking, typing, scrolling, and layered PNG screenshot generation. +""" + +from __future__ import annotations + +import http.cookiejar +import http.server +import io +import json +import os +import shutil +import stat +import struct +import subprocess +import sys +import tempfile +import threading +import unittest +import urllib.request +import urllib.response +import zlib +from email.message import Message +from unittest.mock import Mock, patch + +from fable_engine.browser import ( + DEFAULT_MAX_RESPONSE_BYTES, + DEFAULT_VIEWPORT_HEIGHT, + DEFAULT_VIEWPORT_WIDTH, + DOMElement, + MAX_BROWSER_OPEN_TIMEOUT_SECONDS, + MAX_BROWSER_SESSIONS, + MAX_DOM_NODES, + MAX_HISTORY_ENTRIES, + MAX_RETAINED_PAGE_HTML_BYTES, + MAX_SCREENSHOT_LAYERS, + ProfileManager, + StealthBrowserEngine, + StealthBrowserSession, + generate_minimal_png, +) +from fable_engine.schema import BROWSER_TOOL_SCHEMAS +import fable_engine.server as browser_server + + +class _BrowserTestHandler(http.server.BaseHTTPRequestHandler): + redirect_history_one = False + downgrade_requests = 0 + downgrade_cookie = None + + def do_GET(self): + if self.path == "/set-cookie": + self.send_response(200) + self.send_header("Set-Cookie", "session_token=abc123secret; Path=/; HttpOnly") + body = b"cookie set" + elif self.path == "/large": + self.send_response(200) + body = b"01234567890" + elif self.path == "/oversized-dom": + self.send_response(200) + body = ("" * (MAX_DOM_NODES + 1)).encode("utf-8") + elif self.path == "/history-one" and self.redirect_history_one: + self.send_response(302) + self.send_header("Location", "/redirected") + self.end_headers() + return + elif self.path == "/downgrade-target": + type(self).downgrade_requests += 1 + type(self).downgrade_cookie = self.headers.get("Cookie") + self.send_response(200) + body = b"downgrade followed" + else: + self.send_response(200) + body = (self.headers.get("Cookie") or "").encode("utf-8") + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format, *args): + pass + + +class TestStealthAgentBrowser(unittest.TestCase): + + def setUp(self): + self.tmp_dir = tempfile.mkdtemp() + self.profile_dir = os.path.join(self.tmp_dir, "browser-profile") + _BrowserTestHandler.redirect_history_one = False + _BrowserTestHandler.downgrade_requests = 0 + _BrowserTestHandler.downgrade_cookie = None + self.httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _BrowserTestHandler) + self.http_thread = threading.Thread(target=self.httpd.serve_forever, daemon=True) + self.http_thread.start() + self.port = self.httpd.server_address[1] + + def tearDown(self): + self.httpd.shutdown() + self.httpd.server_close() + self.http_thread.join(timeout=2) + shutil.rmtree(self.tmp_dir, ignore_errors=True) + + def test_browser_schemas_exist(self): + tool_names = [t["name"] for t in BROWSER_TOOL_SCHEMAS] + self.assertIn("browser_open", tool_names) + self.assertIn("browser_navigate", tool_names) + self.assertIn("browser_click", tool_names) + self.assertIn("browser_type", tool_names) + self.assertIn("browser_scroll", tool_names) + self.assertIn("browser_snapshot_layers", tool_names) + self.assertIn("browser_screenshot", tool_names) + self.assertIn("browser_close", tool_names) + self.assertIn("browser_back", tool_names) + self.assertIn("browser_forward", tool_names) + self.assertIn("browser_wait", tool_names) + self.assertIn("browser_press", tool_names) + self.assertIn("browser_reload", tool_names) + + press_schema = next( + schema for schema in BROWSER_TOOL_SCHEMAS + if schema["name"] == "browser_press" + ) + self.assertNotIn("active window", press_schema["description"]) + + for tool in BROWSER_TOOL_SCHEMAS: + self.assertTrue(len(tool["description"]) > 0) + self.assertIn("type", tool["inputSchema"]) + + schemas = {tool["name"]: tool["inputSchema"] for tool in BROWSER_TOOL_SCHEMAS} + layers_schema = schemas["browser_snapshot_layers"]["properties"]["max_layers"] + self.assertEqual(layers_schema["minimum"], 1) + self.assertEqual(layers_schema["maximum"], MAX_SCREENSHOT_LAYERS) + open_timeout_schema = schemas["browser_open"]["properties"]["timeout"] + navigate_timeout_schema = schemas["browser_navigate"]["properties"]["timeout"] + self.assertEqual(open_timeout_schema["maximum"], MAX_BROWSER_OPEN_TIMEOUT_SECONDS) + self.assertEqual(open_timeout_schema["default"], 15.0) + self.assertEqual(navigate_timeout_schema["maximum"], MAX_BROWSER_OPEN_TIMEOUT_SECONDS) + wait_schema = schemas["browser_wait"]["properties"]["seconds"] + self.assertEqual(wait_schema["maximum"], 10) + click_schema = next( + tool for tool in BROWSER_TOOL_SCHEMAS if tool["name"] == "browser_click" + ) + self.assertIn("href", click_schema["description"]) + self.assertIn( + "href", click_schema["inputSchema"]["properties"]["element_id"]["description"] + ) + + def test_profile_manager_persistence(self): + pm = ProfileManager(profile_dir=self.profile_dir) + pm.cookies.set_cookie(http.cookiejar.Cookie( + version=0, name="session_token", value="abc123secret", + port=None, port_specified=False, domain="localhost.local", + domain_specified=False, domain_initial_dot=False, path="/", + path_specified=True, secure=False, expires=None, discard=False, + comment=None, comment_url=None, rest={}, rfc2109=False, + )) + pm.local_storage["user_pref"] = {"theme": "dark"} + pm.save() + + # Reload from disk + pm2 = ProfileManager(profile_dir=self.profile_dir) + cookies = {cookie.name: cookie.value for cookie in pm2.cookies} + self.assertEqual(cookies["session_token"], "abc123secret") + self.assertEqual(pm2.local_storage.get("user_pref"), {"theme": "dark"}) + + def test_profile_permissions_are_explicitly_private(self): + old_umask = os.umask(0) + try: + pm = ProfileManager(profile_dir=self.profile_dir) + pm.save() + finally: + os.umask(old_umask) + + if os.name != "nt": + self.assertEqual(stat.S_IMODE(os.stat(self.profile_dir).st_mode), 0o700) + self.assertEqual(stat.S_IMODE(os.stat(pm.cookie_file).st_mode), 0o600) + self.assertEqual(stat.S_IMODE(os.stat(pm.storage_file).st_mode), 0o600) + + def test_cookie_jar_does_not_send_cookie_to_another_origin(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + + session.open(f"http://localhost:{self.port}/set-cookie") + session.open(f"http://localhost:{self.port}/echo") + self.assertIn("session_token=abc123secret", session.page_html) + session.open(f"http://127.0.0.1:{self.port}/echo") + + self.assertNotIn("session_token", session.page_html) + + def test_remote_http_never_receives_persistent_cookies(self): + pm = ProfileManager(profile_dir=self.profile_dir) + pm.cookies.set_cookie(http.cookiejar.Cookie( + version=0, name="session_token", value="abc123secret", + port=None, port_specified=False, domain="example.test", + domain_specified=False, domain_initial_dot=False, path="/", + path_specified=True, secure=False, expires=None, discard=False, + comment=None, comment_url=None, rest={}, rfc2109=False, + )) + + cleartext_request = urllib.request.Request("http://example.test/path") + pm.cookies.add_cookie_header(cleartext_request) + secure_request = urllib.request.Request("https://example.test/path") + pm.cookies.add_cookie_header(secure_request) + + self.assertIsNone(cleartext_request.get_header("Cookie")) + self.assertEqual(secure_request.get_header("Cookie"), "session_token=abc123secret") + + def test_remote_http_uses_credential_free_opener_for_redirect_chain(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + session.opener = Mock() + session.credential_free_opener = Mock() + session.credential_free_opener.open.side_effect = OSError( + "stop before network access" + ) + + session.open("http://example.test/path") + + session.opener.open.assert_not_called() + request = session.credential_free_opener.open.call_args.args[0] + self.assertEqual(request.full_url, "http://example.test/path") + + def test_http_redirect_to_remote_target_does_not_receive_cookie(self): + class FakeHTTPRedirectHandler(urllib.request.BaseHandler): + handler_order = 100 + + def __init__(inner_self): + inner_self.requests = [] + + def http_open(inner_self, request): + inner_self.requests.append(request) + headers = Message() + if request.full_url.startswith("http://localhost/"): + headers.add_header("Location", "http://example.test/final") + body = io.BytesIO() + body.msg = headers + return urllib.response.addinfourl( + body, headers, request.full_url, 302 + ) + body = io.BytesIO(b"redirect complete") + headers.add_header("Content-Type", "text/html; charset=utf-8") + body.msg = headers + return urllib.response.addinfourl( + body, headers, request.full_url, 200 + ) + + pm = ProfileManager(profile_dir=self.profile_dir) + pm.cookies.set_cookie(http.cookiejar.Cookie( + version=0, name="session_token", value="abc123secret", + port=None, port_specified=False, domain="example.test", + domain_specified=False, domain_initial_dot=False, path="/", + path_specified=True, secure=False, expires=None, discard=False, + comment=None, comment_url=None, rest={}, rfc2109=False, + )) + session = StealthBrowserSession("test_tab", pm) + handler = FakeHTTPRedirectHandler() + session.opener.add_handler(handler) + + result = session.open("http://localhost/start") + + self.assertEqual(result["url"], "http://example.test/final") + self.assertEqual(len(handler.requests), 2) + self.assertIsNone(handler.requests[1].get_header("Cookie")) + + def test_redirect_hops_receive_remaining_navigation_deadline(self): + class DelayedRedirectHandler(urllib.request.BaseHandler): + handler_order = 100 + + def __init__(inner_self): + inner_self.timeouts = [] + + def http_open(inner_self, request): + inner_self.timeouts.append(request.timeout) + headers = Message() + headers.add_header("Location", f"/hop-{len(inner_self.timeouts)}") + body = io.BytesIO() + body.msg = headers + return urllib.response.addinfourl( + body, headers, request.full_url, 302 + ) + + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + handler = DelayedRedirectHandler() + session.opener.add_handler(handler) + + with ( + patch( + "fable_engine.browser.time.monotonic", + side_effect=(100.0, 100.1, 100.4, 101.1), + ), + patch.object(session, "_read_response_body") as read_response_body, + ): + result = session.open("http://localhost/start", timeout=1.0) + + self.assertEqual(result["status"], "error") + self.assertIn("timed out", result["error"]) + self.assertEqual(len(handler.timeouts), 2) + self.assertAlmostEqual(handler.timeouts[0], 0.9) + self.assertAlmostEqual(handler.timeouts[1], 0.6) + read_response_body.assert_not_called() + + def test_https_cookie_redirect_to_http_is_not_followed(self): + class FakeHTTPSRedirectHandler(urllib.request.BaseHandler): + handler_order = 100 + + def https_open(inner_self, request): + headers = Message() + headers.add_header( + "Location", f"http://localhost:{self.port}/downgrade-target" + ) + headers.add_header( + "Set-Cookie", "session_token=should-not-leak; Path=/" + ) + body = io.BytesIO() + body.msg = headers + return urllib.response.addinfourl( + body, headers, request.full_url, 302 + ) + + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + session.opener.add_handler(FakeHTTPSRedirectHandler()) + + result = session.open(f"https://localhost:{self.port}/redirect") + + self.assertEqual(result["status"], "error") + self.assertIn("HTTP Error 302", result["error"]) + self.assertEqual( + {cookie.name: cookie.value for cookie in pm.cookies}["session_token"], + "should-not-leak", + ) + self.assertEqual(_BrowserTestHandler.downgrade_requests, 0) + self.assertIsNone(_BrowserTestHandler.downgrade_cookie) + + def test_browser_session_about_blank(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + res = session.open("about:blank") + + self.assertEqual(res["url"], "about:blank") + self.assertEqual(res["title"], "Blank Page") + self.assertEqual(res["viewport"], [DEFAULT_VIEWPORT_WIDTH, DEFAULT_VIEWPORT_HEIGHT]) + + def test_dom_parsing_and_element_interaction(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + session.page_html = """ + + Test Localhost Page + +

Welcome to Localhost App

+ + + + + Return Home + + + """ + session._parse_and_layout() + status = session._build_status() + + self.assertEqual(status["title"], "Test Localhost Page") + self.assertIn("username_input", session.elements_by_id) + self.assertIn("login_btn", session.elements_by_id) + + # Test type_text + type_res = session.type_text("username_input", "agent_user") + self.assertEqual(type_res["status"], "typed") + self.assertEqual(session.elements_by_id["username_input"].attrs.get("value"), "agent_user") + self.assertEqual(session.type_text("notes", "editable")["status"], "typed") + self.assertIn( + "notes", + {element["element_id"] for element in status["interactive_elements"]}, + ) + + self.assertEqual(session.type_text("login_btn", "nope")["status"], "error") + self.assertEqual(session.type_text("home_link", "nope")["status"], "error") + self.assertEqual(session.type_text("remember_me", "nope")["status"], "error") + + self.assertEqual(session.click("login_btn")["status"], "error") + self.assertEqual(session.press_key("!", "username_input")["status"], "pressed") + self.assertEqual( + session.elements_by_id["username_input"].attrs["value"], "agent_user!" + ) + self.assertEqual(session.press_key("Backspace")["status"], "pressed") + self.assertEqual( + session.elements_by_id["username_input"].attrs["value"], "agent_user" + ) + self.assertEqual(session.press_key("Tab", "username_input")["element_id"], "remember_me") + self.assertEqual(session.press_key("Enter", "username_input")["status"], "error") + self.assertEqual(session.press_key("Enter", "login_btn")["status"], "error") + self.assertEqual(session.press_key("Enter", "notes")["status"], "pressed") + self.assertTrue(session.elements_by_id["notes"].attrs["value"].endswith("\n")) + + # Test click link + click_res = session.click("home_link") + self.assertEqual(click_res["url"], "about:blank") + + def test_password_values_are_redacted_without_mutating_dom(self): + password = DOMElement( + "password", "input", {"type": "PASSWORD", "value": "secret", "name": "password"} + ) + regular = DOMElement("username", "input", {"type": "text", "value": "agent"}) + + self.assertEqual(password.to_dict()["attrs"]["value"], "[REDACTED]") + self.assertEqual(password.to_dict()["attrs"]["name"], "password") + self.assertEqual(password.attrs["value"], "secret") + self.assertEqual(regular.to_dict()["attrs"]["value"], "agent") + + def test_oversized_dom_returns_an_explicit_error(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + + result = session.open(f"http://127.0.0.1:{self.port}/oversized-dom") + + self.assertEqual(result["status"], "error") + self.assertIn(f"{MAX_DOM_NODES} DOM node limit", result["error"]) + self.assertLess(len(session.elements_by_id), MAX_DOM_NODES) + + def test_failed_navigation_preserves_previous_page_state(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + session.open("about:blank") + session.scroll_y = 275 + previous_state = ( + session.url, + session.page_html, + session.dom_root, + session.elements_by_id, + session.page_title, + session.document_height, + list(session.history), + session.history_index, + session.scroll_y, + ) + + session.opener = Mock() + session.opener.open.side_effect = OSError("navigation failed") + result = session.open("example.test") + + self.assertEqual(result["status"], "error") + self.assertEqual(result["url"], "https://example.test") + self.assertEqual( + ( + session.url, + session.page_html, + session.dom_root, + session.elements_by_id, + session.page_title, + session.document_height, + session.history, + session.history_index, + session.scroll_y, + ), + previous_state, + ) + + def test_layout_failure_preserves_previous_page_state(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + session.open("about:blank") + session.scroll_y = 275 + previous_state = ( + session.url, + session.page_html, + session.dom_root, + session.elements_by_id, + session.page_title, + session.document_height, + list(session.history), + session.history_index, + session.scroll_y, + ) + + result = session.open(f"http://127.0.0.1:{self.port}/oversized-dom") + + self.assertEqual(result["status"], "error") + self.assertEqual( + ( + session.url, + session.page_html, + session.dom_root, + session.elements_by_id, + session.page_title, + session.document_height, + session.history, + session.history_index, + session.scroll_y, + ), + previous_state, + ) + + def test_navigation_resets_scroll_before_layout(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + session.scroll_y = 275 + + session.open("about:blank") + self.assertEqual(session.scroll_y, 0) + + session.scroll_y = 275 + session.open(f"http://127.0.0.1:{self.port}/echo") + self.assertEqual(session.scroll_y, 0) + + def test_duplicate_explicit_and_generated_element_ids_are_unique(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + session.page_html = ( + '
' + '

' + ) + session._parse_and_layout() + + ids = [element.element_id for element in session.elements_by_id.values()] + self.assertEqual(len(ids), len(set(ids))) + self.assertIn("duplicate", ids) + self.assertIn("duplicate_2", ids) + self.assertEqual(len(ids), 6) + + def test_layered_png_snapshot(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + session.page_html = "" + "".join([f"

Header {i}

Paragraph text {i}

" for i in range(50)]) + "" + session._parse_and_layout() + + snapshot_res = session.snapshot_layers(max_layers=3) + self.assertEqual(snapshot_res["session_id"], "test_tab") + self.assertGreaterEqual(snapshot_res["total_layers"], 1) + self.assertTrue(len(snapshot_res["layers"]) > 0) + + first_layer = snapshot_res["layers"][0] + self.assertEqual(first_layer["layer_index"], 1) + self.assertTrue(first_layer["b64_png"].startswith("iVBORw0KGgo")) # PNG magic header in base64 + + def test_snapshot_limits_layers_and_viewport_uses_current_scroll(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm, viewport_width=100, viewport_height=100) + session.page_html = "

content

" + session._parse_and_layout() + session.document_height = 5000 + session.scroll_y = 275 + + result = session.snapshot_layers(max_layers=10_000) + self.assertEqual(result["total_layers"], MAX_SCREENSHOT_LAYERS) + self.assertEqual(session.snapshot_viewport()["layers"][0]["scroll_top"], 275) + + def test_response_body_limit_rejects_oversized_content(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm, max_response_bytes=10) + + result = session.open(f"http://127.0.0.1:{self.port}/large") + + self.assertEqual(result["status"], "error") + self.assertIn("10 byte limit", result["error"]) + self.assertEqual(session.history, []) + + def test_history_traversal_does_not_mutate_history_on_redirect(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + one = f"http://127.0.0.1:{self.port}/history-one" + two = f"http://127.0.0.1:{self.port}/history-two" + session.open(one) + session.open(two) + original_history = list(session.history) + _BrowserTestHandler.redirect_history_one = True + + session.back() + self.assertEqual(session.history, original_history) + self.assertEqual(session.history_index, 0) + self.assertTrue(session.url.endswith("/redirected")) + + session.reload() + self.assertEqual(session.history, original_history) + self.assertEqual(session.history_index, 0) + self.assertTrue(session.url.endswith("/redirected")) + + session.forward() + self.assertEqual(session.history, original_history) + self.assertEqual(session.history_index, 1) + self.assertEqual(session.url, two) + + def test_history_is_bounded_and_remains_consistent_after_branching(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + urls = [f"about:page-{index}" for index in range(MAX_HISTORY_ENTRIES + 5)] + + for url in urls: + session._record_history(url) + + self.assertEqual(session.history, urls[-MAX_HISTORY_ENTRIES:]) + self.assertEqual(session.history_index, MAX_HISTORY_ENTRIES - 1) + + session.back() + session.open("about:branched") + session.open("about:branched") + + self.assertEqual(len(session.history), MAX_HISTORY_ENTRIES) + self.assertEqual(session.history[-1], "about:branched") + self.assertEqual(session.history_index, len(session.history) - 1) + self.assertEqual(session.forward()["url"], "about:branched") + + def test_same_url_navigation_clears_forward_history(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + for url in ("about:A", "about:B", "about:C"): + session.open(url) + + session.back() + session.open("about:B") + + self.assertEqual(session.history, ["about:A", "about:B"]) + self.assertEqual(session.history_index, 1) + self.assertEqual(session.forward()["url"], "about:B") + + def test_url_scheme_checks_are_case_insensitive_without_rewriting_url(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + + about_url = "AbOuT:blank" + self.assertEqual(session.open(about_url)["url"], about_url) + + session.opener = Mock() + session.opener.open.side_effect = OSError("stop before network access") + with patch("fable_engine.browser.urllib.request.Request") as request: + session.open("HtTp://Example.test/Path") + + request.assert_called_once_with("HtTp://Example.test/Path") + + def test_bare_urls_use_https_except_for_loopback_targets(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + session.opener = Mock() + session.credential_free_opener = Mock() + session.opener.open.side_effect = OSError("stop before network access") + session.credential_free_opener.open.side_effect = OSError( + "stop before network access" + ) + + for target, expected in ( + ("example.test/path", "https://example.test/path"), + ("localhost:3000/path", "http://localhost:3000/path"), + ("service.localhost:3000/path", "https://service.localhost:3000/path"), + ("127.0.0.1:3000/path", "http://127.0.0.1:3000/path"), + ("[::1]:3000/path", "http://[::1]:3000/path"), + ("http://example.test/path", "http://example.test/path"), + ): + with self.subTest(target=target): + session.opener.reset_mock() + session.credential_free_opener.reset_mock() + session.open(target) + opener = ( + session.credential_free_opener + if expected.startswith("http://example.test") + else session.opener + ) + self.assertEqual(opener.open.call_args.args[0].full_url, expected) + + def test_actionable_elements_are_not_hidden_by_summary_limit(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + session.page_html = ( + "" + + "".join(f"

Heading {index}

" for index in range(60)) + + '' + + "" + ) + session._parse_and_layout() + + elements = session._build_status()["interactive_elements"] + + self.assertEqual(elements[0]["element_id"], "late-input") + self.assertLessEqual(len(elements), 50) + + def test_layout_walk_handles_deep_dom_in_preorder(self): + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + session.page_html = "".join( + f'
' for index in range(1_500) + ) + "
" * 1_500 + + session._parse_and_layout() + + ids = list(session.elements_by_id) + self.assertEqual(ids[:4], ["elem_0", "node-0", "node-1", "node-2"]) + self.assertEqual(ids[-1], "node-1499") + + def test_png_generator(self): + elems = [DOMElement("elem_1", "button", {"id": "elem_1"})] + elems[0].x = 10 + elems[0].y = 10 + elems[0].width = 100 + elems[0].height = 40 + + png_bytes = generate_minimal_png(200, 100, elems) + self.assertTrue(png_bytes.startswith(b"\x89PNG\r\n\x1a\n")) + self.assertTrue(b"IHDR" in png_bytes) + self.assertTrue(b"IDAT" in png_bytes) + self.assertTrue(b"IEND" in png_bytes) + + def test_png_background_scanlines_preserve_rgb_layout(self): + bg_color = (1, 127, 255) + png_bytes = generate_minimal_png(137, 103, [], bg_color=bg_color) + offset = 8 + compressed = bytearray() + while offset < len(png_bytes): + chunk_length = struct.unpack(">I", png_bytes[offset:offset + 4])[0] + chunk_type = png_bytes[offset + 4:offset + 8] + chunk_data = png_bytes[offset + 8:offset + 8 + chunk_length] + if chunk_type == b"IDAT": + compressed.extend(chunk_data) + offset += 12 + chunk_length + + expected_scanline = b"\x00" + bytes(bg_color) * 137 + self.assertEqual(zlib.decompress(compressed), expected_scanline * 103) + + def test_browser_engine_session_lifecycle(self): + engine = StealthBrowserEngine(profile_dir=self.profile_dir, max_sessions=2) + session1 = engine.get_or_create_session("tab1") + session2 = engine.get_or_create_session("tab2") + + self.assertIn("tab1", engine.sessions) + self.assertIn("tab2", engine.sessions) + self.assertIs(engine.get_or_create_session("tab1"), session1) + self.assertEqual(list(engine.sessions), ["tab1", "tab2"]) + + close_res = engine.close_session("tab1") + self.assertEqual(close_res["status"], "closed") + self.assertNotIn("tab1", engine.sessions) + + session2.page_html = "retained page" + session2.opener.close = Mock() + session2.credential_free_opener.close = Mock() + engine.get_or_create_session("tab3") + engine.get_or_create_session("tab4") + self.assertNotIn("tab2", engine.sessions) + self.assertEqual(list(engine.sessions), ["tab3", "tab4"]) + self.assertLessEqual(len(engine.sessions), MAX_BROWSER_SESSIONS) + session2.opener.close.assert_called_once_with() + session2.credential_free_opener.close.assert_called_once_with() + self.assertEqual(session2.page_html, "") + + def test_default_session_page_retention_has_a_safe_aggregate_limit(self): + self.assertEqual( + DEFAULT_MAX_RESPONSE_BYTES * MAX_BROWSER_SESSIONS, + MAX_RETAINED_PAGE_HTML_BYTES, + ) + self.assertLess(MAX_RETAINED_PAGE_HTML_BYTES, 20 * 1024 * 1024) + + def test_browser_module_import_does_not_create_profile(self): + script = """ +from unittest.mock import patch +with patch('pathlib.Path.mkdir', side_effect=PermissionError('read-only home')): + import fable_engine.browser as browser + assert browser.GLOBAL_BROWSER_ENGINE._engine is None +""" + env = os.environ.copy() + repo_root = str(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + env["PYTHONPATH"] = repo_root + os.pathsep + env.get("PYTHONPATH", "") + result = subprocess.run( + [sys.executable, "-c", script], + cwd=self.tmp_dir, + env=env, + text=True, + capture_output=True, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_browser_close_dispatch_does_not_create_unknown_session(self): + engine = Mock() + engine.close_session.return_value = {"status": "not_found", "session_id": "missing"} + request = json.dumps({ + "jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": {"name": "browser_close", "arguments": {"session_id": "missing"}}, + }) + "\n" + output = io.StringIO() + with ( + patch.object(browser_server, "GLOBAL_BROWSER_ENGINE", engine), + patch.object(browser_server, "AutoUpdater") as auto_updater, + patch("sys.stdin", io.StringIO(request)), + patch("sys.stdout", output), + ): + auto_updater.return_value.trigger_silent_background_update.return_value = None + browser_server.main() + + engine.get_or_create_session.assert_not_called() + engine.close_session.assert_called_once_with("missing") + result = json.loads(json.loads(output.getvalue())["result"]["content"][0]["text"]) + self.assertEqual(result["status"], "not_found") + + def test_unknown_browser_tool_does_not_create_session(self): + engine = Mock() + request = json.dumps({ + "jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": {"name": "browser_typo", "arguments": {"session_id": "unused"}}, + }) + "\n" + output = io.StringIO() + with ( + patch.object(browser_server, "GLOBAL_BROWSER_ENGINE", engine), + patch.object(browser_server, "AutoUpdater") as auto_updater, + patch("sys.stdin", io.StringIO(request)), + patch("sys.stdout", output), + ): + auto_updater.return_value.trigger_silent_background_update.return_value = None + browser_server.main() + + engine.get_or_create_session.assert_not_called() + engine.close_session.assert_not_called() + response = json.loads(output.getvalue()) + self.assertEqual(response["error"]["code"], -32601) + + def test_browser_open_timeout_is_bounded_and_must_be_finite(self): + engine = Mock() + session = engine.get_or_create_session.return_value + session.open.return_value = {"status": "opened"} + requests = "".join([ + json.dumps({ + "jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": { + "name": "browser_open", + "arguments": {"url": "about:blank", "timeout": 10_000}, + }, + }) + "\n", + json.dumps({ + "jsonrpc": "2.0", "id": 2, "method": "tools/call", + "params": { + "name": "browser_open", + "arguments": {"url": "about:blank", "timeout": float("nan")}, + }, + }) + "\n", + ]) + output = io.StringIO() + with ( + patch.object(browser_server, "GLOBAL_BROWSER_ENGINE", engine), + patch.object(browser_server, "AutoUpdater") as auto_updater, + patch("sys.stdin", io.StringIO(requests)), + patch("sys.stdout", output), + ): + auto_updater.return_value.trigger_silent_background_update.return_value = None + browser_server.main() + + session.open.assert_called_once_with( + "about:blank", timeout=MAX_BROWSER_OPEN_TIMEOUT_SECONDS + ) + engine.get_or_create_session.assert_called_once_with(None) + responses = [json.loads(line) for line in output.getvalue().splitlines()] + self.assertFalse(responses[0]["result"]["isError"]) + self.assertTrue(responses[1]["result"]["isError"]) + self.assertIn("must be finite", responses[1]["result"]["content"][0]["text"]) + + def test_browser_operation_error_results_set_mcp_error_flag(self): + engine = Mock() + session = engine.get_or_create_session.return_value + session.open.side_effect = [ + {"status": "error", "error": "navigation failed"}, + {"status": "opened", "url": "about:blank"}, + ] + session.click.return_value = {"error": "missing element"} + session.type_text.return_value = { + "status": "error", + "error": "not editable", + } + session.scroll.return_value = {"error": "scroll failed"} + requests = "".join( + json.dumps({ + "jsonrpc": "2.0", + "id": index, + "method": "tools/call", + "params": {"name": name, "arguments": arguments}, + }) + "\n" + for index, (name, arguments) in enumerate(( + ("browser_open", {"url": "bad.test"}), + ("browser_click", {"element_id": "missing"}), + ("browser_type", {"element_id": "readonly", "text": "value"}), + ("browser_scroll", {"delta_y": 10}), + ("browser_open", {"url": "about:blank"}), + ), start=1) + ) + output = io.StringIO() + with ( + patch.object(browser_server, "GLOBAL_BROWSER_ENGINE", engine), + patch.object(browser_server, "AutoUpdater") as auto_updater, + patch("sys.stdin", io.StringIO(requests)), + patch("sys.stdout", output), + ): + auto_updater.return_value.trigger_silent_background_update.return_value = None + browser_server.main() + + responses = [json.loads(line) for line in output.getvalue().splitlines()] + self.assertEqual( + [response["result"]["isError"] for response in responses], + [True, True, True, True, False], + ) + + def test_browser_open_enforces_one_deadline_across_reads(self): + class FakeSocket: + def __init__(self): + self.timeouts = [] + + def settimeout(self, timeout): + self.timeouts.append(timeout) + + class FakeResponse: + def __init__(self): + fake_raw = type("FakeRaw", (), {"_sock": FakeSocket()})() + self.fp = type("FakeFP", (), {"raw": fake_raw})() + self.chunks = iter((b"partial", b"response", b"")) + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def read1(self, _size): + return next(self.chunks) + + def read(self, _size): + return next(self.chunks) + + pm = ProfileManager(profile_dir=self.profile_dir) + session = StealthBrowserSession("test_tab", pm) + response = FakeResponse() + session.opener = Mock() + session.opener.open.return_value = response + + with patch( + "fable_engine.browser.time.monotonic", + side_effect=(100.0, 100.1, 100.2, 101.1), + ): + result = session.open("example.test", timeout=1.0) + + self.assertEqual(result["status"], "error") + self.assertIn("timed out", result["error"]) + self.assertAlmostEqual(session.opener.open.call_args.kwargs["timeout"], 0.9) + self.assertEqual(len(response.fp.raw._sock.timeouts), 1) + + def test_browser_wait_reports_bounded_duration(self): + engine = Mock() + request = json.dumps({ + "jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": { + "name": "browser_wait", + "arguments": {"seconds": 25, "session_id": "new-session"}, + }, + }) + "\n" + output = io.StringIO() + with ( + patch.object(browser_server, "GLOBAL_BROWSER_ENGINE", engine), + patch.object(browser_server, "AutoUpdater") as auto_updater, + patch("sys.stdin", io.StringIO(request)), + patch("sys.stdout", output), + patch("time.sleep") as sleep, + ): + auto_updater.return_value.trigger_silent_background_update.return_value = None + browser_server.main() + + sleep.assert_called_once_with(10) + engine.get_or_create_session.assert_not_called() + engine.close_session.assert_not_called() + result = json.loads(json.loads(output.getvalue())["result"]["content"][0]["text"]) + self.assertEqual(result["seconds"], 10) + + +if __name__ == "__main__": + unittest.main()