4b8c01f691
Camofox VNC one-shot, computer-use aux-vision verdict, tirith binary path, MCP discovery lock path, remote-backend probe text, learned image token costs, auxiliary per-task semaphores and the custom-endpoint /models memo all held one profile's config-derived value for the whole process. The skill-sync debounce Timer ran with empty ContextVars, so a secondary's write pushed as the launch profile (and cancelled its pending push). Each memo is now keyed by hermes_home_key() (or credential fingerprint for the per-key catalog) under an override; the timer is per home and runs its callback inside the scheduling turn's copied context. Unscoped slots are unchanged.
602 lines
28 KiB
Python
602 lines
28 KiB
Python
"""Camofox browser backend — local anti-detection browser via REST API.
|
|
|
|
Camofox-browser (https://github.com/jo-inc/camofox-browser) is a self-hosted Node.js
|
|
server wrapping Camoufox (Firefox fork with C++ fingerprint spoofing); its REST API maps
|
|
1:1 to our browser tool interface (accessibility snapshots with element refs, click/type/
|
|
scroll by ref, screenshots). Setup: ``npm start`` in a checkout or ``docker run -p 9377:9377
|
|
-e CAMOFOX_PORT=9377 jo-inc/camofox-browser``, then ``CAMOFOX_URL=http://localhost:9377`` in
|
|
``~/.hermes/.env`` (Docker: see ``CAMOFOX_REWRITE_LOOPBACK_URLS`` below).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import ipaddress
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import threading
|
|
import uuid
|
|
from typing import Any, Callable, Dict, Optional
|
|
from urllib.parse import SplitResult, urlsplit, urlunsplit
|
|
|
|
import requests
|
|
|
|
from agent.secret_scope import get_secret
|
|
from hermes_cli.config import cfg_get, load_config, read_raw_config
|
|
from hermes_constants import get_hermes_home_override, hermes_home_key
|
|
from tools.browser_camofox_state import get_camofox_identity
|
|
from tools.registry import tool_error
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---- Configuration ----
|
|
|
|
_DEFAULT_TIMEOUT = 30 # fallback when config is unreadable
|
|
_NO_SESSION_ERROR = "No browser session. Call browser_navigate first."
|
|
_vnc_url: Optional[str] = None # cached from /health response
|
|
_vnc_url_checked = False # only probe once per process
|
|
# Routed profiles (multiplexed gateway) each point CAMOFOX_URL at their own server, so the one-shot
|
|
# slot above would hand the launch profile's VNC address to every other profile: memo per server URL.
|
|
_vnc_url_by_camofox_url: Dict[str, Optional[str]] = {}
|
|
# browser.command_timeout, resolved lazily like browser_tool; keyed by profile home because the
|
|
# multiplexed gateway serves every profile from one process.
|
|
_cached_cmd_timeout: Optional[Dict[str, int]] = None
|
|
_cmd_timeout_resolved = False
|
|
|
|
|
|
def _get_command_timeout() -> int:
|
|
"""``browser.command_timeout`` (floor 5s, default 30s), cached per profile home after first read."""
|
|
global _cached_cmd_timeout, _cmd_timeout_resolved
|
|
home = hermes_home_key()
|
|
if _cached_cmd_timeout is None:
|
|
_cached_cmd_timeout = {}
|
|
if _cmd_timeout_resolved and home in _cached_cmd_timeout:
|
|
return _cached_cmd_timeout[home]
|
|
timeout = _DEFAULT_TIMEOUT
|
|
try:
|
|
val = cfg_get(read_raw_config(), "browser", "command_timeout")
|
|
if val is not None:
|
|
timeout = max(int(val), 5)
|
|
except Exception as exc:
|
|
logger.debug("Could not read browser.command_timeout: %s", exc)
|
|
_cached_cmd_timeout[home] = timeout
|
|
_cmd_timeout_resolved = True
|
|
return timeout
|
|
|
|
|
|
def _auth_headers() -> Dict[str, str]:
|
|
"""Return Authorization header when CAMOFOX_API_KEY is set."""
|
|
key = (get_secret("CAMOFOX_API_KEY", "") or "").strip()
|
|
return {"Authorization": f"Bearer {key}"} if key else {}
|
|
|
|
|
|
def get_camofox_url() -> str:
|
|
"""Return the configured Camofox server URL, or empty string."""
|
|
return (get_secret("CAMOFOX_URL", "") or "").rstrip("/")
|
|
|
|
|
|
def _config_cdp_url() -> str:
|
|
"""Persistent ``browser.cdp_url`` from config.yaml, or "" (read here, not via
|
|
``browser_tool_cdp._get_cdp_override`` — circular import)."""
|
|
try:
|
|
from hermes_cli.config import read_raw_config # late-bound: tests patch the source module
|
|
browser_cfg = read_raw_config().get("browser", {})
|
|
if isinstance(browser_cfg, dict):
|
|
return str(browser_cfg.get("cdp_url", "") or "").strip()
|
|
except Exception:
|
|
pass
|
|
return ""
|
|
|
|
|
|
def is_camofox_mode() -> bool:
|
|
"""True when the Camofox backend is selected and no CDP override is active.
|
|
|
|
Selection is ``browser.cloud_provider: camofox``; ``CAMOFOX_URL`` is only the address
|
|
and never overrides a different stored selection (legacy: with no selection ever
|
|
written, a set ``CAMOFOX_URL`` still activates Camofox). A CDP override (``BROWSER_CDP_URL``
|
|
env or ``browser.cdp_url``, same precedence as ``browser_tool_cdp._get_cdp_override()``) wins.
|
|
"""
|
|
if os.getenv("BROWSER_CDP_URL", "").strip() or _config_cdp_url():
|
|
return False
|
|
try:
|
|
from tools.tool_backend_helpers import read_selection
|
|
selected = read_selection("browser")
|
|
except Exception: # pragma: no cover — helpers are in-repo
|
|
selected = None
|
|
if selected is not None:
|
|
return selected == "camofox"
|
|
return bool(get_camofox_url())
|
|
|
|
|
|
def _vnc_url_from_health(url: str, resp: Any) -> Optional[str]:
|
|
try:
|
|
vnc_port = resp.json().get("vncPort")
|
|
if isinstance(vnc_port, int) and 1 <= vnc_port <= 65535:
|
|
return f"http://{urlsplit(url).hostname or 'localhost'}:{vnc_port}"
|
|
except (ValueError, KeyError):
|
|
pass
|
|
return None
|
|
|
|
|
|
def check_camofox_available() -> bool:
|
|
"""Verify the Camofox server is reachable (and cache its VNC URL once)."""
|
|
global _vnc_url, _vnc_url_checked
|
|
url = get_camofox_url()
|
|
if not url:
|
|
return False
|
|
try:
|
|
resp = requests.get(f"{url}/health", timeout=5)
|
|
except Exception:
|
|
return False
|
|
if resp.status_code == 200:
|
|
if get_hermes_home_override() is not None:
|
|
if url not in _vnc_url_by_camofox_url:
|
|
_vnc_url_by_camofox_url[url] = _vnc_url_from_health(url, resp)
|
|
elif not _vnc_url_checked:
|
|
_vnc_url = _vnc_url_from_health(url, resp) or _vnc_url
|
|
_vnc_url_checked = True
|
|
return resp.status_code == 200
|
|
|
|
|
|
def get_vnc_url() -> Optional[str]:
|
|
"""Return the VNC URL if the Camofox server exposes one, or None."""
|
|
if get_hermes_home_override() is not None:
|
|
url = get_camofox_url()
|
|
if url not in _vnc_url_by_camofox_url:
|
|
check_camofox_available()
|
|
return _vnc_url_by_camofox_url.get(url)
|
|
if not _vnc_url_checked:
|
|
check_camofox_available()
|
|
return _vnc_url
|
|
|
|
|
|
def _get_camofox_config() -> Dict[str, Any]:
|
|
"""Return the ``browser.camofox`` config block, or an empty dict."""
|
|
try:
|
|
camofox_cfg = load_config().get("browser", {}).get("camofox", {})
|
|
except Exception as exc:
|
|
logger.warning("camofox config check failed, defaulting to disabled: %s", exc)
|
|
return {}
|
|
return camofox_cfg if isinstance(camofox_cfg, dict) else {}
|
|
|
|
|
|
def _managed_persistence_enabled(camofox_cfg: Optional[Dict[str, Any]] = None) -> bool:
|
|
"""``browser.camofox.managed_persistence``: stable profile-scoped userId vs random per session."""
|
|
return bool((_get_camofox_config() if camofox_cfg is None else camofox_cfg).get("managed_persistence"))
|
|
|
|
|
|
def _env_or_cfg(env_name: str, camofox_cfg: Dict[str, Any], cfg_key: str, *, secret: bool = False) -> str:
|
|
"""Env/secret-scope value first, then the ``browser.camofox`` config key, else ""."""
|
|
raw = get_secret(env_name, "") if secret else os.getenv(env_name, "")
|
|
return (raw or "").strip() or str(camofox_cfg.get(cfg_key) or "").strip()
|
|
|
|
|
|
def _camofox_identity_override(task_id: Optional[str], camofox_cfg: Dict[str, Any]) -> Optional[Dict[str, str]]:
|
|
"""Externally configured identity (integrations owning the visible Camofox browser
|
|
share a user ID so Hermes uses the same profile), or None."""
|
|
user_id = _env_or_cfg("CAMOFOX_USER_ID", camofox_cfg, "user_id", secret=True)
|
|
if not user_id:
|
|
return None
|
|
session_key = _env_or_cfg("CAMOFOX_SESSION_KEY", camofox_cfg, "session_key", secret=True)
|
|
return {"user_id": user_id, "session_key": session_key or f"task_{(task_id or 'default')[:16]}"}
|
|
|
|
|
|
def _flag(env_name: str, camofox_cfg: Dict[str, Any], cfg_key: str) -> bool:
|
|
"""Boolean toggle: env var wins when set to a valid value, else config key."""
|
|
raw = os.getenv(env_name, "").strip().lower()
|
|
if raw in {"1", "true", "yes", "on"}:
|
|
return True
|
|
if raw in {"0", "false", "no", "off"}:
|
|
return False
|
|
if raw:
|
|
logger.debug("Ignoring invalid boolean env %s=%r", env_name, raw)
|
|
return bool(camofox_cfg.get(cfg_key))
|
|
|
|
|
|
def _is_loopback_hostname(hostname: Optional[str]) -> bool:
|
|
if not hostname:
|
|
return False
|
|
host = hostname.strip().strip("[]").lower()
|
|
if host in {"localhost", "localhost.localdomain"}:
|
|
return True
|
|
try:
|
|
return ipaddress.ip_address(host).is_loopback
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _rewrite_loopback_url_for_camofox(url: str) -> tuple[str, Optional[Dict[str, str]]]:
|
|
"""Rewrite loopback page URLs for Docker-hosted Camofox, if configured.
|
|
|
|
``CAMOFOX_URL`` may point at a host-published Docker port, but page URLs are opened by
|
|
the browser *inside* the container, where loopback is the container, not the host.
|
|
Opt-in (``CAMOFOX_REWRITE_LOOPBACK_URLS`` / config) because non-Docker installs run the
|
|
browser on the host. Returns ``(rewritten_url, metadata)``; ``metadata`` is present only
|
|
when a rewrite happened so the tool result can disclose the change to the model.
|
|
"""
|
|
camofox_cfg = _get_camofox_config()
|
|
if not _flag("CAMOFOX_REWRITE_LOOPBACK_URLS", camofox_cfg, "rewrite_loopback_urls"):
|
|
return url, None
|
|
try:
|
|
parsed = urlsplit(url)
|
|
except ValueError:
|
|
return url, None
|
|
alias = _env_or_cfg("CAMOFOX_LOOPBACK_HOST_ALIAS", camofox_cfg, "loopback_host_alias") or "host.docker.internal"
|
|
if parsed.scheme not in {"http", "https"} or not _is_loopback_hostname(parsed.hostname) or not alias:
|
|
return url, None
|
|
userinfo = (parsed.username + (f":{parsed.password}" if parsed.password else "") + "@") if parsed.username else ""
|
|
host_part = f"[{alias}]" if ":" in alias and not alias.startswith("[") else alias
|
|
port_part = f":{parsed.port}" if parsed.port else ""
|
|
rewritten = urlunsplit(
|
|
SplitResult(parsed.scheme, f"{userinfo}{host_part}{port_part}", parsed.path, parsed.query, parsed.fragment))
|
|
return rewritten, {"from": parsed.hostname or "", "to": alias, "original_url": url, "rewritten_url": rewritten}
|
|
|
|
|
|
# ---- Session management ----
|
|
_sessions: Dict[str, Dict[str, Any]] = {} # task_id -> {"user_id": str, "tab_id": str|None, ...}
|
|
_sessions_lock = threading.Lock()
|
|
|
|
|
|
def _adopt_existing_tab(session: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Rehydrate tab_id from an already-open managed tab: gateway restarts empty the
|
|
in-memory cache while Camofox still holds the integration-owned tab."""
|
|
if session.get("tab_id") or not session.get("adopt_existing_tab") or not get_camofox_url():
|
|
return session
|
|
try:
|
|
tabs = _get("/tabs", params=_user_params(session), timeout=5).get("tabs", [])
|
|
except Exception as exc:
|
|
logger.debug("Camofox tab adoption failed for %s: %s", session.get("user_id"), exc)
|
|
return session
|
|
dict_tabs = [tab for tab in tabs if isinstance(tab, dict)] if isinstance(tabs, list) else []
|
|
candidates = [tab for tab in dict_tabs if tab.get("listItemId") == session.get("session_key")] or dict_tabs
|
|
tab_id = candidates[-1].get("tabId") if candidates else None
|
|
if isinstance(tab_id, str) and tab_id:
|
|
session["tab_id"] = tab_id
|
|
logger.debug("Adopted existing Camofox tab %s for %s", tab_id, session.get("user_id"))
|
|
return session
|
|
|
|
|
|
def _get_session(task_id: Optional[str]) -> Dict[str, Any]:
|
|
"""Get or create the task's session. Identity precedence: external override
|
|
(CAMOFOX_USER_ID / config) → profile-scoped identity when managed persistence
|
|
is on → random ephemeral userId."""
|
|
task_id = task_id or "default"
|
|
with _sessions_lock:
|
|
if task_id in _sessions:
|
|
return _adopt_existing_tab(_sessions[task_id])
|
|
camofox_cfg = _get_camofox_config()
|
|
identity = _camofox_identity_override(task_id, camofox_cfg)
|
|
if identity is None and _managed_persistence_enabled(camofox_cfg):
|
|
identity = get_camofox_identity(task_id)
|
|
if identity is None:
|
|
identity = {"user_id": f"hermes_{uuid.uuid4().hex[:10]}", "session_key": f"task_{task_id[:16]}"}
|
|
managed, adopt = False, False
|
|
else:
|
|
managed, adopt = True, _flag("CAMOFOX_ADOPT_EXISTING_TAB", camofox_cfg, "adopt_existing_tab")
|
|
session = {"user_id": identity["user_id"], "tab_id": None, "session_key": identity["session_key"],
|
|
"managed": managed, "adopt_existing_tab": adopt}
|
|
_sessions[task_id] = session
|
|
return _adopt_existing_tab(session)
|
|
|
|
|
|
def _ensure_tab(task_id: Optional[str], url: str = "about:blank") -> Dict[str, Any]:
|
|
"""Ensure a tab exists for the session, creating one if needed."""
|
|
session = _get_session(task_id)
|
|
if not session["tab_id"]:
|
|
data = _post("/tabs", {"userId": session["user_id"], "listItemId": session["session_key"], "url": url})
|
|
session["tab_id"] = data.get("tabId")
|
|
return session
|
|
|
|
|
|
def _drop_session(task_id: Optional[str]) -> Optional[Dict[str, Any]]:
|
|
"""Remove and return session info."""
|
|
with _sessions_lock:
|
|
return _sessions.pop(task_id or "default", None)
|
|
|
|
|
|
def camofox_soft_cleanup(task_id: Optional[str] = None) -> bool:
|
|
"""Drop only the local tracking entry (``True``) for managed profiles, which must
|
|
survive across agent tasks; ``False`` for ephemeral sessions so the caller falls back
|
|
to :func:`camofox_close`."""
|
|
camofox_cfg = _get_camofox_config()
|
|
if _managed_persistence_enabled(camofox_cfg) or _camofox_identity_override(task_id, camofox_cfg):
|
|
_drop_session(task_id)
|
|
logger.debug("Camofox soft cleanup for task %s (managed persistence)", task_id)
|
|
return True
|
|
return False
|
|
|
|
|
|
# ---- HTTP helpers ----
|
|
def _request(method: str, path: str, timeout: Optional[int] = None, **kwargs: Any) -> requests.Response:
|
|
"""Issue an authenticated request to camofox and return the raised-for-status response."""
|
|
resp = getattr(requests, method)(f"{get_camofox_url()}{path}", headers=_auth_headers(),
|
|
timeout=_get_command_timeout() if timeout is None else timeout, **kwargs)
|
|
resp.raise_for_status()
|
|
return resp
|
|
|
|
|
|
def _post(path: str, body: dict, timeout: Optional[int] = None) -> dict:
|
|
return _request("post", path, timeout, json=body).json()
|
|
|
|
|
|
def _get(path: str, params: dict = None, timeout: Optional[int] = None) -> dict:
|
|
return _request("get", path, timeout, params=params).json()
|
|
|
|
|
|
def _get_raw(path: str, params: dict = None, timeout: Optional[int] = None) -> requests.Response:
|
|
"""GET and return the raw response (for binary data)."""
|
|
return _request("get", path, timeout, params=params)
|
|
|
|
|
|
def _delete(path: str, body: dict = None, timeout: Optional[int] = None) -> dict:
|
|
return _request("delete", path, timeout, json=body).json()
|
|
|
|
|
|
# ---- Tool implementations ----
|
|
def _tab_path(session: Dict[str, Any], suffix: str) -> str:
|
|
return f"/tabs/{session['tab_id']}/{suffix}"
|
|
|
|
|
|
def _user_params(session: Dict[str, Any]) -> Dict[str, str]:
|
|
return {"userId": session["user_id"]}
|
|
|
|
|
|
def _snapshot_data(session: Dict[str, Any]) -> dict:
|
|
return _get(_tab_path(session, "snapshot"), params=_user_params(session))
|
|
|
|
|
|
def _parse_snapshot_images(snapshot: str) -> list[Dict[str, str]]:
|
|
"""Images from an accessibility snapshot: ``img "alt" [eN]`` entries with the URL on
|
|
the following ``/url:`` line (Camofox has no /images endpoint)."""
|
|
images = []
|
|
lines = snapshot.split("\n")
|
|
for i, line in enumerate(lines):
|
|
stripped = line.strip()
|
|
if not stripped.startswith(("- img ", "img ")):
|
|
continue
|
|
alt_match = re.search(r'img\s+"([^"]*)"', stripped)
|
|
url_match = re.search(r'/url:\s*(\S+)', lines[i + 1].strip()) if i + 1 < len(lines) else None
|
|
alt, src = (alt_match.group(1) if alt_match else ""), (url_match.group(1) if url_match else "")
|
|
if alt or src:
|
|
images.append({"src": src, "alt": alt})
|
|
return images
|
|
|
|
|
|
def _fetch_snapshot(session: Dict[str, Any]) -> tuple[str, int]:
|
|
"""``(snapshot_text, refs_count)`` truncated like the main browser tool (line boundaries,
|
|
full tree stored to cache/web, read_file pointer appended). Lazy import: ``browser_tool``
|
|
imports this module."""
|
|
from tools.browser_tool_snapshot import _truncate_snapshot
|
|
from tools.browser_tool import get_browser_snapshot_threshold
|
|
data = _snapshot_data(session)
|
|
snapshot, threshold = data.get("snapshot", ""), get_browser_snapshot_threshold()
|
|
if len(snapshot) > threshold:
|
|
snapshot = _truncate_snapshot(snapshot, max_chars=threshold)
|
|
return snapshot, data.get("refsCount", 0)
|
|
|
|
|
|
def _navigate_tab(task_id: Optional[str], browser_url: str) -> tuple[Dict[str, Any], dict]:
|
|
"""Open ``browser_url`` in the task's tab (creating it if missing) and return
|
|
``(session, navigate_response)``. A 404 on the existing tab means the server
|
|
garbage-collected it — recreate instead of failing."""
|
|
session = _get_session(task_id)
|
|
if session["tab_id"]:
|
|
try:
|
|
data = _post(_tab_path(session, "navigate"), {"userId": session["user_id"], "url": browser_url}, timeout=60)
|
|
return session, data
|
|
except requests.HTTPError as e:
|
|
if e.response is None or e.response.status_code != 404:
|
|
raise
|
|
logger.warning("Camofox tab %s returned 404 — tab was garbage collected. Creating a fresh tab.",
|
|
session["tab_id"])
|
|
session["tab_id"] = None
|
|
return _ensure_tab(task_id, browser_url), {"ok": True, "url": browser_url}
|
|
|
|
|
|
def camofox_navigate(url: str, task_id: Optional[str] = None) -> str:
|
|
"""Navigate to a URL via Camofox."""
|
|
try:
|
|
browser_url, rewrite_info = _rewrite_loopback_url_for_camofox(url)
|
|
session, data = _navigate_tab(task_id, browser_url)
|
|
result = {"success": True, "url": data.get("url", browser_url), "title": data.get("title", "")}
|
|
if rewrite_info:
|
|
result["requested_url"], result["url_rewrite"] = url, rewrite_info
|
|
result["warning"] = ("Rewrote loopback URL for Docker-hosted Camofox: "
|
|
f"{rewrite_info['from']} -> {rewrite_info['to']}")
|
|
vnc = get_vnc_url()
|
|
if vnc:
|
|
result["vnc_url"] = vnc
|
|
result["vnc_hint"] = ("Browser is visible via VNC. "
|
|
"Share this link with the user so they can watch the browser live.")
|
|
try: # Auto-take a compact snapshot so the model can act immediately.
|
|
result["snapshot"], result["element_count"] = _fetch_snapshot(session)
|
|
except Exception:
|
|
pass # Navigation succeeded; snapshot is a bonus
|
|
return json.dumps(result)
|
|
except requests.HTTPError as e:
|
|
return tool_error(f"Navigation failed: {e}", success=False)
|
|
except requests.ConnectionError:
|
|
return json.dumps({"success": False, "error": (
|
|
f"Cannot connect to Camofox at {get_camofox_url()}. "
|
|
"Is the server running? Start with: npm start (in camofox-browser dir) "
|
|
"or: docker run -p 9377:9377 -e CAMOFOX_PORT=9377 jo-inc/camofox-browser")})
|
|
except Exception as e:
|
|
return tool_error(str(e), success=False)
|
|
|
|
|
|
def _camofox_private_page_block(session: Dict[str, Any], task_id: Optional[str], action: str) -> Optional[str]:
|
|
"""Blocked payload when the current page is private/internal, else None.
|
|
|
|
Mirrors the ``_camofox_eval`` guard in browser_tool.py: page-state reads on a non-local
|
|
backend can leak an intranet/metadata page the terminal can't reach. Only active when
|
|
the SSRF guard applies (non-local backend, not a local sidecar, ``allow_private_urls``
|
|
unset); fail-open on probe failure like sibling guards. Lazy import (cycle).
|
|
"""
|
|
from tools.browser_tool_eval_policy import _camofox_current_page_private_url, _eval_ssrf_guard_active
|
|
if not _eval_ssrf_guard_active(task_id or "default"):
|
|
return None
|
|
blocked_url = _camofox_current_page_private_url(session["tab_id"], session["user_id"])
|
|
if not blocked_url:
|
|
return None
|
|
return json.dumps({"success": False, "error": (
|
|
"Blocked: page URL targets a private or internal address "
|
|
f"({blocked_url}). Refusing to {action} on this page in this browser mode.")}, ensure_ascii=False)
|
|
|
|
|
|
def _require_tab(task_id: Optional[str], action: Optional[str] = None) -> tuple[Dict[str, Any], Optional[str]]:
|
|
"""Return ``(session, error_payload)``: error when no tab exists or, if ``action`` given, the page is private."""
|
|
session = _get_session(task_id)
|
|
if not session["tab_id"]:
|
|
return session, tool_error(_NO_SESSION_ERROR, success=False)
|
|
return session, (_camofox_private_page_block(session, task_id, action) if action is not None else None)
|
|
|
|
|
|
def _with_tab(task_id: Optional[str], guard_action: Optional[str], body: Callable[[Dict[str, Any]], str]) -> str:
|
|
"""Require a tab (+ private-page guard when ``guard_action`` is set), then run ``body(session)``;
|
|
any exception becomes a ``tool_error``."""
|
|
try:
|
|
session, blocked = _require_tab(task_id, guard_action)
|
|
if blocked:
|
|
return blocked
|
|
return body(session)
|
|
except Exception as e:
|
|
return tool_error(str(e), success=False)
|
|
|
|
|
|
def _tab_action(task_id: Optional[str], guard_action: Optional[str], suffix: str,
|
|
body: Dict[str, Any], result: Callable[[dict], dict]) -> str:
|
|
"""Simple tab action: POST ``body`` to ``/tabs/<id>/<suffix>``, build the result."""
|
|
return _with_tab(task_id, guard_action, lambda session: json.dumps(
|
|
result(_post(_tab_path(session, suffix), {"userId": session["user_id"], **body}))))
|
|
|
|
|
|
def camofox_snapshot(full: bool = False, task_id: Optional[str] = None, user_task: Optional[str] = None) -> str:
|
|
"""Accessibility tree snapshot. ``user_task`` is deprecated and ignored —
|
|
oversized snapshots always truncate-and-store (no LLM summarization)."""
|
|
def body(session):
|
|
snapshot, refs_count = _fetch_snapshot(session)
|
|
return json.dumps({"success": True, "snapshot": snapshot, "element_count": refs_count})
|
|
return _with_tab(task_id, "read a page snapshot", body)
|
|
|
|
|
|
def camofox_click(ref: str, task_id: Optional[str] = None) -> str:
|
|
"""Click an element by ref via Camofox."""
|
|
clean_ref = ref.lstrip("@") # our tool convention prefixes refs with @
|
|
return _tab_action(task_id, "click", "click", {"ref": clean_ref},
|
|
lambda data: {"success": True, "clicked": clean_ref, "url": data.get("url", "")})
|
|
|
|
|
|
def camofox_type(ref: str, text: str, task_id: Optional[str] = None) -> str:
|
|
"""Type text into an element by ref via Camofox."""
|
|
try:
|
|
session, blocked = _require_tab(task_id, "type")
|
|
if blocked:
|
|
return blocked
|
|
clean_ref = ref.lstrip("@")
|
|
_post(_tab_path(session, "type"), {"userId": session["user_id"], "ref": clean_ref, "text": text})
|
|
from agent.display import redact_browser_typed_text_for_display, redact_tool_args_for_display
|
|
# Match browser_tool.browser_type: the raw text is typed into the page, but the
|
|
# returned display value is run through the secret-pattern redactor so API keys /
|
|
# tokens don't leak into tool progress or chat history.
|
|
display_text = (redact_tool_args_for_display("browser_type", {"text": text}) or {})["text"]
|
|
response = {"success": True, "typed": display_text, "element": clean_ref}
|
|
return json.dumps(redact_browser_typed_text_for_display(response, text))
|
|
except Exception as e:
|
|
from agent.display import redact_browser_typed_text_for_display
|
|
return tool_error(redact_browser_typed_text_for_display(str(e), text), success=False)
|
|
|
|
|
|
def camofox_scroll(direction: str, task_id: Optional[str] = None) -> str:
|
|
"""Scroll the page via Camofox."""
|
|
return _tab_action(task_id, None, "scroll", {"direction": direction},
|
|
lambda data: {"success": True, "scrolled": direction})
|
|
|
|
|
|
def camofox_back(task_id: Optional[str] = None) -> str:
|
|
"""Navigate back via Camofox."""
|
|
return _tab_action(task_id, None, "back", {}, lambda data: {"success": True, "url": data.get("url", "")})
|
|
|
|
|
|
def camofox_press(key: str, task_id: Optional[str] = None) -> str:
|
|
"""Press a keyboard key via Camofox."""
|
|
return _tab_action(task_id, "press", "press", {"key": key}, lambda data: {"success": True, "pressed": key})
|
|
|
|
|
|
def camofox_close(task_id: Optional[str] = None) -> str:
|
|
"""Close the browser session via Camofox."""
|
|
try:
|
|
session = _drop_session(task_id)
|
|
if session:
|
|
_delete(f"/sessions/{session['user_id']}")
|
|
return json.dumps({"success": True, "closed": True})
|
|
except Exception as e:
|
|
return json.dumps({"success": True, "closed": True, "warning": str(e)})
|
|
|
|
|
|
def camofox_get_images(task_id: Optional[str] = None) -> str:
|
|
"""Get images on the current page via Camofox (parsed from the snapshot)."""
|
|
def body(session):
|
|
images = _parse_snapshot_images(_snapshot_data(session).get("snapshot", ""))
|
|
return json.dumps({"success": True, "images": images, "count": len(images)})
|
|
return _with_tab(task_id, "extract page images", body)
|
|
|
|
|
|
def _vision_llm_settings() -> tuple[float, float]:
|
|
"""``auxiliary.vision`` ``(timeout, temperature)``; defaults 120s / 0.1 on any config error."""
|
|
try:
|
|
cfg = cfg_get(load_config(), "auxiliary", "vision", default={})
|
|
return float(cfg.get("timeout", 120)), float(cfg.get("temperature", 0.1))
|
|
except Exception:
|
|
return 120.0, 0.1
|
|
|
|
|
|
def _save_screenshot(content: bytes) -> str:
|
|
"""Write PNG bytes under ``$HERMES_HOME/browser_screenshots`` and return the path."""
|
|
from hermes_constants import get_hermes_home
|
|
screenshots_dir = get_hermes_home() / "browser_screenshots"
|
|
screenshots_dir.mkdir(parents=True, exist_ok=True)
|
|
screenshot_path = str(screenshots_dir / f"browser_screenshot_{uuid.uuid4().hex[:8]}.png")
|
|
with open(screenshot_path, "wb") as f:
|
|
f.write(content)
|
|
return screenshot_path
|
|
|
|
|
|
def camofox_vision(question: str, annotate: bool = False, task_id: Optional[str] = None) -> str:
|
|
"""Take a screenshot and analyze it with vision AI via Camofox."""
|
|
def body(session):
|
|
resp = _get_raw(_tab_path(session, "screenshot"), params=_user_params(session))
|
|
screenshot_path = _save_screenshot(resp.content)
|
|
img_b64 = base64.b64encode(resp.content).decode("utf-8")
|
|
annotation_context = ""
|
|
if annotate:
|
|
try:
|
|
snapshot = _snapshot_data(session).get("snapshot", "")
|
|
annotation_context = f"\n\nAccessibility tree (element refs for interaction):\n{snapshot[:3000]}"
|
|
except Exception:
|
|
pass
|
|
# The screenshot itself cannot be redacted, but the text-based accessibility snippet
|
|
# sent alongside it must not leak secret values.
|
|
from agent.redact import redact_sensitive_text
|
|
from agent.auxiliary_client import call_llm
|
|
vision_prompt = f"Analyze this browser screenshot and answer: {question}{redact_sensitive_text(annotation_context)}"
|
|
timeout, temperature = _vision_llm_settings()
|
|
response = call_llm(
|
|
messages=[{"role": "user", "content": [
|
|
{"type": "text", "text": vision_prompt},
|
|
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_b64}"}}]}],
|
|
task="vision", temperature=temperature, timeout=timeout)
|
|
analysis = (response.choices[0].message.content or "").strip() if response.choices else ""
|
|
# Redact secrets the vision LLM may have read from the screenshot.
|
|
return json.dumps({"success": True, "analysis": redact_sensitive_text(analysis), "screenshot_path": screenshot_path})
|
|
return _with_tab(task_id, "capture a screenshot", body)
|
|
|
|
|
|
def camofox_console(clear: bool = False, task_id: Optional[str] = None) -> str:
|
|
"""Console output is not exposed by the Camofox REST API; return an empty result with a note."""
|
|
return json.dumps({
|
|
"success": True, "console_messages": [], "js_errors": [], "total_messages": 0, "total_errors": 0,
|
|
"note": "Console log capture is not available with the Camofox backend. "
|
|
"Use browser_snapshot or browser_vision to inspect page state."})
|