ling/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-04-28 06:51:16 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(browser): auto-spawn local Chromium for LAN/localhost URLs in cloud mode (#16136)

Browse Source 
When a cloud browser provider (Browserbase / Browser-Use / Firecrawl) is
configured, browser_navigate now transparently spawns a local Chromium
sidecar for URLs whose host resolves to a private/loopback/LAN address
(localhost, 127.0.0.1, 192.168.x.x, 10.x.x.x, *.local, *.lan, *.internal,
::1, 169.254.x.x). Public URLs continue to use the cloud provider in the
same conversation.

Previously, setting BROWSERBASE_API_KEY / cloud_provider: browserbase
pinned the whole tool to cloud for the process — localhost URLs were
either SSRF-blocked (default) or sent to Browserbase (where they 404'd
because the cloud can't reach your LAN). Users who wanted 'cloud for
public, local for localhost' had no way to express it short of toggling
providers mid-session.

Implementation uses a composite session key scheme: the bare task_id
serves the cloud session, and a '{task_id}::local' sidecar serves the
local Chromium. _last_active_session_key[task_id] tracks which of the
two served the most recent nav so snapshot/click/fill/etc. hit the
correct one. cleanup_browser(bare_task_id) reaps both.

Feature is on by default. Opt out via:
  browser:
    auto_local_for_private_urls: false

The cloud provider never sees private URLs. Post-redirect SSRF guard
is preserved: redirects from public onto private addresses still block.
This commit is contained in:
Teknium
2026-04-26 09:57:58 -07:00
committed by GitHub
parent 0e2a53eab2
commit 42c076d349
4 changed files with 563 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
hermes_cli/config.py
View File

@@ -465,6 +465,7 @@ DEFAULT_CONFIG = {
"command_timeout": 30, # Timeout for browser commands in seconds (screenshot, navigate, etc.)
"record_sessions": False, # Auto-record browser sessions as WebM videos
"allow_private_urls": False, # Allow navigating to private/internal IPs (localhost, 192.168.x.x, etc.)
"auto_local_for_private_urls": True, # When a cloud provider is set, auto-spawn local Chromium for LAN/localhost URLs instead of sending them to the cloud
"cdp_url": "", # Optional persistent CDP endpoint for attaching to an existing Chromium/Chrome
# CDP supervisor — dialog + frame detection via a persistent WebSocket.
# Active only when a CDP-capable backend is attached (Browserbase or

248
tests/tools/test_browser_hybrid_routing.py Normal file
View File

@@ -0,0 +1,248 @@
"""Tests for hybrid browser-backend routing (LAN/localhost auto-local).
When a cloud browser provider (Browserbase / Browser-Use / Firecrawl) is
configured globally, ``browser.auto_local_for_private_urls`` (default True)
causes ``browser_navigate`` to transparently spawn a local Chromium sidecar
for URLs whose host resolves to a private/loopback/LAN address, while
public URLs continue to hit the cloud session in the same conversation.
These tests cover the routing decision layer — session_key selection,
sidecar detection, last-active-session tracking, and the config toggle.
The downstream session creation is covered by test_browser_cloud_fallback.py.
"""
from unittest.mock import Mock
import pytest
import tools.browser_tool as browser_tool
@pytest.fixture(autouse=True)
def _reset_routing_state(monkeypatch):
"""Clear module-level caches so each test starts clean."""
monkeypatch.setattr(browser_tool, "_active_sessions", {})
monkeypatch.setattr(browser_tool, "_last_active_session_key", {})
monkeypatch.setattr(browser_tool, "_cached_cloud_provider", None)
monkeypatch.setattr(browser_tool, "_cloud_provider_resolved", False)
monkeypatch.setattr(browser_tool, "_auto_local_for_private_urls_resolved", False)
monkeypatch.setattr(browser_tool, "_cached_auto_local_for_private_urls", True)
monkeypatch.setattr(browser_tool, "_start_browser_cleanup_thread", lambda: None)
monkeypatch.setattr(browser_tool, "_update_session_activity", lambda t: None)
# Default: no CDP override, no Camofox
monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: False)
class TestNavigationSessionKey:
"""Tests for _navigation_session_key URL-based routing decisions."""
def test_public_url_uses_bare_task_id(self, monkeypatch):
"""Public URL with cloud provider configured → bare task_id (cloud)."""
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: Mock())
key = browser_tool._navigation_session_key("default", "https://github.com/x/y")
assert key == "default"
def test_localhost_routes_to_local_sidecar(self, monkeypatch):
"""``localhost`` URL → ``::local`` suffix when cloud configured + flag on."""
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: Mock())
key = browser_tool._navigation_session_key("default", "http://localhost:3000/")
assert key == "default::local"
def test_loopback_ipv4_routes_to_local_sidecar(self, monkeypatch):
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: Mock())
key = browser_tool._navigation_session_key("default", "http://127.0.0.1:8080/")
assert key == "default::local"
def test_rfc1918_lan_routes_to_local_sidecar(self, monkeypatch):
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: Mock())
key = browser_tool._navigation_session_key("default", "http://192.168.1.50:8000/")
assert key == "default::local"
def test_ipv6_loopback_routes_to_local_sidecar(self, monkeypatch):
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: Mock())
key = browser_tool._navigation_session_key("default", "http://[::1]:3000/")
assert key == "default::local"
def test_public_ip_literal_uses_bare_task_id(self, monkeypatch):
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: Mock())
key = browser_tool._navigation_session_key("default", "https://8.8.8.8/")
assert key == "default"
def test_mdns_local_hostname_routes_to_sidecar(self, monkeypatch):
"""``*.local`` mDNS / ``*.lan`` / ``*.internal`` hostnames route to sidecar."""
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: Mock())
for host in ("raspberrypi.local", "printer.lan", "db.internal"):
key = browser_tool._navigation_session_key("default", f"http://{host}/")
assert key == "default::local", f"host {host!r} did not route to sidecar"
def test_no_cloud_provider_stays_on_bare_task_id(self, monkeypatch):
"""When cloud provider is not configured, no hybrid routing happens."""
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: None)
key = browser_tool._navigation_session_key("default", "http://localhost:3000/")
assert key == "default"
def test_camofox_mode_stays_on_bare_task_id(self, monkeypatch):
"""Camofox is already local — no hybrid routing needed."""
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: Mock())
monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: True)
key = browser_tool._navigation_session_key("default", "http://localhost:3000/")
assert key == "default"
def test_cdp_override_stays_on_bare_task_id(self, monkeypatch):
"""A user-supplied CDP endpoint owns the whole session — no hybrid."""
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: Mock())
monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: "ws://localhost:9222")
key = browser_tool._navigation_session_key("default", "http://localhost:3000/")
assert key == "default"
def test_feature_flag_off_disables_hybrid_routing(self, monkeypatch):
"""``auto_local_for_private_urls: false`` keeps private URLs on cloud."""
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: Mock())
monkeypatch.setattr(browser_tool, "_auto_local_for_private_urls", lambda: False)
key = browser_tool._navigation_session_key("default", "http://localhost:3000/")
assert key == "default"
def test_none_task_id_defaults(self, monkeypatch):
"""``None`` task_id resolves to 'default'."""
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: Mock())
key = browser_tool._navigation_session_key(None, "http://localhost:3000/")
assert key == "default::local"
class TestSessionKeyHelpers:
def test_is_local_sidecar_key(self):
assert browser_tool._is_local_sidecar_key("default::local")
assert browser_tool._is_local_sidecar_key("my_task::local")
assert not browser_tool._is_local_sidecar_key("default")
assert not browser_tool._is_local_sidecar_key("my_task")
def test_last_session_key_falls_back_to_task_id(self, monkeypatch):
"""Without a recorded last-active key, returns the bare task_id."""
monkeypatch.setattr(browser_tool, "_last_active_session_key", {})
assert browser_tool._last_session_key("default") == "default"
assert browser_tool._last_session_key("task-42") == "task-42"
assert browser_tool._last_session_key(None) == "default"
def test_last_session_key_returns_recorded_key(self, monkeypatch):
monkeypatch.setattr(
browser_tool,
"_last_active_session_key",
{"default": "default::local", "task-42": "task-42"},
)
assert browser_tool._last_session_key("default") == "default::local"
assert browser_tool._last_session_key("task-42") == "task-42"
# Unknown task_id still falls back
assert browser_tool._last_session_key("other") == "other"
class TestHybridRoutingSessionCreation:
"""_get_session_info must force a local session when the key carries ``::local``."""
def test_local_sidecar_key_skips_cloud_provider(self, monkeypatch):
"""A ``::local``-suffixed key creates a local session even when cloud is set."""
provider = Mock()
provider.create_session.return_value = {
"session_name": "should_not_be_used",
"bb_session_id": "bb_xxx",
"cdp_url": "wss://fake.browserbase.com/ws",
}
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
monkeypatch.setattr(browser_tool, "_ensure_cdp_supervisor", lambda t: None)
session = browser_tool._get_session_info("default::local")
assert provider.create_session.call_count == 0
assert session["bb_session_id"] is None
assert session["cdp_url"] is None
assert session["features"]["local"] is True
def test_bare_task_id_with_cloud_provider_uses_cloud(self, monkeypatch):
"""A bare task_id with cloud provider configured hits the cloud path."""
provider = Mock()
provider.create_session.return_value = {
"session_name": "cloud-sess",
"bb_session_id": "bb_123",
"cdp_url": "wss://real.browserbase.com/ws",
}
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
monkeypatch.setattr(browser_tool, "_ensure_cdp_supervisor", lambda t: None)
monkeypatch.setattr(browser_tool, "_resolve_cdp_override", lambda u: u)
session = browser_tool._get_session_info("default")
assert provider.create_session.call_count == 1
assert session["bb_session_id"] == "bb_123"
class TestCleanupHybridSessions:
"""cleanup_browser(bare_task_id) must reap both cloud + local sidecar sessions."""
def test_cleanup_reaps_both_primary_and_sidecar(self, monkeypatch):
"""Given a bare task_id with both sessions alive, both get cleaned."""
reaped = []
def _fake_cleanup_one(key):
reaped.append(key)
monkeypatch.setattr(browser_tool, "_cleanup_single_browser_session", _fake_cleanup_one)
monkeypatch.setattr(
browser_tool,
"_active_sessions",
{
"default": {"session_name": "cloud_sess"},
"default::local": {"session_name": "local_sess"},
},
)
monkeypatch.setattr(
browser_tool, "_last_active_session_key", {"default": "default::local"}
)
browser_tool.cleanup_browser("default")
assert set(reaped) == {"default", "default::local"}
# last-active pointer dropped
assert "default" not in browser_tool._last_active_session_key
def test_cleanup_reaps_only_primary_when_no_sidecar(self, monkeypatch):
"""When no sidecar exists, only the primary is reaped."""
reaped = []
def _fake_cleanup_one(key):
reaped.append(key)
monkeypatch.setattr(browser_tool, "_cleanup_single_browser_session", _fake_cleanup_one)
monkeypatch.setattr(
browser_tool,
"_active_sessions",
{"default": {"session_name": "cloud_sess"}},
)
browser_tool.cleanup_browser("default")
assert reaped == ["default"]
def test_cleanup_sidecar_directly_keeps_primary(self, monkeypatch):
"""Calling cleanup with a ``::local`` key reaps only the sidecar."""
reaped = []
def _fake_cleanup_one(key):
reaped.append(key)
monkeypatch.setattr(browser_tool, "_cleanup_single_browser_session", _fake_cleanup_one)
monkeypatch.setattr(
browser_tool,
"_active_sessions",
{
"default": {"session_name": "cloud_sess"},
"default::local": {"session_name": "local_sess"},
},
)
monkeypatch.setattr(
browser_tool, "_last_active_session_key", {"default": "default::local"}
)
browser_tool.cleanup_browser("default::local")
assert reaped == ["default::local"]
# Last-active pointer NOT dropped (primary task is still alive)
assert browser_tool._last_active_session_key.get("default") == "default::local"

