diff --git a/CHANGELOG.md b/CHANGELOG.md index c8223722d2..d07b310ec8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -460,6 +460,18 @@ This release closes that milestone with **301 merged pull requests**. ### Fixed +- **browser:** The agentic browser can no longer be steered to a private or + cloud-metadata host by a DNS answer that changes after the SSRF check. The + navigate screen and the per-request guard resolve a hostname to vet it, but + Chromium resolved it again to connect, so a rebinding DNS server could answer + the checks with a public address and the connection with a private one. Each + launched browser now sends every TCP connection through a per-session + loopback SOCKS5 proxy: Chromium hands it the hostname, and the proxy resolves + it once under the same `allow_private_addresses` policy and connects to + exactly the vetted addresses. Loopback traffic goes through the proxy too. + WebRTC UDP does not traverse the proxy and is not covered. CDP-attached Chrome + is unchanged, and delegated fetch services (crawl4ai, Browserless, fastcrw) + still resolve on their own side, which the Gateway cannot pin. ([#6201]) - **channels:** The Discord typing indicator is now actually sent while the agent works on a reply. `_start_typing()` called `channel.trigger_typing()`, which discord.py removed in 2.0 (the project requires `>=2.7.0`), and its loop @@ -7585,4 +7597,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6135]: https://github.com/bytedance/deer-flow/pull/6135 [#6138]: https://github.com/bytedance/deer-flow/pull/6138 [#6140]: https://github.com/bytedance/deer-flow/pull/6140 +[#6201]: https://github.com/bytedance/deer-flow/pull/6201 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 695784769d..3e3dfb735e 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -419,6 +419,14 @@ ### 修复 +- **浏览器：** 智能浏览器不会再因为 SSRF 检查之后发生变化的 DNS 应答而被引向 + 内网或云元数据主机。导航检查和逐请求守卫会解析主机名进行筛查，但 Chromium + 建立连接时会再次解析，因此重绑定 DNS 服务器可以对检查返回公网地址、对连接返回 + 内网地址。现在每个启动的浏览器的所有 TCP 连接都经过一个按会话创建的本地回环 SOCKS5 代理： + Chromium 把主机名交给代理，代理按相同的 `allow_private_addresses` 策略只解析一次， + 并且只连接筛查通过的地址。回环流量同样经过代理。WebRTC UDP 不经过代理，不在覆盖范围内。 + 通过 CDP 连接的 Chrome 不受影响； + 委托抓取服务（crawl4ai、Browserless、fastcrw）仍在其自身一侧解析，Gateway 无法固定。([#6201]) - **渠道：** Discord 现在会在智能体生成回复期间真正显示"正在输入"提示。`_start_typing()` 调用的 `channel.trigger_typing()` 已在 discord.py 2.0 中移除（项目要求 `>=2.7.0`），而其循环吞掉了 所有异常，因此每次都抛出 `AttributeError`，提示从未发送。现在改为 await 2.x 的 @@ -6345,3 +6353,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6135]: https://github.com/bytedance/deer-flow/pull/6135 [#6138]: https://github.com/bytedance/deer-flow/pull/6138 [#6140]: https://github.com/bytedance/deer-flow/pull/6140 +[#6201]: https://github.com/bytedance/deer-flow/pull/6201 diff --git a/README.md b/README.md index 23c2a32b12..47b2b3fc3a 100644 --- a/README.md +++ b/README.md @@ -1941,7 +1941,7 @@ order within a tool entry, with either indented or indentless YAML lists. Reading a page is not the same as *using* one. Alongside the read-only `web_fetch` and `web_capture` tools, DeerFlow ships an optional agentic browser tool group that keeps a live, per-conversation browser session so the agent can actually operate a page — navigate, read the interactive elements, click, type, submit forms, and follow multi-step flows on JavaScript-heavy sites. -Each action returns a fresh snapshot of the page's interactive elements, each addressed by a stable `[ref]` number, so the agent acts on what it just observed instead of guessing selectors. Outbound URLs are SSRF-screened by default. It is powered by Playwright and shipped as an optional extra so the core install stays lean: +Each action returns a fresh snapshot of the page's interactive elements, each addressed by a stable `[ref]` number, so the agent acts on what it just observed instead of guessing selectors. Outbound URLs are SSRF-screened by default, and the browser's TCP connections go through a local proxy that pins each one to the screened addresses, so a DNS answer that changes after the check cannot redirect them to a private host (WebRTC UDP is not covered). It is powered by Playwright and shipped as an optional extra so the core install stays lean: ```bash cd backend diff --git a/README_zh.md b/README_zh.md index 22ceb3b0fd..708b36703f 100644 --- a/README_zh.md +++ b/README_zh.md @@ -788,7 +788,7 @@ DeerFlow 不只是“会说它能做”，它是真的有一台自己的“电 读取页面和真正“使用”页面不是一回事。除了只读的 `web_fetch` 和 `web_capture` 工具外，DeerFlow 还提供一组可选的 agentic browser 工具，为每次对话保持一个实时浏览器会话，让 agent 真正操作页面——导航、读取可交互元素、点击、输入、提交表单，并在重度 JavaScript 站点上完成多步流程。 -每次操作都会返回页面可交互元素的最新快照，每个元素用稳定的 `[ref]` 编号寻址，因此 agent 基于刚观察到的内容行动，而不是猜测选择器。出站 URL 默认会经过 SSRF 筛查。该能力由 Playwright 提供，作为 optional extra 发布，以保持核心安装精简： +每次操作都会返回页面可交互元素的最新快照，每个元素用稳定的 `[ref]` 编号寻址，因此 agent 基于刚观察到的内容行动，而不是猜测选择器。出站 URL 默认会经过 SSRF 筛查，浏览器的 TCP 连接也都经过一个本地代理，该代理把每个连接固定到筛查通过的地址，因此检查之后才改变的 DNS 应答无法把连接引向内网主机（WebRTC UDP 不在覆盖范围内）。该能力由 Playwright 提供，作为 optional extra 发布，以保持核心安装精简： ```bash cd backend diff --git a/backend/app/gateway/routers/browser.py b/backend/app/gateway/routers/browser.py index 2e9b622772..87aeec036c 100644 --- a/backend/app/gateway/routers/browser.py +++ b/backend/app/gateway/routers/browser.py @@ -261,6 +261,7 @@ async def browser_stream(websocket: WebSocket, thread_id: ThreadId) -> None: BrowserLiveViewerError, BrowserSessionCapacityError, get_browser_session_manager, + resolve_browser_egress, validate_browser_url, ) except ImportError: @@ -326,6 +327,7 @@ async def browser_stream(websocket: WebSocket, thread_id: ThreadId) -> None: cdp_url=_cfg_str("cdp_url"), allow_unguarded_cdp=_cfg_bool("allow_unguarded_cdp", False), url_guard=validate_browser_url, + egress_resolver=resolve_browser_egress, ) session = session_lease.__enter__() except BrowserSessionCapacityError: diff --git a/backend/packages/harness/deerflow/community/browser_automation/__init__.py b/backend/packages/harness/deerflow/community/browser_automation/__init__.py index 7d87ed0ad9..8b415b3712 100644 --- a/backend/packages/harness/deerflow/community/browser_automation/__init__.py +++ b/backend/packages/harness/deerflow/community/browser_automation/__init__.py @@ -21,6 +21,7 @@ from .tools import ( browser_snapshot_tool, browser_type_tool, navigate_and_capture, + resolve_browser_egress, validate_browser_url, ) @@ -45,5 +46,6 @@ __all__ = [ "navigate_and_capture", "redact_browser_url", "reset_browser_session_manager", + "resolve_browser_egress", "validate_browser_url", ] diff --git a/backend/packages/harness/deerflow/community/browser_automation/egress.py b/backend/packages/harness/deerflow/community/browser_automation/egress.py new file mode 100644 index 0000000000..7c86f65852 --- /dev/null +++ b/backend/packages/harness/deerflow/community/browser_automation/egress.py @@ -0,0 +1,206 @@ +"""Pinned browser egress: a loopback SOCKS5 proxy that connects only to vetted addresses. + +The context request guard screens a URL by resolving its hostname, but +Chromium resolves the name again when it opens the connection. A rebinding DNS +server can pass the screen with a public answer and then hand Chromium a +private or cloud-metadata address. Launching Chromium through this proxy moves +name resolution out of the browser: over SOCKS5, Chromium sends the hostname +itself, the proxy resolves it exactly once through the session's egress +resolver, and it opens the TCP connection only to the addresses that resolver +vetted. + +Only the SOCKS5 ``CONNECT`` command without authentication is supported, which +is everything Chromium sends for HTTP, HTTPS, and WebSocket traffic. The proxy +never parses the tunnelled bytes. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import ipaddress +import logging +from collections.abc import Callable + +logger = logging.getLogger(__name__) + +# Resolves a destination host (DNS name or IP literal) to the addresses the +# browser may connect to. Raises ``ValueError`` when the host must be refused; +# any other exception is logged as a resolver fault and the connection fails. +EgressResolver = Callable[[str], list[str]] + +_SOCKS_VERSION = 5 +_NO_AUTHENTICATION = 0x00 +_NO_ACCEPTABLE_METHODS = 0xFF +_CONNECT = 0x01 +_ADDRESS_IPV4 = 0x01 +_ADDRESS_DOMAIN = 0x03 +_ADDRESS_IPV6 = 0x04 +_REPLY_SUCCEEDED = 0x00 +_REPLY_GENERAL_FAILURE = 0x01 +_REPLY_NOT_ALLOWED = 0x02 +_REPLY_HOST_UNREACHABLE = 0x04 +_REPLY_COMMAND_NOT_SUPPORTED = 0x07 +_REPLY_ADDRESS_TYPE_NOT_SUPPORTED = 0x08 + +_HANDSHAKE_TIMEOUT_S = 15 +_RESOLVE_TIMEOUT_S = 15 +_CONNECT_TIMEOUT_S = 15 +_CLOSE_TIMEOUT_S = 2 + + +class _RefusedRequest(Exception): + def __init__(self, reply: int) -> None: + super().__init__(reply) + self.reply = reply + + +class BrowserEgressProxy: + """A per-session SOCKS5 proxy bound to 127.0.0.1 on an ephemeral port.""" + + def __init__(self, resolve: EgressResolver) -> None: + self._resolve = resolve + self._server: asyncio.Server | None = None + self._client_writers: set[asyncio.StreamWriter] = set() + + async def start(self) -> str: + """Start listening and return the ``socks5://`` server URL for Chromium.""" + if self._server is None: + self._server = await asyncio.start_server(self._handle, host="127.0.0.1", port=0) + port = self._server.sockets[0].getsockname()[1] + return f"socks5://127.0.0.1:{port}" + + async def close(self) -> None: + server, self._server = self._server, None + if server is None: + return + server.close() + # Server.wait_closed() waits for every open connection, so close the + # tunnels the browser left behind instead of waiting on them. + for writer in list(self._client_writers): + writer.close() + # A handler still resolving, or accepted just before close, can outlive + # the sweep; it must not hold up session close or eviction. + with contextlib.suppress(Exception): + await asyncio.wait_for(server.wait_closed(), timeout=_CLOSE_TIMEOUT_S) + + async def _handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + if self._server is None: + # Accepted just before close(); never open a tunnel after it. + writer.close() + return + self._client_writers.add(writer) + try: + try: + host, port = await asyncio.wait_for(_read_connect_request(reader, writer), timeout=_HANDSHAKE_TIMEOUT_S) + except _RefusedRequest as exc: + await _reply(writer, exc.reply) + return + try: + addresses = await asyncio.wait_for(asyncio.to_thread(self._resolve, host), timeout=_RESOLVE_TIMEOUT_S) + except ValueError as exc: + logger.warning("browser egress refused for %s:%d: %s", host, port, exc) + await _reply(writer, _REPLY_NOT_ALLOWED) + return + except TimeoutError: + logger.warning("browser egress resolution timed out for %s:%d", host, port) + await _reply(writer, _REPLY_HOST_UNREACHABLE) + return + except Exception: + # A resolver fault is not a policy refusal; keep the two + # distinguishable in the log and in the browser's error. + logger.exception("browser egress resolver failed for %s:%d", host, port) + await _reply(writer, _REPLY_GENERAL_FAILURE) + return + upstream = await _open_first(addresses, port) + if upstream is None: + await _reply(writer, _REPLY_HOST_UNREACHABLE) + return + upstream_reader, upstream_writer = upstream + try: + await _reply(writer, _REPLY_SUCCEEDED) + await asyncio.gather(_relay(reader, upstream_writer), _relay(upstream_reader, writer)) + finally: + with contextlib.suppress(Exception): + upstream_writer.close() + await upstream_writer.wait_closed() + except (asyncio.IncompleteReadError, ConnectionError, TimeoutError): + pass + finally: + self._client_writers.discard(writer) + with contextlib.suppress(Exception): + writer.close() + await writer.wait_closed() + + +async def _read_connect_request(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> tuple[str, int]: + """Negotiate no-auth SOCKS5 and return the ``CONNECT`` destination.""" + version, method_count = await reader.readexactly(2) + methods = await reader.readexactly(method_count) + if version != _SOCKS_VERSION: + raise ConnectionError("not a SOCKS5 client") + if _NO_AUTHENTICATION not in methods: + writer.write(bytes([_SOCKS_VERSION, _NO_ACCEPTABLE_METHODS])) + await writer.drain() + raise ConnectionError("SOCKS5 client requires authentication") + writer.write(bytes([_SOCKS_VERSION, _NO_AUTHENTICATION])) + await writer.drain() + + version, command, _reserved, address_type = await reader.readexactly(4) + if version != _SOCKS_VERSION: + raise ConnectionError("not a SOCKS5 request") + if address_type == _ADDRESS_IPV4: + host = str(ipaddress.IPv4Address(await reader.readexactly(4))) + elif address_type == _ADDRESS_IPV6: + host = str(ipaddress.IPv6Address(await reader.readexactly(16))) + elif address_type == _ADDRESS_DOMAIN: + (length,) = await reader.readexactly(1) + raw_host = await reader.readexactly(length) + try: + host = raw_host.decode("ascii") + except UnicodeDecodeError: + raise _RefusedRequest(_REPLY_ADDRESS_TYPE_NOT_SUPPORTED) from None + if not host: + raise _RefusedRequest(_REPLY_ADDRESS_TYPE_NOT_SUPPORTED) + else: + raise _RefusedRequest(_REPLY_ADDRESS_TYPE_NOT_SUPPORTED) + port = int.from_bytes(await reader.readexactly(2), "big") + if command != _CONNECT: + raise _RefusedRequest(_REPLY_COMMAND_NOT_SUPPORTED) + return host, port + + +async def _reply(writer: asyncio.StreamWriter, reply: int) -> None: + # The bound address is unused by Chromium, so always report 0.0.0.0:0. + writer.write(bytes([_SOCKS_VERSION, reply, 0x00, _ADDRESS_IPV4, 0, 0, 0, 0, 0, 0])) + await writer.drain() + + +async def _open_first(addresses: list[str], port: int) -> tuple[asyncio.StreamReader, asyncio.StreamWriter] | None: + """Connect to the first reachable vetted address within one shared deadline.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + _CONNECT_TIMEOUT_S + for index, address in enumerate(addresses): + remaining = deadline - loop.time() + if remaining <= 0: + break + # Leave every later address an equal share of the remaining budget so a + # black-holed first answer cannot strand reachable candidates. + try: + return await asyncio.wait_for(asyncio.open_connection(address, port), timeout=remaining / (len(addresses) - index)) + except (OSError, TimeoutError): + continue + return None + + +async def _relay(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + try: + while chunk := await reader.read(65_536): + writer.write(chunk) + await writer.drain() + except ConnectionError: + pass + finally: + with contextlib.suppress(Exception): + writer.close() + await writer.wait_closed() diff --git a/backend/packages/harness/deerflow/community/browser_automation/session.py b/backend/packages/harness/deerflow/community/browser_automation/session.py index 95d319e20e..2db3bc9e2e 100644 --- a/backend/packages/harness/deerflow/community/browser_automation/session.py +++ b/backend/packages/harness/deerflow/community/browser_automation/session.py @@ -26,6 +26,8 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, Literal, TypeVar from urllib.parse import urlparse +from .egress import BrowserEgressProxy, EgressResolver + if TYPE_CHECKING: from playwright.async_api import Browser, BrowserContext, Page, Playwright @@ -276,6 +278,7 @@ class BrowserSession: viewport: dict[str, int], cdp_url: str | None = None, url_guard: Callable[[str], str | None] | None = None, + egress_resolver: EgressResolver | None = None, on_activity: Callable[[], None] | None = None, ) -> None: self._loop = loop @@ -291,6 +294,13 @@ class BrowserSession: # cloud-metadata host. self._url_guard = url_guard self._request_guard_bound = False + # The request guard resolves a URL's host to screen it, but Chromium + # resolves it again to connect. A launched browser therefore sends every + # TCP connection through a loopback SOCKS5 proxy that resolves once through + # ``egress_resolver`` and connects only to the addresses it vetted, so a + # rebinding DNS answer cannot reach a private or metadata host. + self._egress_resolver = egress_resolver + self._egress_proxy: BrowserEgressProxy | None = None # When set, attach to an already-running Chrome via the DevTools # Protocol (like Codex's "connect to your real browser") instead of # launching a private headless instance. The user watches the agent @@ -399,7 +409,7 @@ class BrowserSession: return self._page if self._browser is None or not self._browser.is_connected(): - self._browser = await self._playwright.chromium.launch(headless=self._headless) + self._browser = await self._playwright.chromium.launch(headless=self._headless, proxy=await self._egress_proxy_settings()) # device_scale_factor=2 renders screenshots at retina density so the # panel stays crisp when the image is scaled up to fill the view. self._context = await self._browser.new_context(viewport=self._viewport, device_scale_factor=2) @@ -409,6 +419,15 @@ class BrowserSession: self._bind_new_page_listener() return self._page + async def _egress_proxy_settings(self) -> dict[str, str] | None: + if self._egress_resolver is None: + return None + if self._egress_proxy is None: + self._egress_proxy = BrowserEgressProxy(self._egress_resolver) + # Playwright already proxies loopback for Chromium unless an environment + # override disables it; spell it out so localhost is never exempt. + return {"server": await self._egress_proxy.start(), "bypass": "<-loopback>"} + def _set_active_page(self, page: Page) -> None: """Adopt *page* as the active page and keep the live screencast on it. @@ -693,7 +712,11 @@ class BrowserSession: if self._playwright is not None: with contextlib.suppress(Exception): await self._playwright.stop() + if self._egress_proxy is not None: + with contextlib.suppress(Exception): + await self._egress_proxy.close() finally: + self._egress_proxy = None self._playwright = None self._browser = None self._context = None @@ -918,6 +941,7 @@ class BrowserSessionManager: cdp_url: str | None = None, allow_unguarded_cdp: bool = False, url_guard: Callable[[str], str | None] | None = None, + egress_resolver: EgressResolver | None = None, pin: bool = False, ) -> BrowserSession: ensure_browser_worker_compatibility() @@ -942,6 +966,7 @@ class BrowserSessionManager: viewport=viewport or {"width": 1280, "height": 720}, cdp_url=cdp_url, url_guard=url_guard, + egress_resolver=egress_resolver, on_activity=lambda: self._touch_session(key), ) self._sessions[key] = session diff --git a/backend/packages/harness/deerflow/community/browser_automation/tools.py b/backend/packages/harness/deerflow/community/browser_automation/tools.py index ab6474fec0..32693b2b44 100644 --- a/backend/packages/harness/deerflow/community/browser_automation/tools.py +++ b/backend/packages/harness/deerflow/community/browser_automation/tools.py @@ -8,7 +8,9 @@ addressed by a stable ``[ref]`` index, so the model acts on what it just observed instead of guessing selectors. All URLs are SSRF-screened with the shared :func:`validate_public_http_url` -helper (opt-out only for intentional internal targets). +helper (opt-out only for intentional internal targets), and a launched browser +connects only through its pinned egress proxy (:mod:`.egress`), which applies +the same policy to the addresses it actually connects to. """ from __future__ import annotations @@ -27,7 +29,7 @@ from langchain_core.messages import ToolMessage from langgraph.types import Command from deerflow.community.url_safety import resolve_host_addresses as _resolve_host_addresses -from deerflow.community.url_safety import validate_public_http_url +from deerflow.community.url_safety import resolve_public_addresses, validate_public_http_url from deerflow.config import get_app_config from deerflow.config.paths import VIRTUAL_PATH_PREFIX from deerflow.constants import BROWSER_FRAMES_DIRNAME @@ -139,6 +141,7 @@ def _resolve_session(runtime: Runtime, tool_name: str) -> _SessionLease: cdp_url=cdp_url, allow_unguarded_cdp=_as_bool(cfg.get("allow_unguarded_cdp"), False), url_guard=validate_browser_url, + egress_resolver=resolve_browser_egress, pin=True, ) return _SessionLease(manager, thread_id, session) @@ -162,6 +165,24 @@ def validate_browser_url(url: str, *, tool_name: str = "browser_navigate") -> st ) +def resolve_browser_egress(host: str, *, tool_name: str = "browser_navigate") -> list[str]: + """Return the addresses the browser may connect to for *host*. + + The session's egress proxy calls this for every connection Chromium opens, + so the same ``allow_private_addresses`` policy as :func:`validate_browser_url` + decides it, and the browser connects to exactly the addresses vetted here. + Raises ``ValueError`` when the host must be refused. Blocking (DNS). + """ + cfg = _get_tool_config(tool_name) + if _as_bool(cfg.get("allow_private_addresses"), False): + addresses = _resolve_host_addresses(host) + if not addresses: + raise ValueError("Error: URL host could not be resolved") + else: + addresses = resolve_public_addresses(host, action="browse", resolver=_resolve_host_addresses) + return list(dict.fromkeys(str(address) for address in addresses)) + + def _validate_url(tool_name: str, url: str) -> str | None: return validate_browser_url(url, tool_name=tool_name) @@ -255,6 +276,7 @@ async def navigate_and_capture(*, thread_id: str | None, url: str, outputs_path: cdp_url=_as_str(cfg.get("cdp_url")), allow_unguarded_cdp=_as_bool(cfg.get("allow_unguarded_cdp"), False), url_guard=validate_browser_url, + egress_resolver=resolve_browser_egress, ) as session: snapshot = await session.navigate(url) screenshot_path: str | None = None diff --git a/backend/packages/harness/deerflow/community/url_safety.py b/backend/packages/harness/deerflow/community/url_safety.py index 451b0727b5..69c0bd921d 100644 --- a/backend/packages/harness/deerflow/community/url_safety.py +++ b/backend/packages/harness/deerflow/community/url_safety.py @@ -35,6 +35,42 @@ def is_blocked_address(address: ipaddress._BaseAddress) -> bool: return address.is_private or address.is_loopback or address.is_link_local or address.is_reserved or address.is_multicast or address.is_unspecified +def resolve_public_addresses( + hostname: str, + *, + action: str = "connect to", + resolver: Callable[[str], list[ipaddress._BaseAddress]] | None = None, +) -> list[ipaddress._BaseAddress]: + """Resolve *hostname* once and return the addresses a connection may use. + + Raises ``ValueError`` carrying the same ``"Error: ..."`` message + :func:`validate_public_http_url` returns when the host must be refused. A + caller that connects to exactly these addresses, instead of resolving the + name again at connect time, closes the DNS-rebinding window a check-only + screen leaves open. Blocking like :func:`resolve_host_addresses`. + """ + normalized_host = hostname.strip().rstrip(".").lower() + if normalized_host in _BLOCKED_HOSTNAMES: + raise ValueError(f"Error: Refusing to {action} a private or loopback address") + + try: + literal_ip = ipaddress.ip_address(normalized_host) + except ValueError: + literal_ip = None + + if literal_ip is not None: + candidates = [literal_ip] + else: + resolve = resolver or resolve_host_addresses + candidates = resolve(hostname) + if not candidates: + raise ValueError("Error: URL host could not be resolved") + + if any(is_blocked_address(addr) for addr in candidates): + raise ValueError(f"Error: Refusing to {action} a private, loopback, or metadata address") + return candidates + + def validate_public_http_url( url: str, *, @@ -55,7 +91,10 @@ def validate_public_http_url( The check runs at validation time only. A caller that connects later resolves the name again, so a rebinding DNS server can still hand that connect a private address unless the connection is pinned to the vetted - IPs, as ``deerflow.mcp.personal_network`` does. + IPs (:func:`resolve_public_addresses`), as ``deerflow.mcp.personal_network`` + and the browser egress proxy do. Delegated fetch services (crawl4ai, + Browserless, fastcrw) resolve on their own side and cannot be pinned from + here. """ parsed = urlparse(url) if parsed.scheme not in {"http", "https"} or not parsed.netloc: @@ -68,23 +107,8 @@ def validate_public_http_url( if not hostname: return "Error: URL host could not be parsed" - normalized_host = hostname.strip().rstrip(".").lower() - if normalized_host in _BLOCKED_HOSTNAMES: - return f"Error: Refusing to {action} a private or loopback address" - try: - literal_ip = ipaddress.ip_address(normalized_host) - except ValueError: - literal_ip = None - - if literal_ip is not None: - candidates = [literal_ip] - else: - resolve = resolver or resolve_host_addresses - candidates = resolve(hostname) - if not candidates: - return "Error: URL host could not be resolved" - - if any(is_blocked_address(addr) for addr in candidates): - return f"Error: Refusing to {action} a private, loopback, or metadata address" + resolve_public_addresses(hostname, action=action, resolver=resolver) + except ValueError as exc: + return str(exc) return None diff --git a/backend/packages/harness/deerflow/tools/AGENTS.md b/backend/packages/harness/deerflow/tools/AGENTS.md index 2098018267..f623a61931 100644 --- a/backend/packages/harness/deerflow/tools/AGENTS.md +++ b/backend/packages/harness/deerflow/tools/AGENTS.md @@ -59,7 +59,7 @@ Durable MCP task-management tools are added only while the process-local task su - `browser_automation/` - Agentic browser control (stateful `navigate → observe → click/type` loop) via Playwright, distinct from the read-only `web_fetch`/`web_capture` tools. Tools: `browser_navigate`, `browser_snapshot`, `browser_click`, `browser_type`, `browser_get_text`, `browser_back`, `browser_screenshot`, `browser_close` (config `group: browser`). A process-local `BrowserSessionManager` owns one private, loop-affine Playwright event-loop thread (same pattern as the BoxLite provider) so a per-thread browser session survives across turns regardless of the caller's loop (Gateway / TUI / test). Each action returns a fresh page snapshot whose interactive elements are addressed by a stable numeric `[ref]` index (stamped as `data-df-ref` during snapshot), so the model acts on what it just observed instead of holding stale handles or guessing selectors. URLs are SSRF-screened via the shared `validate_public_http_url` (opt-out `allow_private_addresses` only for intentional internal targets). CDP attachment cannot install the request guard on an existing Chrome context, so `cdp_url` fails closed unless the operator explicitly sets `allow_unguarded_cdp: true` for a trusted local browser. Browser REST/Live access also requires an exact non-NULL thread owner, rather than the general legacy shared-thread policy, because retained pages may contain authenticated state. Session admission is a hard `max_sessions` cap: pinned Live/operation sessions are never evicted, and a new thread is rejected when no unpinned session can be closed; one Live viewer owns a session at a time. Optional dependency: `cd backend && uv sync --extra browser && uv run playwright install chromium`; `scripts/detect_uv_extras.py` preserves the extra when `config.yaml` enables `browser_navigate`, and Gateway startup fails fast if configured browser control cannot import Playwright. Tests: `tests/test_browser_automation.py` (mocked tools + a real-Chromium integration test guarded by `importorskip`); `tests/manual_browser_live_check.py` is a manual DeepSeek-driven end-to-end check (not collected by pytest). Live UI input dispatch is kept independent from JPEG capture: non-move actions start a rate-limited background refresh loop, so pointer, wheel, or keyboard input stays responsive while continuous gestures still produce frames throughout the interaction. -The shared SSRF guard (`community/url_safety.py`) resolves hostnames with blocking `socket.getaddrinfo`, so async tools, Gateway routes, and the Playwright request guard (`BrowserSession._install_request_guard`, which runs on the shared browser loop) call `validate_public_http_url` (or a wrapper such as `validate_browser_url`) through `asyncio.to_thread`. The guard checks at validation time only; fetches that re-resolve later are not pinned to the vetted IPs (only `mcp/personal_network.py` pins). The strict blocking-IO gate carries a `socket.getaddrinfo` rule for this; `tests/blocking_io/test_web_tool_url_validation.py` drives each web tool entry point and the request guard. +The shared SSRF guard (`community/url_safety.py`) resolves hostnames with blocking `socket.getaddrinfo`, so async tools, Gateway routes, and the Playwright request guard (`BrowserSession._install_request_guard`, which runs on the shared browser loop) call `validate_public_http_url` (or a wrapper such as `validate_browser_url`) through `asyncio.to_thread`. The guard checks at validation time only, so a launched browser sends every TCP connection (HTTP, HTTPS, WebSocket) through its per-session loopback SOCKS5 proxy (`browser_automation/egress.py`, `bypass: <-loopback>`): Chromium hands it every hostname, and it resolves once via `resolve_browser_egress` → `url_safety.resolve_public_addresses` and connects to exactly those addresses, closing the DNS-rebinding window (`mcp/personal_network.py` pins the same way). Every `acquire_session`/`get_session` caller must pass `egress_resolver`, because a thread's first caller fixes its launch options. WebRTC UDP does not traverse a SOCKS proxy and is not pinned (out of scope, so this is not a complete egress guarantee). Delegated fetch services (crawl4ai, Browserless, fastcrw) resolve on their own side and cannot be pinned; CDP-attached Chrome has neither guard. The strict blocking-IO gate carries a `socket.getaddrinfo` rule for this; `tests/blocking_io/test_web_tool_url_validation.py` drives each web tool entry point, the request guard, and the egress proxy; `tests/test_browser_egress.py` pins the proxy protocol, the session/manager/router wiring, and (with Playwright installed) a real-Chromium rebinding case. Additional providers also live here (`boxlite`, `brave`, `browserless`, `crawl4ai`, `ddg_search`, `e2b_sandbox`, `exa`, `fastcrw`, `groundroute`, `infoquest`, `searxng`, `serper`, `serply`, `sofya`, `tencent_wsa`, `tenki`, `unbrowse`); see each subpackage for specifics. `tencent_wsa` uses Tencent Cloud Web Search's service API key endpoint (`TENCENTCLOUD_WSA_APIKEY`), not Tencent Cloud SecretId/SecretKey signing. Its `max_results` is capped at 50; requests above 10 use Tencent's optional `Cnt` parameter, which needs a Tencent Cloud plan that supports it. E2B bootstrap is required. If it fails, the provider kills and closes the unusable remote sandbox. New sandbox creation raises an error. Warm-pool reclaim and remote discovery discard the sandbox and continue acquisition. E2B mounts remain optional. diff --git a/backend/tests/blocking_io/test_web_tool_url_validation.py b/backend/tests/blocking_io/test_web_tool_url_validation.py index 33039cde45..bd2a1bedb5 100644 --- a/backend/tests/blocking_io/test_web_tool_url_validation.py +++ b/backend/tests/blocking_io/test_web_tool_url_validation.py @@ -6,12 +6,14 @@ traffic. The strict gate's ``socket.getaddrinfo`` rule fails each test if the tool resolves on the loop instead of in a worker thread. """ +import asyncio from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest from deerflow.community.browser_automation import tools as browser_tools +from deerflow.community.browser_automation.egress import BrowserEgressProxy from deerflow.community.browserless import tools as browserless_tools from deerflow.community.crawl4ai import tools as crawl4ai_tools @@ -114,3 +116,24 @@ async def test_browser_request_guard_resolves_off_loop() -> None: await captured["handler"](route) assert route.aborted_with == "blockedbyclient" + + +async def test_browser_egress_proxy_resolves_off_loop() -> None: + # Chromium hands every hostname to the session's egress proxy, which runs on + # the same shared Playwright loop as the request guard. + proxy = BrowserEgressProxy(browser_tools.resolve_browser_egress) + with patch.object(browser_tools, "_get_tool_config", return_value={}): + proxy_url = await proxy.start() + try: + reader, writer = await asyncio.open_connection("127.0.0.1", int(proxy_url.rsplit(":", 1)[1])) + writer.write(b"\x05\x01\x00") + await writer.drain() + assert await reader.readexactly(2) == b"\x05\x00" + writer.write(b"\x05\x01\x00\x03\x05127.1\x00\x50") + await writer.drain() + reply = await reader.readexactly(10) + writer.close() + finally: + await proxy.close() + + assert reply[1] == 0x02 # refused: 127.1 resolves to loopback diff --git a/backend/tests/test_browser_automation.py b/backend/tests/test_browser_automation.py index 83a302a05e..147c2b5bb3 100644 --- a/backend/tests/test_browser_automation.py +++ b/backend/tests/test_browser_automation.py @@ -567,7 +567,7 @@ async def test_ensure_page_serializes_concurrent_rebuilds(): browser.is_connected.return_value = True browser.new_context = AsyncMock(return_value=context) - async def launch(*, headless): + async def launch(*, headless, proxy): launch_started.set() await release_launch.wait() return browser @@ -590,7 +590,7 @@ async def test_ensure_page_serializes_concurrent_rebuilds(): release_launch.set() assert await asyncio.wait_for(asyncio.gather(first, second), timeout=1.0) == [page, page] - chromium.launch.assert_awaited_once_with(headless=True) + chromium.launch.assert_awaited_once_with(headless=True, proxy=None) browser.new_context.assert_awaited_once() context.new_page.assert_awaited_once() diff --git a/backend/tests/test_browser_egress.py b/backend/tests/test_browser_egress.py new file mode 100644 index 0000000000..9c176ab5aa --- /dev/null +++ b/backend/tests/test_browser_egress.py @@ -0,0 +1,461 @@ +"""Pinned browser egress: the SOCKS5 proxy, its policy, and the session wiring. + +The proxy tests drive the real asyncio server with a minimal SOCKS5 client, so +the protocol bytes Chromium exchanges with it are exercised end to end. The last +test repeats the rebinding scenario against a real headless Chromium and is +skipped when Playwright is not installed. +""" + +from __future__ import annotations + +import asyncio +import ipaddress +import socket +import sys +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from types import ModuleType, SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from deerflow.community import url_safety +from deerflow.community.browser_automation import egress +from deerflow.community.browser_automation import session as session_mod +from deerflow.community.browser_automation import tools as browser_tools +from deerflow.community.browser_automation.egress import BrowserEgressProxy +from deerflow.community.browser_automation.session import BrowserSession + + +async def _start_echo_server() -> tuple[asyncio.Server, int]: + async def echo(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + writer.write(await reader.read(1024)) + await writer.drain() + writer.close() + + server = await asyncio.start_server(echo, host="127.0.0.1", port=0) + return server, server.sockets[0].getsockname()[1] + + +async def _socks_connect(proxy_url: str, address_type: int, address: bytes, port: int, *, command: int = 0x01, methods: bytes = b"\x00"): + """Run the client half of a SOCKS5 CONNECT; return (method reply, reply code, reader, writer).""" + proxy_port = int(proxy_url.rsplit(":", 1)[1]) + reader, writer = await asyncio.open_connection("127.0.0.1", proxy_port) + writer.write(bytes([5, len(methods)]) + methods) + await writer.drain() + _version, method = await reader.readexactly(2) + if method == 0xFF: + return method, None, reader, writer + writer.write(bytes([5, command, 0, address_type]) + address + port.to_bytes(2, "big")) + await writer.drain() + reply = await reader.readexactly(10) + return method, reply[1], reader, writer + + +def _domain(host: str) -> bytes: + return bytes([len(host)]) + host.encode("ascii") + + +@pytest.mark.asyncio +async def test_proxy_connects_to_the_vetted_address_not_the_requested_name(): + echo, echo_port = await _start_echo_server() + asked: list[str] = [] + + def resolve(host: str) -> list[str]: + asked.append(host) + return ["127.0.0.1"] + + proxy = BrowserEgressProxy(resolve) + try: + _method, reply, reader, writer = await _socks_connect(await proxy.start(), 0x03, _domain("rebind.example"), echo_port) + assert reply == 0x00 + writer.write(b"ping") + await writer.drain() + assert await reader.read(1024) == b"ping" + writer.close() + finally: + await proxy.close() + echo.close() + assert asked == ["rebind.example"] + + +@pytest.mark.asyncio +async def test_proxy_refuses_a_host_the_resolver_rejects_without_connecting(): + connections: list[str] = [] + + async def record(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + connections.append("upstream") + writer.close() + + upstream = await asyncio.start_server(record, host="127.0.0.1", port=0) + upstream_port = upstream.sockets[0].getsockname()[1] + + def resolve(host: str) -> list[str]: + raise ValueError("Error: Refusing to browse a private, loopback, or metadata address") + + proxy = BrowserEgressProxy(resolve) + try: + _method, reply, _reader, writer = await _socks_connect(await proxy.start(), 0x03, _domain("metadata.example"), upstream_port) + writer.close() + finally: + await proxy.close() + upstream.close() + assert reply == 0x02 # connection not allowed by ruleset + assert connections == [] + + +@pytest.mark.asyncio +async def test_proxy_hands_ip_literals_to_the_resolver(): + echo, echo_port = await _start_echo_server() + asked: list[str] = [] + + def resolve(host: str) -> list[str]: + asked.append(host) + return [host] + + proxy = BrowserEgressProxy(resolve) + try: + _method, reply, _reader, writer = await _socks_connect(await proxy.start(), 0x01, socket.inet_aton("127.0.0.1"), echo_port) + writer.close() + finally: + await proxy.close() + echo.close() + assert reply == 0x00 + assert asked == ["127.0.0.1"] + + +@pytest.mark.asyncio +async def test_proxy_falls_through_to_the_next_vetted_address(): + echo, echo_port = await _start_echo_server() + # The echo server listens on IPv4 loopback only, so the IPv6 loopback answer + # is refused (or unavailable) and only the second vetted address connects. + proxy = BrowserEgressProxy(lambda _host: ["::1", "127.0.0.1"]) + try: + _method, reply, _reader, writer = await _socks_connect(await proxy.start(), 0x03, _domain("multi.example"), echo_port) + writer.close() + finally: + await proxy.close() + echo.close() + assert reply == 0x00 + + +@pytest.mark.asyncio +async def test_proxy_reports_unreachable_when_no_vetted_address_answers(): + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + closed_port = probe.getsockname()[1] + + proxy = BrowserEgressProxy(lambda _host: ["127.0.0.1"]) + try: + _method, reply, _reader, writer = await _socks_connect(await proxy.start(), 0x03, _domain("down.example"), closed_port) + writer.close() + finally: + await proxy.close() + assert reply == 0x04 + + +@pytest.mark.asyncio +async def test_proxy_rejects_commands_other_than_connect(): + resolve = MagicMock() + proxy = BrowserEgressProxy(resolve) + try: + _method, reply, _reader, writer = await _socks_connect(await proxy.start(), 0x03, _domain("bind.example"), 80, command=0x02) + writer.close() + finally: + await proxy.close() + assert reply == 0x07 + resolve.assert_not_called() + + +@pytest.mark.asyncio +async def test_proxy_rejects_clients_that_require_authentication(): + resolve = MagicMock() + proxy = BrowserEgressProxy(resolve) + try: + method, _reply, _reader, writer = await _socks_connect(await proxy.start(), 0x03, _domain("auth.example"), 80, methods=b"\x02") + writer.close() + finally: + await proxy.close() + assert method == 0xFF + resolve.assert_not_called() + + +def _blocking_resolver(): + """A resolver that hangs until released, like a wedged getaddrinfo.""" + release = threading.Event() + + def resolve(_host: str) -> list[str]: + release.wait(timeout=10) + return ["127.0.0.1"] + + return resolve, release + + +@pytest.mark.asyncio +async def test_proxy_reports_a_resolver_fault_as_general_failure(caplog): + def resolve(_host: str) -> list[str]: + raise RuntimeError("config store unavailable") + + proxy = BrowserEgressProxy(resolve) + try: + _method, reply, _reader, writer = await _socks_connect(await proxy.start(), 0x03, _domain("fault.example"), 80) + writer.close() + finally: + await proxy.close() + # 0x01 (general failure), distinct from the 0x02 a policy refusal returns. + assert reply == 0x01 + assert "browser egress resolver failed for fault.example:80" in caplog.text + assert "config store unavailable" in caplog.text + + +@pytest.mark.asyncio +async def test_proxy_bounds_a_stuck_resolution(monkeypatch): + monkeypatch.setattr(egress, "_RESOLVE_TIMEOUT_S", 0.2) + resolve, release = _blocking_resolver() + proxy = BrowserEgressProxy(resolve) + try: + _method, reply, _reader, writer = await asyncio.wait_for(_socks_connect(await proxy.start(), 0x03, _domain("wedged.example"), 80), timeout=2.0) + writer.close() + finally: + release.set() + await proxy.close() + assert reply == 0x04 + + +@pytest.mark.asyncio +async def test_proxy_close_is_bounded_by_a_tunnel_it_could_not_sweep(monkeypatch): + monkeypatch.setattr(egress, "_CLOSE_TIMEOUT_S", 0.2) + proxy = BrowserEgressProxy(MagicMock()) + proxy_port = int((await proxy.start()).rsplit(":", 1)[1]) + reader, writer = await asyncio.open_connection("127.0.0.1", proxy_port) + try: + while not proxy._client_writers: + await asyncio.sleep(0.01) + # Simulate a connection whose handler registered after the sweep: its + # transport stays open in the 15 s handshake read, which would otherwise + # hold Server.wait_closed() and with it session close. + proxy._client_writers.clear() + await asyncio.wait_for(proxy.close(), timeout=1.0) + finally: + writer.close() + + +@pytest.mark.asyncio +async def test_proxy_drops_a_connection_accepted_after_close(): + resolve = MagicMock(side_effect=ValueError("refused")) + proxy = BrowserEgressProxy(resolve) + await proxy.start() + await proxy.close() + reader = asyncio.StreamReader() + reader.feed_data(b"\x05\x01\x00\x05\x01\x00\x03" + _domain("late.example") + (80).to_bytes(2, "big")) + reader.feed_eof() + writer = MagicMock() + writer.drain = AsyncMock() + writer.wait_closed = AsyncMock() + + # A handler scheduled for a connection accepted just before close() runs now. + await asyncio.wait_for(proxy._handle(reader, writer), timeout=1.0) + + writer.close.assert_called_once() + resolve.assert_not_called() + + +@pytest.mark.asyncio +async def test_proxy_close_ends_open_tunnels(): + held = asyncio.Event() + + async def hold(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + held.set() + await reader.read() + writer.close() + + upstream = await asyncio.start_server(hold, host="127.0.0.1", port=0) + upstream_port = upstream.sockets[0].getsockname()[1] + proxy = BrowserEgressProxy(lambda _host: ["127.0.0.1"]) + try: + _method, reply, reader, writer = await _socks_connect(await proxy.start(), 0x03, _domain("idle.example"), upstream_port) + assert reply == 0x00 + await asyncio.wait_for(held.wait(), timeout=1.0) + await asyncio.wait_for(proxy.close(), timeout=2.0) + assert await asyncio.wait_for(reader.read(), timeout=1.0) == b"" + writer.close() + finally: + upstream.close() + + +def test_resolve_public_addresses_returns_the_vetted_answer(): + public = [ipaddress.ip_address("93.184.215.14"), ipaddress.ip_address("2606:2800:21f:cb07:6820:80da:af6b:8b2c")] + + assert url_safety.resolve_public_addresses("example.com", resolver=lambda _host: public) == public + + +@pytest.mark.parametrize( + ("hostname", "answer", "message"), + [ + ("localhost", None, "Error: Refusing to connect to a private or loopback address"), + ("169.254.169.254", None, "Error: Refusing to connect to a private, loopback, or metadata address"), + ("unresolvable.example", [], "Error: URL host could not be resolved"), + ("mixed.example", ["93.184.215.14", "10.0.0.5"], "Error: Refusing to connect to a private, loopback, or metadata address"), + ], +) +def test_resolve_public_addresses_refuses_with_the_validator_message(hostname, answer, message): + def resolver(_host): + assert answer is not None, "literal and blocked hosts must not be resolved" + return [ipaddress.ip_address(address) for address in answer] + + with pytest.raises(ValueError) as exc_info: + url_safety.resolve_public_addresses(hostname, resolver=resolver) + assert str(exc_info.value) == message + + +def test_resolve_browser_egress_applies_the_browse_policy(): + answers = {"public.example": ["93.184.215.14", "93.184.215.14"], "internal.example": ["10.0.0.5"]} + + def resolver(host): + return [ipaddress.ip_address(address) for address in answers[host]] + + with patch.object(browser_tools, "_get_tool_config", return_value={}), patch.object(browser_tools, "_resolve_host_addresses", side_effect=resolver): + assert browser_tools.resolve_browser_egress("public.example") == ["93.184.215.14"] + with pytest.raises(ValueError, match="Refusing to browse a private, loopback, or metadata address"): + browser_tools.resolve_browser_egress("internal.example") + + with patch.object(browser_tools, "_get_tool_config", return_value={"allow_private_addresses": True}), patch.object(browser_tools, "_resolve_host_addresses", side_effect=resolver): + assert browser_tools.resolve_browser_egress("internal.example") == ["10.0.0.5"] + + +def _fake_playwright_modules(): + fake_async_api = ModuleType("playwright.async_api") + fake_async_api.async_playwright = MagicMock() + fake_playwright = ModuleType("playwright") + fake_playwright.async_api = fake_async_api + return patch.dict(sys.modules, {"playwright": fake_playwright, "playwright.async_api": fake_async_api}) + + +def _fake_browser(): + page = MagicMock() + page.is_closed.return_value = False + context = MagicMock() + context.new_page = AsyncMock(return_value=page) + context.route = AsyncMock() + browser = MagicMock() + browser.is_connected.return_value = True + browser.new_context = AsyncMock(return_value=context) + browser.close = AsyncMock() + context.close = AsyncMock() + chromium = MagicMock() + chromium.launch = AsyncMock(return_value=browser) + return chromium + + +@pytest.mark.asyncio +async def test_session_launches_chromium_through_its_egress_proxy_and_closes_it(): + session = BrowserSession(MagicMock(), headless=True, timeout_ms=1000, viewport={"width": 1000, "height": 500}, egress_resolver=lambda _host: []) + chromium = _fake_browser() + session._playwright = SimpleNamespace(chromium=chromium, stop=AsyncMock()) + + with _fake_playwright_modules(): + await session._ensure_page() + + proxy = chromium.launch.await_args.kwargs["proxy"] + assert proxy["server"].startswith("socks5://127.0.0.1:") + assert proxy["bypass"] == "<-loopback>" + proxy_port = int(proxy["server"].rsplit(":", 1)[1]) + _reader, writer = await asyncio.open_connection("127.0.0.1", proxy_port) + writer.close() + + await session._close() + with pytest.raises(OSError): + await asyncio.open_connection("127.0.0.1", proxy_port) + + +@pytest.mark.asyncio +async def test_session_without_egress_resolver_launches_directly(): + session = BrowserSession(MagicMock(), headless=True, timeout_ms=1000, viewport={"width": 1000, "height": 500}) + chromium = _fake_browser() + session._playwright = SimpleNamespace(chromium=chromium) + + with _fake_playwright_modules(): + await session._ensure_page() + + chromium.launch.assert_awaited_once_with(headless=True, proxy=None) + + +@pytest.mark.asyncio +async def test_manager_passes_the_egress_resolver_to_new_sessions(): + manager = session_mod.BrowserSessionManager() + with patch.object(session_mod, "ensure_browser_worker_compatibility"), patch.object(manager, "_ensure_loop", return_value=MagicMock()): + session = manager.get_session("egress-thread", egress_resolver=browser_tools.resolve_browser_egress) + assert session._egress_resolver is browser_tools.resolve_browser_egress + + +@pytest.mark.asyncio +async def test_browser_tools_acquire_sessions_with_pinned_egress(tmp_path): + manager = MagicMock() + page_snapshot = MagicMock(url="https://example.com/", title="Example") + session = MagicMock() + session.navigate = AsyncMock(return_value=page_snapshot) + session.screenshot_bytes = AsyncMock(side_effect=RuntimeError("no screenshot")) + manager.acquire_session.return_value.__enter__.return_value = session + manager.get_session.return_value = session + with ( + patch.object(browser_tools, "_get_tool_config", return_value={}), + patch.object(browser_tools, "get_browser_session_manager", return_value=manager), + patch.object(browser_tools, "_resolve_host_addresses", return_value=[ipaddress.ip_address("93.184.215.14")]), + ): + await browser_tools.navigate_and_capture(thread_id="thread-1", url="https://example.com/", outputs_path=tmp_path) + browser_tools._resolve_session(SimpleNamespace(context={"thread_id": "thread-1"}, state={}), "browser_navigate") + + assert manager.acquire_session.call_args.kwargs["egress_resolver"] is browser_tools.resolve_browser_egress + assert manager.get_session.call_args.kwargs["egress_resolver"] is browser_tools.resolve_browser_egress + + +@pytest.mark.asyncio +async def test_real_chromium_cannot_follow_a_rebinding_answer(): + """A name that passes both URL screens but answers loopback to the connect is refused.""" + pytest.importorskip("playwright.async_api") + + class _Internal(BaseHTTPRequestHandler): + def do_GET(self): + body = b"metadataiam-credentials" + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *_args): + pass + + internal = ThreadingHTTPServer(("127.0.0.1", 0), _Internal) + threading.Thread(target=internal.serve_forever, daemon=True).start() + # Chromium resolves *.localhost to loopback by itself, so without the pinned + # egress proxy this navigation would reach the internal server directly. + url = f"http://rebind.localhost:{internal.server_address[1]}/" + lookups = 0 + + def rebinding_resolver(_host): + nonlocal lookups + lookups += 1 + # The navigate screen and the request guard see a public answer; the + # lookup the connection actually uses sees loopback. + return [ipaddress.ip_address("93.184.215.14" if lookups <= 2 else "127.0.0.1")] + + session_mod.reset_browser_session_manager() + manager = session_mod.get_browser_session_manager() + runtime = SimpleNamespace(context={"thread_id": "rebind-thread"}, state={"thread_data": {}}) + try: + with ( + patch.object(browser_tools, "_get_tool_config", return_value={"timeout_ms": 15000}), + patch.object(browser_tools, "_resolve_host_addresses", side_effect=rebinding_resolver), + patch.object(browser_tools, "_capture_step_screenshot", AsyncMock(return_value=None)), + ): + result = await browser_tools.browser_navigate_tool.coroutine(runtime=runtime, url=url, tool_call_id="call-1") + finally: + await manager.close_session("rebind-thread") + session_mod.reset_browser_session_manager() + internal.shutdown() + + content = result.update["messages"][0].content + assert "ERR_SOCKS_CONNECTION_FAILED" in content + assert "iam-credentials" not in content + # navigate screen, request guard, then the proxy's own lookup for the connect + assert lookups >= 3 diff --git a/backend/tests/test_browser_router.py b/backend/tests/test_browser_router.py index 0309f62e47..63e5084e3f 100644 --- a/backend/tests/test_browser_router.py +++ b/backend/tests/test_browser_router.py @@ -389,3 +389,31 @@ def test_browser_stream_resolves_seed_and_navigate_urls_off_the_loop(): session.navigate.assert_awaited_once_with("https://seed.example/") session.dispatch_input.assert_not_awaited() assert resolved_on_loop == {"seed.example": False, "internal.example": False} + + +def test_browser_stream_acquires_the_session_with_pinned_egress(): + # The first caller's launch options stick to a thread's session, so a Live + # viewer that opens it first must not launch Chromium without the egress proxy. + store = MagicMock() + store.check_access = AsyncMock(return_value=True) + store.get = AsyncMock(return_value={"thread_id": "thread-1", "user_id": "browser-user"}) + app = _browser_ws_app(store) + session = MagicMock() + for method in ("current_url", "start_screencast", "stop_screencast", "tabs"): + setattr(session, method, AsyncMock()) + manager = MagicMock() + manager.acquire_session.return_value.__enter__.return_value = session + + from deerflow.community.browser_automation import resolve_browser_egress + + with ( + patch.object(browser_router, "_authenticate_ws", AsyncMock(return_value=_user())), + patch.object(browser_router, "_browser_tools_enabled", return_value=True), + patch("deerflow.config.get_app_config", return_value=SimpleNamespace(get_tool_config=lambda _name: None)), + patch("deerflow.community.browser_automation.get_browser_session_manager", return_value=manager), + TestClient(app) as client, + client.websocket_connect("/api/threads/thread-1/browser/stream"), + ): + pass + + assert manager.acquire_session.call_args.kwargs["egress_resolver"] is resolve_browser_egress diff --git a/config.example.yaml b/config.example.yaml index 4cc3ad9224..e508c5ade7 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1111,7 +1111,8 @@ tools: # timeout_ms: 30000 # Per-action navigation/interaction timeout # viewport_width: 1280 # viewport_height: 720 - # # allow_private_addresses: false # SSRF guard: keep false in production + # # allow_private_addresses: false # SSRF guard: keep false in production. Also applies to the + # # # pinned egress proxy every launched browser connects through. # # cdp_url: http://127.0.0.1:9222 # Attach to YOUR running Chrome (Codex-style) # # allow_unguarded_cdp: false # REQUIRED true with cdp_url: attaching to an # # # existing Chrome prevents DeerFlow from