mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix(browser): pin browser connections to the SSRF-vetted addresses (#6201)
* fix(browser): pin browser connections to the SSRF-vetted addresses The navigate screen and the per-request guard resolve a hostname to vet it, but Chromium resolved the name again to open the connection. A rebinding DNS server could answer both checks with a public address and the connection with a private or cloud-metadata one, and the browser followed it. Launch each private browser through a per-session loopback SOCKS5 proxy. Over SOCKS5 Chromium sends the hostname itself, so the proxy resolves it exactly once through the session's egress resolver and connects only to the addresses that resolver vetted. resolve_browser_egress applies the same allow_private_addresses policy as validate_browser_url, on top of a new url_safety.resolve_public_addresses helper that validate_public_http_url now shares. Loopback is proxied explicitly with bypass "<-loopback>". The tools, navigate_and_capture and the Live WebSocket all pass the resolver, because a thread's first caller fixes its launch options. CDP-attached Chrome is unchanged. Delegated fetch services (crawl4ai, Browserless, fastcrw) still resolve on their own side, which the Gateway cannot pin. * docs(changelog): reference #6201 in the browser egress pinning entry * fix(browser): bound egress proxy faults and close, and record the WebRTC gap Review follow-ups for the pinned browser egress proxy: - A resolver exception other than ValueError escaped the connection handler as an unretrieved task error. Log it with its traceback and reply SOCKS5 general failure (0x01), so a resolver fault stays distinguishable from a policy refusal (0x02). - Bound the resolve with its own timeout. The handshake timeout did not cover it, so a wedged getaddrinfo held the half-open tunnel until Chromium gave up. A timeout replies host unreachable. - Bound close(): Server.wait_closed() also waits for a connection whose handler registered after the writer sweep, which could stall session close and eviction for the 15 s handshake. A handler that starts after close now drops its connection without opening a tunnel. - Docs said a launched browser connects "solely" through the proxy. WebRTC UDP does not traverse a SOCKS proxy, so AGENTS.md, README (en/zh) and both changelogs now scope the guarantee to TCP and name the gap.
This commit is contained in:
Hyeonsang Cho
GitHub
parent
8d8f882b4a
commit
bb4f7ac5e2
13
CHANGELOG.md
13
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
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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",
|
||||
]
|
||||
|
||||
@ -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()
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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.
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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()
|
||||
|
||||
|
||||
461
backend/tests/test_browser_egress.py
Normal file
461
backend/tests/test_browser_egress.py
Normal file
@ -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"<title>metadata</title>iam-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
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user