diff --git a/tests/tools/test_browser_cdp_tool.py b/tests/tools/test_browser_cdp_tool.py index 205ca2ef9f8..19480070159 100644 --- a/tests/tools/test_browser_cdp_tool.py +++ b/tests/tools/test_browser_cdp_tool.py @@ -388,6 +388,104 @@ def test_dispatch_through_registry(cdp_server): assert result["method"] == "Target.getTargets" +# --------------------------------------------------------------------------- +# Private-network guard +# --------------------------------------------------------------------------- + + +PRIVATE_URL = "http://169.254.169.254/latest/meta-data/" + + +def test_runtime_evaluate_blocked_when_current_page_is_private(monkeypatch): + calls = [] + + monkeypatch.setattr( + browser_cdp_tool, + "_resolve_cdp_endpoint", + lambda: "ws://127.0.0.1:9222/devtools/browser/mock", + ) + + import tools.browser_tool as bt + + monkeypatch.setattr(bt, "_eval_ssrf_guard_active", lambda task_id: True) + monkeypatch.setattr(bt, "_current_page_private_url", lambda task_id: PRIVATE_URL) + + async def fake_call(*args, **kwargs): + calls.append((args, kwargs)) + return {"result": {"value": "private data"}} + + monkeypatch.setattr(browser_cdp_tool, "_cdp_call", fake_call) + + result = json.loads( + browser_cdp_tool.browser_cdp( + method="Runtime.evaluate", + params={"expression": "document.body.innerText"}, + task_id="task-1", + ) + ) + + assert "error" in result + assert PRIVATE_URL in result["error"] + assert "private or internal address" in result["error"] + assert calls == [] + + +def test_page_navigate_to_private_url_blocked_before_cdp(monkeypatch): + calls = [] + + monkeypatch.setattr( + browser_cdp_tool, + "_resolve_cdp_endpoint", + lambda: "ws://127.0.0.1:9222/devtools/browser/mock", + ) + + import tools.browser_tool as bt + + monkeypatch.setattr(bt, "_eval_ssrf_guard_active", lambda task_id: True) + + async def fake_call(*args, **kwargs): + calls.append((args, kwargs)) + return {"frameId": "f"} + + monkeypatch.setattr(browser_cdp_tool, "_cdp_call", fake_call) + + result = json.loads( + browser_cdp_tool.browser_cdp( + method="Page.navigate", + params={"url": PRIVATE_URL}, + task_id="task-1", + ) + ) + + assert "error" in result + assert PRIVATE_URL in result["error"] + assert calls == [] + + +def test_private_guard_inactive_does_not_probe(monkeypatch, cdp_server): + cdp_server.on("Runtime.evaluate", lambda params, sid: {"result": {"value": "ok"}}) + + import tools.browser_tool as bt + + monkeypatch.setattr(bt, "_eval_ssrf_guard_active", lambda task_id: False) + + def fail_probe(task_id): + raise AssertionError("_current_page_private_url must not be probed") + + monkeypatch.setattr(bt, "_current_page_private_url", fail_probe) + + result = json.loads( + browser_cdp_tool.browser_cdp( + method="Runtime.evaluate", + params={"expression": "document.title"}, + task_id="task-1", + ) + ) + + assert result["success"] is True + assert result["result"]["result"]["value"] == "ok" + + # --------------------------------------------------------------------------- # check_fn gating # --------------------------------------------------------------------------- diff --git a/tools/browser_cdp_tool.py b/tools/browser_cdp_tool.py index 5d036c5aa74..ca7497bb62b 100644 --- a/tools/browser_cdp_tool.py +++ b/tools/browser_cdp_tool.py @@ -28,6 +28,19 @@ logger = logging.getLogger(__name__) CDP_DOCS_URL = "https://chromedevtools.github.io/devtools-protocol/" +_CDP_PRIVATE_PAGE_ALLOWED_METHODS = { + # Browser/target inspection does not read the current page body, cookies, + # DOM, storage, or screenshots. Keep these working so the model can list + # tabs or navigate away from a blocked page. + "Browser.getVersion", + "Target.getTargets", + "Target.attachToTarget", + "Target.detachFromTarget", + "Page.navigate", + "Page.reload", + "Page.stopLoading", +} + def _redact_cdp_output(value: Any) -> Any: """Redact browser-originated CDP result data before returning it.""" @@ -101,6 +114,72 @@ def _resolve_cdp_endpoint() -> str: return "" +def _private_page_guard_error(blocked_url: str, method: str) -> str: + return tool_error( + "Blocked: page URL targets a private or internal address " + f"({blocked_url}). Raw CDP method {method!r} could expose private " + "page content or state.", + method=method, + cdp_docs=CDP_DOCS_URL, + ) + + +def _browser_cdp_private_guard( + *, + task_id: str, + method: str, + params: Dict[str, Any], +) -> Optional[str]: + """Apply the browser SSRF/private-page guard to raw CDP calls. + + ``browser_cdp`` is intentionally an escape hatch, but it still shares the + same cloud/private-network boundary as ``browser_snapshot``, + ``browser_console`` and ``browser_eval``. If a cloud browser has landed on + a private/internal URL (for example via a prior eval navigation), raw CDP + calls like ``Runtime.evaluate`` or ``DOM.getDocument`` must not become the + sibling bypass for the guarded browser tools. + """ + try: + from tools import browser_tool as bt # type: ignore[import-not-found] + + if not bt._eval_ssrf_guard_active(task_id): # type: ignore[attr-defined] + return None + + if method == "Page.navigate": + target_url = str((params or {}).get("url") or "").strip() + if target_url and ( + bt._is_always_blocked_url(target_url) # type: ignore[attr-defined] + or not bt._is_safe_url(target_url) # type: ignore[attr-defined] + ): + return tool_error( + "Blocked: CDP Page.navigate target is a private or " + f"internal address ({target_url}).", + method=method, + cdp_docs=CDP_DOCS_URL, + ) + + if method == "Runtime.evaluate": + expression = str((params or {}).get("expression") or "") + blocked_literal = bt._expression_targets_private_url(expression) # type: ignore[attr-defined] + if blocked_literal: + return tool_error( + "Blocked: CDP Runtime.evaluate expression targets a " + f"private or internal address ({blocked_literal}).", + method=method, + cdp_docs=CDP_DOCS_URL, + ) + + if method not in _CDP_PRIVATE_PAGE_ALLOWED_METHODS: + blocked_url = bt._current_page_private_url(task_id) # type: ignore[attr-defined] + if blocked_url: + return _private_page_guard_error(blocked_url, method) + except Exception as exc: # noqa: BLE001 + # Match the existing browser guards' posture: guard probes are + # best-effort and should not break local/custom CDP workflows. + logger.debug("browser_cdp: private-page guard probe failed: %s", exc) + return None + + # --------------------------------------------------------------------------- # Core CDP call # --------------------------------------------------------------------------- @@ -345,16 +424,17 @@ def browser_cdp( JSON string ``{"success": True, "method": ..., "result": {...}}`` on success, or ``{"error": "..."}`` on failure. """ + effective_task_id = task_id or "default" + # --- Route iframe-scoped calls through the supervisor --------------- if frame_id: return _browser_cdp_via_supervisor( - task_id=task_id or "default", + task_id=effective_task_id, frame_id=frame_id, method=method, params=params, timeout=timeout, ) - del task_id # stateless path below if not method or not isinstance(method, str): return tool_error( @@ -392,6 +472,14 @@ def browser_cdp( f"'params' must be an object/dict, got {type(call_params).__name__}" ) + blocked = _browser_cdp_private_guard( + task_id=effective_task_id, + method=method, + params=call_params, + ) + if blocked: + return blocked + try: safe_timeout = float(timeout) if timeout else 30.0 except (TypeError, ValueError):