Browser security hardening: SSRF and JS execution guards

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #54132 fix(browser): seal eval-path SSRF bypass + land snapshot/vision guards (#44731)

  merged to main

  files: tests/tools/test_browser_eval_ssrf.py, tests/tools/test_browser_snapshot_ssrf.py, tools/browser_tool.py

  Show the diff

  • tests/tools/test_browser_eval_ssrf.py

    @@ -0,0 +1,229 @@
    +"""Tests that browser_console(expression=...) cannot bypass the SSRF guard.
    +
    +browser_snapshot / browser_vision re-check the page URL before returning
    +content, but ``_browser_eval`` returns arbitrary JS results directly. Two
    +sub-paths could read private content without ever touching snapshot/vision:
    +
    +  1. Direct fetch:  ``fetch('http://127.0.0.1/secret').then(r => r.text())``
    +     — the page URL stays public, so the post-eval recheck can't see it.
    +     Closed by a pre-scan of the expression for private-host URL literals.
    +  2. Navigate-then-read:  ``location.href = 'http://127.0.0.1/'`` then a later
    +     eval reads ``document.body.innerText`` — closed by re-checking the page
    +     URL after the eval runs.
    +
    +This is the sibling fix for the eval return-value path of issue #44731.
    +"""
    +
    +import json
    +
    +import pytest
    +
    +from tools import browser_tool
    +
    +
    +PRIVATE_URL = "http://127.0.0.1:8080/secret"
    +PUBLIC_URL = "https://example.com/page"
    +METADATA_URL = "http://169.254.169.254/latest/meta-data/"
    +
    +
    [redacted:email](autouse=True)
    +def _no_camofox(monkeypatch):
    +    monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: False)
    +    # No supervisor — force the subprocess fallback path by default.
    +    monkeypatch.setattr(browser_tool, "_last_session_key", lambda key: key)
    +
    +
    +def _eval(expression, task_id="test"):
    +    return json.loads(browser_tool._browser_eval(expression, task_id=task_id))
    +
    +
    +# ---------------------------------------------------------------------------
    +# Sub-path 1: direct private-host fetch literal in the expression (pre-scan)
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestExpressionPreScan:
    +    def _guard_on(self, monkeypatch):
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_is_local_sidecar_key", lambda key: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +
    +    def test_blocks_private_fetch_literal(self, monkeypatch):
    +        self._guard_on(monkeypatch)
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: False)
    +        monkeypatch.setattr(browser_tool, "_is_always_blocked_url", lambda url: False)
    +
    +        called = {"n": 0}
    +
    +        def _run(task_id, command, args=None, **kwargs):
    +            called["n"] += 1
    +            return {"success": True, "data": {"result": "leaked-content"}}
    +
    +        monkeypatch.setattr(browser_tool, "_run_browser_command", _run)
    +
    +        result = _eval(f"fetch('{PRIVATE_URL}').then(r => r.text())")
    +        assert result["success"] is False
    +        assert "private or internal address" in result["error"]
    +        assert PRIVATE_URL in result["error"]
    +        # Expression never executed — blocked before any browser command.
    +        assert called["n"] == 0
    +
    +    def test_blocks_metadata_fetch_literal(self, monkeypatch):
    +        self._guard_on(monkeypatch)
    +        # Public-safe to is_safe_url, but the always-blocked floor catches IMDS.
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: True)
    +        monkeypatch.setattr(
    +            browser_tool, "_is_always_blocked_url",
    +            lambda url: "169.254.169.254" in url,
    +        )
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command",
    +            lambda *a, **k: {"success": True, "data": {"result": "creds"}},
    +        )
    +
    +        result = _eval(f"fetch('{METADATA_URL}')")
    +        assert result["success"] is False
    +        assert "private or internal address" in result["error"]
    +
    +    def test_allows_public_fetch_literal(self, monkeypatch):
    +        self._guard_on(monkeypatch)
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: True)
    +        monkeypatch.setattr(browser_tool, "_is_always_blocked_url", lambda url: False)
    +        # After the (public) eval, the page-URL recheck must also see a public URL.
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command",
    +            lambda task_id, command, args=None, **k: (
    +                {"success": True, "data": {"result": PUBLIC_URL}}
    +                if args == ["window.location.href"]
    +                else {"success": True, "data": {"result": "ok"}}
    +            ),
    +        )
    +
    +        result = _eval(f"fetch('{PUBLIC_URL}').then(r => r.text())")
    +        assert result["success"] is True
    +        assert result["result"] == "ok"
    +
    +    def test_skips_prescan_for_local_backend(self, monkeypatch):
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: True)
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command",
    +            lambda *a, **k: {"success": True, "data": {"result": "local-ok"}},
    +        )
    +        result = _eval(f"fetch('{PRIVATE_URL}')")
    +        assert result["success"] is True
    +        assert result["result"] == "local-ok"
    +
    +    def test_skips_prescan_for_local_sidecar(self, monkeypatch):
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_is_local_sidecar_key", lambda key: True)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command",
    +            lambda *a, **k: {"success": True, "data": {"result": "sidecar-ok"}},
    +        )
    +        result = _eval(f"fetch('{PRIVATE_URL}')")
    +        assert result["success"] is True
    +
    +    def test_skips_prescan_when_allow_private(self, monkeypatch):
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_is_local_sidecar_key", lambda key: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: True)
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command",
    +            lambda *a, **k: {"success": True, "data": {"result": "allowed"}},
    +        )
    +        result = _eval(f"fetch('{PRIVATE_URL}')")
    +        assert result["success"] is True
    +
    +
    +# ---------------------------------------------------------------------------
    +# Sub-path 2: navigate-then-read (post-eval page-URL recheck)
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestPostEvalPageRecheck:
    +    def _guard_on(self, monkeypatch):
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_is_local_sidecar_key", lambda key: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +
    +    def test_blocks_when_page_navigated_private(self, monkeypatch):
    +        self._guard_on(monkeypatch)
    +        # Expression itself has no URL literal (reads the DOM), so the pre-scan
    +        # passes; the danger is that the page was navigated to a private URL by
    +        # an earlier eval. The recheck must catch it.
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: False)
    +        monkeypatch.setattr(browser_tool, "_is_always_blocked_url", lambda url: False)
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command",
    +            lambda task_id, command, args=None, **k: (
    +                {"success": True, "data": {"result": PRIVATE_URL}}
    +                if args == ["window.location.href"]
    +                else {"success": True, "data": {"result": "secret DOM text"}}
    +            ),
    +        )
    +
    +        result = _eval("document.body.innerText")
    +        assert result["success"] is False
    +        assert "private or internal address" in result["error"]
    +        assert PRIVATE_URL in result["error"]
    +
    +    def test_allows_when_page_public(self, monkeypatch):
    +        self._guard_on(monkeypatch)
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: True)
    +        monkeypatch.setattr(browser_tool, "_is_always_blocked_url", lambda url: False)
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command",
    +            lambda task_id, command, args=None, **k: (
    +                {"success": True, "data": {"result": PUBLIC_URL}}
    +                if args == ["window.location.href"]
    +                else {"success": True, "data": {"result": "public DOM text"}}
    +            ),
    +        )
    +
    +        result = _eval("document.body.innerText")
    +        assert result["success"] is True
    +        assert result["result"] == "public DOM text"
    +
    +    def test_fail_open_when_url_probe_fails(self, monkeypatch):
    +        """If the window.location.href probe errors, don't block (fail-open)."""
    +        self._guard_on(monkeypatch)
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: False)
    +        monkeypatch.setattr(browser_tool, "_is_always_blocked_url", lambda url: False)
    +
    +        def _run(task_id, command, args=None, **k):
    +            if args == ["window.location.href"]:
    +                return {"success": False, "error": "CDP probe failed"}
    +            return {"success": True, "data": {"result": "dom text"}}
    +
    +        monkeypatch.setattr(browser_tool, "_run_browser_command", _run)
    +
    +        result = _eval("document.body.innerText")
    +        assert result["success"] is True
    +        assert result["result"] == "dom text"
    +
    +
    +# ---------------------------------------------------------------------------
    +# Helper-level unit tests
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestExpressionScanHelper:
    +    def test_returns_first_private_literal(self, monkeypatch):
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: "127.0.0.1" not in url)
    +        monkeypatch.setattr(browser_tool, "_is_always_blocked_url", lambda url: False)
    +        out = browser_tool._expression_targets_private_url(
    +            "fetch('https://example.com'); fetch('http://127.0.0.1/x')"
    +        )
    +        assert out == "http://127.0.0.1/x"
    +
    +    def test_none_when_no_url(self, monkeypatch):
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: True)
    +        monkeypatch.setattr(browser_tool, "_is_always_blocked_url", lambda url: False)
    +        assert browser_tool._expression_targets_private_url("document.title") is None
    +
    +    def test_strips_trailing_punctuation(self, monkeypatch):
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: False)
    +        monkeypatch.setattr(browser_tool, "_is_always_blocked_url", lambda url: False)
    +        out = browser_tool._expression_targets_private_url("location.href='http://10.0.0.1/';")
    +        assert out == "http://10.0.0.1/"

  • tests/tools/test_browser_snapshot_ssrf.py

    @@ -0,0 +1,438 @@
    +"""Tests that browser_snapshot blocks content from eval-navigated private pages.
    +
    +When browser_console() changes location.href to a private/internal address,
    +browser_snapshot() must detect this and return an error instead of exposing
    +the private page content.
    +
    +This is the fix for the SSRF bypass described in issue #44731.
    +"""
    +
    +import json
    +
    +import pytest
    +
    +from tools import browser_tool
    +
    +
    +def _make_snapshot_result(snapshot="Public page content", refs=None):
    +    """Return a mock successful snapshot result."""
    +    return {
    +        "success": True,
    +        "data": {
    +            "snapshot": snapshot,
    +            "refs": refs or {"@e1": {"role": "heading", "name": "Public"}},
    +        },
    +    }
    +
    +
    +def _make_eval_result(result):
    +    """Return a mock successful eval result."""
    +    return {"success": True, "data": {"result": result}}
    +
    +
    +# ---------------------------------------------------------------------------
    +# browser_snapshot: private-network guard after eval navigation
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestBrowserSnapshotPrivateNetworkGuard:
    +    """browser_snapshot must block content from private pages navigated via eval."""
    +
    +    PRIVATE_URL = "http://127.0.0.1:8080/secret"
    +    PUBLIC_URL = "https://example.com/page"
    +
    +    @pytest.fixture(autouse=True)
    +    def _setup(self, monkeypatch):
    +        """Common patches for snapshot SSRF tests."""
    +        monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: False)
    +        monkeypatch.setattr(
    +            browser_tool,
    +            "_get_session_info",
    +            lambda task_id: {
    +                "session_name": f"s_{task_id}",
    +                "bb_session_id": None,
    +                "cdp_url": None,
    +                "features": {"local": True},
    +                "_first_nav": False,
    +            },
    +        )
    +
    +    def test_blocks_private_url_after_eval_navigation(self, monkeypatch):
    +        """Snapshot must block when current page URL is private."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: False)
    +
    +        call_count = {"n": 0}
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            call_count["n"] += 1
    +            if command == "snapshot":
    +                return _make_snapshot_result()
    +            elif command == "eval":
    +                return _make_eval_result(self.PRIVATE_URL)
    +            return {"success": False, "error": "unknown command"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result = json.loads(browser_browser_snapshot(task_id="test"))
    +        assert result["success"] is False
    +        assert "private or internal address" in result["error"]
    +        assert self.PRIVATE_URL in result["error"]
    +        # Must have called eval to check URL
    +        assert call_count["n"] == 2  # snapshot + eval
    +
    +    def test_allows_public_url_after_eval_navigation(self, monkeypatch):
    +        """Snapshot must succeed when current page URL is public."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: True)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "snapshot":
    +                return _make_snapshot_result()
    +            elif command == "eval":
    +                return _make_eval_result(self.PUBLIC_URL)
    +            return {"success": False, "error": "unknown command"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result = json.loads(browser_browser_snapshot(task_id="test"))
    +        assert result["success"] is True
    +        assert "snapshot" in result
    +
    +    def test_skips_check_in_local_backend_mode(self, monkeypatch):
    +        """Local backend mode skips SSRF check entirely."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: True)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "snapshot":
    +                return _make_snapshot_result()
    +            return {"success": False, "error": "should not be called"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result = json.loads(browser_browser_snapshot(task_id="test"))
    +        assert result["success"] is True
    +        assert "snapshot" in result
    +
    +
    +    def test_skips_check_for_local_sidecar_session(self, monkeypatch):
    +        """Local sidecar sessions can legitimately access private URLs."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +        # Simulate the effective_task_id being a local sidecar key
    +        monkeypatch.setattr(browser_tool, "_is_local_sidecar_key", lambda key: True)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "snapshot":
    +                return _make_snapshot_result()
    +            return {"success": False, "error": "should not be called"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result = json.loads(browser_browser_snapshot(task_id="test"))
    +        assert result["success"] is True
    +        assert "snapshot" in result
    +
    +    def test_skips_check_when_private_urls_allowed(self, monkeypatch):
    +        """When allow_private_urls is enabled, SSRF check is skipped."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: True)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "snapshot":
    +                return _make_snapshot_result()
    +            return {"success": False, "error": "should not be called"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result = json.loads(browser_browser_snapshot(task_id="test"))
    +        assert result["success"] is True
    +        assert "snapshot" in result
    +
    +    def test_handles_eval_failure_gracefully(self, monkeypatch):
    +        """If URL eval fails, snapshot should still succeed (fail-open)."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "snapshot":
    +                return _make_snapshot_result()
    +            elif command == "eval":
    +                return {"success": False, "error": "eval failed"}
    +            return {"success": False, "error": "unknown"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result = json.loads(browser_browser_snapshot(task_id="test"))
    +        # Should succeed — eval failure means we can't determine URL, fail-open
    +        assert result["success"] is True
    +
    +    def test_handles_empty_url_result(self, monkeypatch):
    +        """If URL eval returns empty string, snapshot should succeed."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "snapshot":
    +                return _make_snapshot_result()
    +            elif command == "eval":
    +                return _make_eval_result("")
    +            return {"success": False, "error": "unknown"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result = json.loads(browser_browser_snapshot(task_id="test"))
    +        assert result["success"] is True
    +
    +    def test_handles_eval_exception(self, monkeypatch):
    +        """If URL eval raises an exception, snapshot should succeed."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "snapshot":
    +                return _make_snapshot_result()
    +            elif command == "eval":
    +                raise RuntimeError("CDP connection lost")
    +            return {"success": False, "error": "unknown"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result = json.loads(browser_browser_snapshot(task_id="test"))
    +        assert result["success"] is True
    +
    +    def test_blocks_loopback_url(self, monkeypatch):
    +        """Loopback URLs (localhost) must be blocked."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: False)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "snapshot":
    +                return _make_snapshot_result()
    +            elif command == "eval":
    +                return _make_eval_result("http://localhost:3000/admin")
    +            return {"success": False, "error": "unknown"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result = json.loads(browser_browser_snapshot(task_id="test"))
    +        assert result["success"] is False
    +        assert "private or internal address" in result["error"]
    +
    +    def test_blocks_private_ip_range(self, monkeypatch):
    +        """Private IP ranges (10.x, 172.16.x, 192.168.x) must be blocked."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: False)
    +
    +        for private_ip in ["http://10.0.0.1/api", "http://172.16.0.1/admin", "http://192.168.1.1/config"]:
    +            def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +                if command == "snapshot":
    +                    return _make_snapshot_result()
    +                elif command == "eval":
    +                    return _make_eval_result(private_ip)
    +                return {"success": False, "error": "unknown"}
    +
    +            monkeypatch.setattr(
    +                browser_tool, "_run_browser_command", mock_run_browser_command
    +            )
    +
    +            result = json.loads(browser_browser_snapshot(task_id="test"))
    +            assert result["success"] is False, f"Expected block for {private_ip}"
    +            assert "private or internal address" in result["error"]
    +
    +
    +# Helper to avoid name collision with the actual function
    +def browser_browser_snapshot(**kwargs):
    +    from tools.browser_tool import browser_snapshot
    +    return browser_snapshot(**kwargs)
    +
    +
    +def browser_browser_vision(**kwargs):
    +    from tools.browser_tool import browser_vision
    +    return browser_vision(**kwargs)
    +
    +
    +def _make_screenshot_result(path="/tmp/test_screenshot.png"):
    +    """Return a mock successful screenshot result."""
    +    return {"success": True, "data": {"path": path}}
    +
    +
    +# ---------------------------------------------------------------------------
    +# browser_vision: private-network guard after eval navigation
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestBrowserVisionPrivateNetworkGuard:
    +    """browser_vision must block screenshots from private pages navigated via eval."""
    +
    +    PRIVATE_URL = "http://127.0.0.1:8080/secret"
    +    PUBLIC_URL = "https://example.com/page"
    +
    +    @pytest.fixture(autouse=True)
    +    def _setup(self, monkeypatch):
    +        """Common patches for vision SSRF tests."""
    +        monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: False)
    +
    +    def test_blocks_private_url_after_eval_navigation(self, monkeypatch):
    +        """Vision must block when current page URL is private."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: False)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "eval":
    +                return _make_eval_result(self.PRIVATE_URL)
    +            return {"success": False, "error": "should not reach screenshot"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result = json.loads(browser_browser_vision(question="what do you see", task_id="test"))
    +        assert result["success"] is False
    +        assert "private or internal address" in result["error"]
    +        assert self.PRIVATE_URL in result["error"]
    +
    +    def test_allows_public_url_after_eval_navigation(self, monkeypatch):
    +        """Vision must proceed when current page URL is public."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: True)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "eval":
    +                return _make_eval_result(self.PUBLIC_URL)
    +            elif command == "screenshot":
    +                return _make_screenshot_result()
    +            return {"success": False, "error": "unknown"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +        # Screenshot file won't exist — that's fine, function returns error
    +        # but the important thing is the guard didn't block it.
    +
    +        result_raw = browser_browser_vision(question="what do you see", task_id="test")
    +        result = json.loads(result_raw)
    +        # Guard passed; function continues to screenshot path.
    +        # Since screenshot file doesn't exist, it returns a file-not-found error,
    +        # NOT the "private or internal address" error.
    +        assert "private or internal address" not in result.get("error", "")
    +
    +    def test_skips_check_in_local_backend_mode(self, monkeypatch):
    +        """Local backend mode skips SSRF check entirely."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: True)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "screenshot":
    +                return _make_screenshot_result()
    +            return {"success": False, "error": "should not be called"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result_raw = browser_browser_vision(question="what", task_id="test")
    +        result = json.loads(result_raw)
    +        assert "private or internal address" not in result.get("error", "")
    +
    +
    +    def test_skips_check_for_local_sidecar_session(self, monkeypatch):
    +        """Local sidecar sessions can legitimately access private URLs."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +        # Simulate the effective_task_id being a local sidecar key
    +        monkeypatch.setattr(browser_tool, "_is_local_sidecar_key", lambda key: True)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "screenshot":
    +                return _make_screenshot_result()
    +            return {"success": False, "error": "should not be called"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result_raw = browser_browser_vision(question="what", task_id="test")
    +        result = json.loads(result_raw)
    +        assert "private or internal address" not in result.get("error", "")
    +
    +    def test_skips_check_when_private_urls_allowed(self, monkeypatch):
    +        """When allow_private_urls is enabled, SSRF check is skipped."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: True)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "screenshot":
    +                return _make_screenshot_result()
    +            return {"success": False, "error": "should not be called"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result_raw = browser_browser_vision(question="what", task_id="test")
    +        result = json.loads(result_raw)
    +        assert "private or internal address" not in result.get("error", "")
    +
    +    def test_handles_eval_failure_gracefully(self, monkeypatch):
    +        """If URL eval fails, vision should still proceed (fail-open)."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "eval":
    +                return {"success": False, "error": "eval failed"}
    +            elif command == "screenshot":
    +                return _make_screenshot_result()
    +            return {"success": False, "error": "unknown"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result_raw = browser_browser_vision(question="what", task_id="test")
    +        result = json.loads(result_raw)
    +        assert "private or internal address" not in result.get("error", "")
    +
    +    def test_handles_eval_exception(self, monkeypatch):
    +        """If URL eval raises an exception, vision should still proceed."""
    +        monkeypatch.setattr(browser_tool, "_is_local_backend", lambda: False)
    +        monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False)
    +
    +        def mock_run_browser_command(task_id, command, args=None, **kwargs):
    +            if command == "eval":
    +                raise RuntimeError("CDP connection lost")
    +            elif command == "screenshot":
    +                return _make_screenshot_result()
    +            return {"success": False, "error": "unknown"}
    +
    +        monkeypatch.setattr(
    +            browser_tool, "_run_browser_command", mock_run_browser_command
    +        )
    +
    +        result_raw = browser_browser_vision(question="what", task_id="test")
    +        result = json.loads(result_raw)
    +        assert "private or internal address" not in result.get("error", "")
    \ No newline at end of file

  • tools/browser_tool.py

    @@ -2695,6 +2695,37 @@ def browser_snapshot(
             snapshot_text = data.get("snapshot", "")
             refs = data.get("refs", {})
     
    +        # ── Private-network guard: block snapshots from eval-navigated private pages ──
    +        # After any eval (browser_console) that may have changed location.href to a
    +        # private/internal address, the snapshot would expose private page content.
    +        # Re-check the current URL before returning the snapshot.
    +        if (
    +            not _is_local_backend()
    +            and not _is_local_sidecar_key(effective_task_id)
    +            and not _allow_private_urls()
    +        ):
    +            try:
    +                _url_result = _run_browser_command(
    +                    effective_task_id, "eval", ["window.location.href"],
    +                    timeout=5, _engine_override="auto",
    +                )
    +                if _url_result.get("success"):
    +                    _current_url = (
    +                        _url_result.get("data", {}).get("result", "")
    +                        .strip().strip('"').strip("'")
    +                    )
    +                    if _current_url and not _is_safe_url(_current_url):
    +                        return json.dumps({
    +                            "success": False,
    +                            "error": (
    +                                "Blocked: page URL targets a private or internal address "
    +                                f"({_current_url}). This may have been caused by a "
    +                                "JavaScript navigation via browser_console."
    +                            ),
    +                        }, ensure_ascii=False)
    +            except Exception as _url_exc:
    +                logger.debug("browser_snapshot: URL safety check failed (%s)", _url_exc)
    +
             # Check if snapshot needs summarization
             if len(snapshot_text) > SNAPSHOT_SUMMARIZE_THRESHOLD and user_task:
                 snapshot_text = _extract_relevant_content(snapshot_text, user_task)
    @@ -3000,6 +3031,74 @@ def browser_console(clear: bool = False, expression: Optional[str] = None, task_
         return json.dumps(response, ensure_ascii=False)
     
     
    +def _eval_ssrf_guard_active(effective_task_id: str) -> bool:
    +    """Return True when eval-driven private-network access must be guarded.
    +
    +    Matches the gating used by ``browser_navigate`` / ``browser_snapshot`` /
    +    ``browser_vision``: the SSRF guard is only meaningful for non-local
    +    backends (cloud browser, or a containerized terminal whose browser-on-host
    +    can reach internal networks the terminal can't), and is skipped for local
    +    sidecar sessions and when ``allow_private_urls`` is set.
    +    """
    +    return (
    +        not _is_local_backend()
    +        and not _is_local_sidecar_key(effective_task_id)
    +        and not _allow_private_urls()
    +    )
    +
    +
    +# URL-shaped literals embedded in a JS expression (http/https only).  Used to
    +# pre-screen ``browser_console(expression=...)`` calls that fetch/XHR/navigate
    +# to a private host directly — that path never updates ``location.href`` so the
    +# post-eval page-URL recheck below can't see it.
    +_JS_URL_LITERAL_RE = re.compile(r"""https?://[^\s'"`)\]<>]+""", re.IGNORECASE)
    +
    +
    +def _expression_targets_private_url(expression: str) -> Optional[str]:
    +    """Return the first private/always-blocked URL literal in a JS expression.
    +
    +    Best-effort: scans for ``http(s)://...`` literals (fetch/XHR/navigation
    +    targets the agent may have embedded) and returns the first one that targets
    +    a private/internal address or the always-blocked cloud-metadata floor.
    +    Returns ``None`` when no such literal is found.
    +    """
    +    if not isinstance(expression, str):
    +        return None
    +    for match in _JS_URL_LITERAL_RE.findall(expression):
    +        candidate = match.rstrip(".,;")
    +        if _is_always_blocked_url(candidate) or not _is_safe_url(candidate):
    +            return candidate
    +    return None
    +
    +
    +def _current_page_private_url(effective_task_id: str) -> Optional[str]:
    +    """Return the current page URL when it targets a private/internal address.
    +
    +    Reads ``window.location.href`` via a low-cost eval and returns it when the
    +    page has been navigated (e.g. via ``location.href = '...'`` in a prior
    +    eval) to an address the SSRF guard would reject.  Returns ``None`` when the
    +    page is public, the URL can't be determined, or the check errors (fail-open
    +    on probe failure, matching the snapshot/vision guards).
    +    """
    +    try:
    +        url_result = _run_browser_command(
    +            effective_task_id, "eval", ["window.location.href"],
    +            timeout=5, _engine_override="auto",
    +        )
    +        if url_result.get("success"):
    +            current_url = (
    +                url_result.get("data", {}).get("result", "")
    +                .strip().strip('"').strip("'")
    +            )
    +            if current_url and (
    +                _is_always_blocked_url(current_url) or not _is_safe_url(current_url)
    +            ):
    +                return current_url
    +    except Exception as exc:
    +        logger.debug("_current_page_private_url: probe failed (%s)", exc)
    +    return None
    +
    +
     def _browser_eval(expression: str, task_id: Optional[str] = None) -> str:
         """Evaluate a JavaScript expression in the page context and return the result."""
         if _is_camofox_mode():
    @@ -3007,6 +3106,27 @@ def _browser_eval(expression: str, task_id: Optional[str] = None) -> str:
     
         effective_task_id = _last_session_key(task_id or "default")
     
    +    # ── Private-network guard (eval return-value path) ──────────────────────
    +    # browser_snapshot / browser_vision re-check the page URL before returning
    +    # content, but eval returns arbitrary JS results directly — an attacker can
    +    # read a private page via `fetch('http://127.0.0.1/secret')` or by reading
    +    # the DOM after `location.href = 'http://127.0.0.1/'`, never touching
    +    # snapshot/vision.  Close both sub-paths on the same gating condition:
    +    #   1. Pre-scan the expression for private-host URL literals (direct fetch).
    +    #   2. After eval, re-check the page URL (navigate-then-read).
    +    if _eval_ssrf_guard_active(effective_task_id):
    +        blocked_literal = _expression_targets_private_url(expression)
    +        if blocked_literal:
    +            return json.dumps({
    +                "success": False,
    +                "error": (
    +                    "Blocked: JavaScript expression targets a private or "
    +                    f"internal address ({blocked_literal}). Reading internal "
    +                    "endpoints via browser_console is not permitted in this "
    +                    "browser mode."
    +                ),
    +            }, ensure_ascii=False)
    +
         # --- Fast path: route through the supervisor's persistent CDP WS ---------
         # When a CDPSupervisor is alive for this task_id, ``Runtime.evaluate`` runs
         # on the already-connected WebSocket — zero subprocess startup cost vs
    @@ -3028,6 +3148,20 @@ def _browser_eval(expression: str, task_id: Optional[str] = None) -> str:
                             parsed = json.loads(raw_result)
                         except (json.JSONDecodeError, ValueError):
                             pass  # keep as string
    +                # Post-eval page-URL recheck: if this (or a prior) eval
    +                # navigated the page to a private address, withhold the result.
    +                if _eval_ssrf_guard_active(effective_task_id):
    +                    _blocked_url = _current_page_private_url(effective_task_id)
    +                    if _blocked_url:
    +                        return json.dumps({
    +                            "success": False,
    +                            "error": (
    +                                "Blocked: page URL targets a private or internal "
    +                                f"address ({_blocked_url}). This may have been "
    +                                "caused by a JavaScript navigation via "
    +                                "browser_console."
    +                            ),
    +                        }, ensure_ascii=False)
                     response = {
                         "success": True,
                         "result": parsed,
    @@ -3103,6 +3237,19 @@ def _browser_eval(expression: str, task_id: Optional[str] = None) -> str:
             "result": parsed,
             "result_type": type(parsed).__name__,
         }
    +    # Post-eval page-URL recheck: if this (or a prior) eval navigated the page
    +    # to a private address, withhold the result (mirrors the supervisor path).
    +    if _eval_ssrf_guard_active(effective_task_id):
    +        _blocked_url = _current_page_private_url(effective_task_id)
    +        if _blocked_url:
    +            return json.dumps({
    +                "success": False,
    +                "error": (
    +                    "Blocked: page URL targets a private or internal address "
    +                    f"({_blocked_url}). This may have been caused by a "
    +                    "JavaScript navigation via browser_console."
    +                ),
    +            }, ensure_ascii=False)
         return json.dumps(_copy_fallback_warning(response, result), ensure_ascii=False, default=str)
     
     
    @@ -3284,6 +3431,37 @@ def browser_vision(question: str, annotate: bool = False, task_id: Optional[str]
         screenshot_path = screenshots_dir / f"browser_screenshot_{uuid_mod.uuid4().hex}.png"
         effective_task_id = _last_session_key(task_id or "default")
     
    +    # ── Private-network guard: block vision from eval-navigated private pages ──
    +    # After any eval (browser_console) that may have changed location.href to a
    +    # private/internal address, the screenshot would expose private page content
    +    # to the vision model.  Re-check the current URL before capturing anything.
    +    if (
    +        not _is_local_backend()
    +        and not _is_local_sidecar_key(effective_task_id)
    +        and not _allow_private_urls()
    +    ):
    +        try:
    +            _url_result = _run_browser_command(
    +                effective_task_id, "eval", ["window.location.href"],
    +                timeout=5, _engine_override="auto",
    +            )
    +            if _url_result.get("success"):
    +                _current_url = (
    +                    _url_result.get("data", {}).get("result", "")
    +                    .strip().strip('"').strip("'")
    +                )
    +                if _current_url and not _is_safe_url(_current_url):
    +                    return json.dumps({
    +                        "success": False,
    +                        "error": (
    +                            "Blocked: page URL targets a private or internal address "
    +                            f"({_current_url}). This may have been caused by a "
    +                            "JavaScript navigation via browser_console."
    +                        ),
    +                    }, ensure_ascii=False)
    +        except Exception as _url_exc:
    +            logger.debug("browser_vision: URL safety check failed (%s)", _url_exc)
    +
         # Lightpanda has no graphical renderer — pre-route screenshots to Chrome
         # via the fallback helper instead of letting the normal path fail with a
         # CDP error or return a placeholder PNG.  The normal analysis path below