329
tools/browser_tool.py
View File

@@ -483,6 +483,147 @@ def _is_local_backend() -> bool:
return _is_camofox_mode() or _get_cloud_provider() is None
_auto_local_for_private_urls_resolved = False
_cached_auto_local_for_private_urls: bool = True
def _auto_local_for_private_urls() -> bool:
"""Return whether a cloud-configured install should auto-spawn a local
Chromium for LAN/localhost URLs.
Reads ``browser.auto_local_for_private_urls`` once (default ``True``) and
caches it for the process lifetime. When enabled, ``browser_navigate``
routes URLs whose host resolves to a private/loopback/LAN address to a
local headless Chromium sidecar even when a cloud provider (Browserbase
/ Browser-Use / Firecrawl) is configured globally. Public URLs continue
to use the cloud provider in the same conversation.
"""
global _auto_local_for_private_urls_resolved, _cached_auto_local_for_private_urls
if _auto_local_for_private_urls_resolved:
return _cached_auto_local_for_private_urls
_auto_local_for_private_urls_resolved = True
try:
from hermes_cli.config import read_raw_config
cfg = read_raw_config()
browser_cfg = cfg.get("browser", {})
if isinstance(browser_cfg, dict) and "auto_local_for_private_urls" in browser_cfg:
_cached_auto_local_for_private_urls = bool(
browser_cfg.get("auto_local_for_private_urls")
)
except Exception as e:
logger.debug("Could not read auto_local_for_private_urls from config: %s", e)
return _cached_auto_local_for_private_urls
def _url_is_private(url: str) -> bool:
"""Return True when the URL's host resolves to a private/LAN/loopback address.
Reuses ``tools.url_safety.is_safe_url`` as the oracle — if the SSRF check
would reject the URL, we treat it as "private" for routing purposes. DNS
resolution failures are treated as NOT private (fall through to whatever
backend is configured, which will surface the DNS error naturally).
"""
try:
from tools.url_safety import is_safe_url
# is_safe_url returns False for private/loopback/link-local/CGNAT AND
# for DNS failures. We only want the private-network case here, so
# we parse + check the host shape as a DNS-failure sieve first.
from urllib.parse import urlparse
import ipaddress
import socket
parsed = urlparse(url)
hostname = (parsed.hostname or "").strip().lower().rstrip(".")
if not hostname:
return False
# Literal IP → check directly
try:
ip = ipaddress.ip_address(hostname)
return (
ip.is_private
or ip.is_loopback
or ip.is_link_local
or ip in ipaddress.ip_network("100.64.0.0/10")
)
except ValueError:
pass
# Hostname — must resolve to confirm it's private (bare "localhost"
# resolves to 127.0.0.1 via /etc/hosts). Short-circuit on obvious
# names to avoid a DNS hop.
if hostname in ("localhost",) or hostname.endswith(".localhost"):
return True
if hostname.endswith(".local") or hostname.endswith(".lan") or hostname.endswith(".internal"):
return True
try:
addr_info = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
except socket.gaierror:
return False # DNS fail → not private, let the normal path fail
for _, _, _, _, sockaddr in addr_info:
try:
ip = ipaddress.ip_address(sockaddr[0])
except ValueError:
continue
if (
ip.is_private
or ip.is_loopback
or ip.is_link_local
or ip in ipaddress.ip_network("100.64.0.0/10")
):
return True
return False
except Exception as exc:
logger.debug("URL-privacy check failed for %s: %s", url, exc)
return False
def _navigation_session_key(task_id: str, url: str) -> str:
"""Pick the session key that should handle ``url`` for ``task_id``.
Returns the bare task_id unless ALL of these are true:
1. A cloud provider is configured (``_get_cloud_provider()`` is not None).
2. Auto-local routing is enabled (``browser.auto_local_for_private_urls``,
default True).
3. The URL resolves to a private/LAN/loopback address.
4. A CDP override is not active (that path owns the whole session).
5. Camofox mode is not active (Camofox is already local-only).
When all are true, returns ``f"{task_id}::local"`` so the hybrid-routing
path spawns a local Chromium sidecar while the cloud session (if any)
continues to serve public URLs.
"""
if task_id is None:
task_id = "default"
if _get_cdp_override():
return task_id
if _is_camofox_mode():
return task_id
if _get_cloud_provider() is None:
return task_id
if not _auto_local_for_private_urls():
return task_id
if not _url_is_private(url):
return task_id
return f"{task_id}{_LOCAL_SUFFIX}"
def _is_local_sidecar_key(session_key: str) -> bool:
"""Return True when ``session_key`` is a hybrid-routing local sidecar."""
return session_key.endswith(_LOCAL_SUFFIX)
def _last_session_key(task_id: str) -> str:
"""Return the session key to use for a non-nav browser tool call.
If a previous ``browser_navigate`` on this task_id set a last-active key,
use it so snapshot/click/fill/etc. hit the same session. Otherwise fall
back to the bare task_id (matches original behavior for tasks that never
triggered hybrid routing).
"""
if task_id is None:
task_id = "default"
return _last_active_session_key.get(task_id, task_id)
def _allow_private_urls() -> bool:
"""Return whether the browser is allowed to navigate to private/internal addresses.
@@ -521,10 +662,25 @@ def _socket_safe_tmpdir() -> str:
return tempfile.gettempdir()
# Track active sessions per task
# Track active sessions per "session key".
#
# A "session key" is either the bare task_id (cloud/default path) OR a composite
# like f"{task_id}::local" when the hybrid-routing feature spawns a local sidecar
# browser for a LAN/localhost URL while a cloud provider is configured globally.
# Both forms flow through the same _active_sessions / _run_browser_command /
# cleanup_browser code paths — the key is opaque to those internals.
#
# Stores: session_name (always), bb_session_id + cdp_url (cloud mode only)
_active_sessions: Dict[str, Dict[str, str]] = {} # task_id -> {session_name, ...}
_recording_sessions: set = set() # task_ids with active recordings
_active_sessions: Dict[str, Dict[str, str]] = {} # session_key -> {session_name, ...}
_recording_sessions: set = set() # session_keys with active recordings
# Tracks the most recent session_key used per task_id. Set by browser_navigate()
# after it chooses a backend for a URL; read by every non-nav browser tool
# (snapshot/click/fill/eval/...) so they target the session that served the last
# navigation. Without this, a task that navigated to localhost on the local
# sidecar would fall back to the cloud session on its next snapshot call.
_last_active_session_key: Dict[str, str] = {} # task_id -> session_key
_LOCAL_SUFFIX = "::local"
# Flag to track if cleanup has been done
_cleanup_done = False
@@ -1014,37 +1170,48 @@ def _create_cdp_session(task_id: str, cdp_url: str) -> Dict[str, str]:
def _get_session_info(task_id: Optional[str] = None) -> Dict[str, str]:
"""
Get or create session info for the given task.
Get or create session info for the given session key.
In cloud mode, creates a Browserbase session with proxies enabled.
In local mode, generates a session name for agent-browser --session.
Also starts the inactivity cleanup thread and updates activity tracking.
Thread-safe: multiple subagents can call this concurrently.
Args:
task_id: Unique identifier for the task
task_id: Session key. Normally the task_id as-is, but may carry the
``::local`` suffix for the hybrid-routing local sidecar — in that
case the cloud provider is skipped even when one is configured,
and a local Chromium session is created instead.
Returns:
Dict with session_name (always), bb_session_id + cdp_url (cloud only)
"""
if task_id is None:
task_id = "default"
# Start the cleanup thread if not running (handles inactivity timeouts)
_start_browser_cleanup_thread()
# Update activity timestamp for this session
_update_session_activity(task_id)
with _cleanup_lock:
# Check if we already have a session for this task
if task_id in _active_sessions:
return _active_sessions[task_id]
# Hybrid routing: session keys ending with ``::local`` force a local
# Chromium regardless of the globally-configured cloud provider. Public
# URLs in the same conversation continue to use the cloud session under
# the bare task_id key.
force_local = _is_local_sidecar_key(task_id)
# Create session outside the lock (network call in cloud mode)
cdp_override = _get_cdp_override()
if cdp_override:
if cdp_override and not force_local:
session_info = _create_cdp_session(task_id, cdp_override)
elif force_local:
session_info = _create_local_session(task_id)
else:
provider = _get_cloud_provider()
if provider is None:
@@ -1081,7 +1248,7 @@ def _get_session_info(task_id: Optional[str] = None) -> Dict[str, str]:
session_info["fallback_from_cloud"] = True
session_info["fallback_reason"] = str(e)
session_info["fallback_provider"] = provider_name
with _cleanup_lock:
# Double-check: another thread may have created a session while we
# were doing the network call. Use the existing one to avoid leaking
@@ -1093,7 +1260,9 @@ def _get_session_info(task_id: Optional[str] = None) -> Dict[str, str]:
# Lazy-start the CDP supervisor now that the session exists (if the
# backend surfaces a CDP URL via override or session_info["cdp_url"]).
# Idempotent; swallows errors. See _ensure_cdp_supervisor for details.
_ensure_cdp_supervisor(task_id)
# Skip for local sidecars — they have no CDP URL.
if not force_local:
_ensure_cdp_supervisor(task_id)
return session_info
@@ -1521,9 +1690,21 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str:
# SSRF protection — block private/internal addresses before navigating.
# Skipped for local backends (Camofox, headless Chromium without a cloud
# provider) because the agent already has full local network access via
# the terminal tool. Can also be opted out for cloud mode via
# ``browser.allow_private_urls`` in config.
if not _is_local_backend() and not _allow_private_urls() and not _is_safe_url(url):
# the terminal tool. Also skipped when hybrid routing will auto-spawn a
# local Chromium sidecar for this URL (cloud provider configured +
# private URL + ``browser.auto_local_for_private_urls`` enabled) — the
# cloud provider never sees the URL in that case. Can also be opted
# out globally via ``browser.allow_private_urls`` in config.
effective_task_id = task_id or "default"
nav_session_key = _navigation_session_key(effective_task_id, url)
auto_local_this_nav = _is_local_sidecar_key(nav_session_key)
if (
not _is_local_backend()
and not auto_local_this_nav
and not _allow_private_urls()
and not _is_safe_url(url)
):
return json.dumps({
"success": False,
"error": "Blocked: URL targets a private or internal address",
@@ -1543,19 +1724,31 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str:
from tools.browser_camofox import camofox_navigate
return camofox_navigate(url, task_id)
effective_task_id = task_id or "default"
if auto_local_this_nav:
logger.info(
"browser_navigate: auto-routing %s to local Chromium sidecar "
"(cloud provider %s stays on cloud for public URLs; "
"set browser.auto_local_for_private_urls: false to disable)",
url,
type(_get_cloud_provider()).__name__ if _get_cloud_provider() else "none",
)
# Get session info to check if this is a new session
# (will create one with features logged if not exists)
session_info = _get_session_info(effective_task_id)
session_info = _get_session_info(nav_session_key)
is_first_nav = session_info.get("_first_nav", True)
# Auto-start recording if configured and this is first navigation
if is_first_nav:
session_info["_first_nav"] = False
_maybe_start_recording(effective_task_id)
_maybe_start_recording(nav_session_key)
result = _run_browser_command(effective_task_id, "open", [url], timeout=max(_get_command_timeout(), 60))
result = _run_browser_command(nav_session_key, "open", [url], timeout=max(_get_command_timeout(), 60))
# Remember which session served this nav so snapshot/click/fill/...
# on the same task_id hit it (critical when hybrid routing has both a
# cloud session and a local sidecar alive concurrently).
_last_active_session_key[effective_task_id] = nav_session_key
if result.get("success"):
data = result.get("data", {})
@@ -1565,10 +1758,17 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str:
# Post-redirect SSRF check — if the browser followed a redirect to a
# private/internal address, block the result so the model can't read
# internal content via subsequent browser_snapshot calls.
# Skipped for local backends (same rationale as the pre-nav check).
if not _is_local_backend() and not _allow_private_urls() and final_url and final_url != url and not _is_safe_url(final_url):
# Skipped for local backends (same rationale as the pre-nav check),
# and for the hybrid local sidecar (we're already on a local browser
# hitting a private URL by design).
if (
not _is_local_backend()
and not auto_local_this_nav
and not _allow_private_urls()
and final_url and final_url != url and not _is_safe_url(final_url)
):
# Navigate away to a blank page to prevent snapshot leaks
_run_browser_command(effective_task_id, "open", ["about:blank"], timeout=10)
_run_browser_command(nav_session_key, "open", ["about:blank"], timeout=10)
return json.dumps({
"success": False,
"error": "Blocked: redirect landed on a private/internal address",
@@ -1612,7 +1812,7 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str:
# Auto-take a compact snapshot so the model can act immediately
# without a separate browser_snapshot call.
try:
snap_result = _run_browser_command(effective_task_id, "snapshot", ["-c"])
snap_result = _run_browser_command(nav_session_key, "snapshot", ["-c"])
if snap_result.get("success"):
snap_data = snap_result.get("data", {})
snapshot_text = snap_data.get("snapshot", "")
@@ -1652,7 +1852,7 @@ def browser_snapshot(
from tools.browser_camofox import camofox_snapshot
return camofox_snapshot(full, task_id, user_task)
effective_task_id = task_id or "default"
effective_task_id = _last_session_key(task_id or "default")
# Build command args based on full flag
args = []
@@ -1714,7 +1914,7 @@ def browser_click(ref: str, task_id: Optional[str] = None) -> str:
from tools.browser_camofox import camofox_click
return camofox_click(ref, task_id)
effective_task_id = task_id or "default"
effective_task_id = _last_session_key(task_id or "default")
# Ensure ref starts with @
if not ref.startswith("@"):
@@ -1750,7 +1950,7 @@ def browser_type(ref: str, text: str, task_id: Optional[str] = None) -> str:
from tools.browser_camofox import camofox_type
return camofox_type(ref, text, task_id)
effective_task_id = task_id or "default"
effective_task_id = _last_session_key(task_id or "default")
# Ensure ref starts with @
if not ref.startswith("@"):
@@ -1804,7 +2004,7 @@ def browser_scroll(direction: str, task_id: Optional[str] = None) -> str:
result = camofox_scroll(direction, task_id)
return result
effective_task_id = task_id or "default"
effective_task_id = _last_session_key(task_id or "default")
result = _run_browser_command(effective_task_id, "scroll", [direction, str(_SCROLL_PIXELS)])
if not result.get("success"):
@@ -1833,7 +2033,7 @@ def browser_back(task_id: Optional[str] = None) -> str:
from tools.browser_camofox import camofox_back
return camofox_back(task_id)
effective_task_id = task_id or "default"
effective_task_id = _last_session_key(task_id or "default")
result = _run_browser_command(effective_task_id, "back", [])
if result.get("success"):
@@ -1864,7 +2064,7 @@ def browser_press(key: str, task_id: Optional[str] = None) -> str:
from tools.browser_camofox import camofox_press
return camofox_press(key, task_id)
effective_task_id = task_id or "default"
effective_task_id = _last_session_key(task_id or "default")
result = _run_browser_command(effective_task_id, "press", [key])
if result.get("success"):
@@ -1906,7 +2106,7 @@ def browser_console(clear: bool = False, expression: Optional[str] = None, task_
from tools.browser_camofox import camofox_console
return camofox_console(clear, task_id)
effective_task_id = task_id or "default"
effective_task_id = _last_session_key(task_id or "default")
console_args = ["--clear"] if clear else []
error_args = ["--clear"] if clear else []
@@ -1945,7 +2145,7 @@ def _browser_eval(expression: str, task_id: Optional[str] = None) -> str:
if _is_camofox_mode():
return _camofox_eval(expression, task_id)
effective_task_id = task_id or "default"
effective_task_id = _last_session_key(task_id or "default")
result = _run_browser_command(effective_task_id, "eval", [expression])
if not result.get("success"):
@@ -2077,7 +2277,7 @@ def browser_get_images(task_id: Optional[str] = None) -> str:
from tools.browser_camofox import camofox_get_images
return camofox_get_images(task_id)
effective_task_id = task_id or "default"
effective_task_id = _last_session_key(task_id or "default")
# Use eval to run JavaScript that extracts images
js_code = """JSON.stringify(
@@ -2147,7 +2347,7 @@ def browser_vision(question: str, annotate: bool = False, task_id: Optional[str]
import base64
import uuid as uuid_mod
effective_task_id = task_id or "default"
effective_task_id = _last_session_key(task_id or "default")
# Save screenshot to persistent location so it can be shared with users
from hermes_constants import get_hermes_dir
@@ -2350,17 +2550,47 @@ def _cleanup_old_recordings(max_age_hours=72):
def cleanup_browser(task_id: Optional[str] = None) -> None:
"""
Clean up browser session for a task.
Clean up browser session(s) for a task.
Called automatically when a task completes or when inactivity timeout is reached.
Closes both the agent-browser/Browserbase session and Camofox sessions.
When ``task_id`` is a bare task identifier (no ``::local`` suffix), reaps
BOTH the cloud/primary session AND any hybrid-routing local sidecar that
may have been spawned for LAN/localhost URLs in the same task. When
``task_id`` already carries a ``::local`` suffix (called from the inactivity
cleanup loop against a specific session key), reaps only that one.
Args:
task_id: Task identifier to clean up
task_id: Task identifier (or explicit session key)
"""
if task_id is None:
task_id = "default"
# Expand to the full set of session keys to reap. For a bare task_id
# that includes the cloud/primary key + the local sidecar if one exists.
if _is_local_sidecar_key(task_id):
session_keys = [task_id]
bare_task_id = task_id[: -len(_LOCAL_SUFFIX)]
else:
session_keys = [task_id]
sidecar_key = f"{task_id}{_LOCAL_SUFFIX}"
with _cleanup_lock:
if sidecar_key in _active_sessions:
session_keys.append(sidecar_key)
bare_task_id = task_id
for session_key in session_keys:
_cleanup_single_browser_session(session_key)
# Drop the last-active pointer only when the bare task is being cleaned
# (i.e. not when we're only reaping a sidecar mid-task).
if not _is_local_sidecar_key(task_id):
_last_active_session_key.pop(bare_task_id, None)
def _cleanup_single_browser_session(task_id: str) -> None:
"""Internal: reap a single browser session by its exact session key."""
# Stop the CDP supervisor for this task FIRST so we close our WebSocket
# before the backend tears down the underlying CDP endpoint.
_stop_cdp_supervisor(task_id)
@@ -2379,32 +2609,33 @@ def cleanup_browser(task_id: Optional[str] = None) -> None:
logger.debug("cleanup_browser called for task_id: %s", task_id)
logger.debug("Active sessions: %s", list(_active_sessions.keys()))
# Check if session exists (under lock), but don't remove yet -
# _run_browser_command needs it to build the close command.
with _cleanup_lock:
session_info = _active_sessions.get(task_id)
if session_info:
bb_session_id = session_info.get("bb_session_id", "unknown")
logger.debug("Found session for task %s: bb_session_id=%s", task_id, bb_session_id)
# Stop auto-recording before closing (saves the file)
_maybe_stop_recording(task_id)
# Try to close via agent-browser first (needs session in _active_sessions)
try:
_run_browser_command(task_id, "close", [], timeout=10)
logger.debug("agent-browser close command completed for task %s", task_id)
except Exception as e:
logger.warning("agent-browser close failed for task %s: %s", task_id, e)
# Now remove from tracking under lock
with _cleanup_lock:
_active_sessions.pop(task_id, None)
_session_last_activity.pop(task_id, None)
# Cloud mode: close the cloud browser session via provider API
# Cloud mode: close the cloud browser session via provider API.
# Local sidecars have bb_session_id=None so this no-ops for them.
if bb_session_id:
provider = _get_cloud_provider()
if provider is not None:

34
website/docs/user-guide/features/browser.md
View File

@@ -86,6 +86,40 @@ FIRECRAWL_API_URL=http://localhost:3002
FIRECRAWL_BROWSER_TTL=600
```
### Hybrid routing: cloud for public URLs, local for LAN/localhost
When a cloud provider is configured, Hermes auto-spawns a **local Chromium sidecar**
for URLs that resolve to a private/loopback/LAN address (`localhost`, `127.0.0.1`,
`192.168.x.x`, `10.x.x.x`, `172.16-31.x.x`, `*.local`, `*.lan`, `*.internal`,
IPv6 loopback `::1`, link-local `169.254.x.x`). Public URLs continue to use the
cloud provider in the same conversation.
This solves the common "I'm developing locally but using Browserbase" workflow —
the agent can screenshot your dashboard at `http://localhost:3000` AND scrape
`https://github.com` without you switching providers or disabling the SSRF guard.
The cloud provider never sees the private URL.
The feature is **on by default**. To disable it (all URLs go to the configured
cloud provider, as before):
```yaml
# ~/.hermes/config.yaml
browser:
cloud_provider: browserbase
auto_local_for_private_urls: false
```
With auto-routing disabled, private URLs are rejected with
`"Blocked: URL targets a private or internal address"` unless you also set
`browser.allow_private_urls: true` (which lets the cloud provider attempt them —
usually won't work since Browserbase etc. can't reach your LAN).
Requirements: the local sidecar uses the same `agent-browser` CLI as pure local
mode, so you need it installed (`hermes setup tools → Browser Automation`
auto-installs it). Post-navigation redirects from a public URL onto a private
address are still blocked (you can't use a redirect-to-internal trick to reach
your LAN through the public path).
### Camofox local mode
[Camofox](https://github.com/jo-inc/camofox-browser) is a self-hosted Node.js server wrapping Camoufox (a Firefox fork with C++ fingerprint spoofing). It provides local anti-detection browsing without cloud dependencies.