Guard browser CDP on private pages

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
 # ---------------------------------------------------------------------------

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):