diff --git a/coworker/connectors/browser_automation.py b/coworker/connectors/browser_automation.py index 93236ebac..2dbebdde6 100644 --- a/coworker/connectors/browser_automation.py +++ b/coworker/connectors/browser_automation.py @@ -20,6 +20,39 @@ from ..web.guard import check_url +def _guard_route(route: Any) -> None: + """Playwright route handler: vet every NAVIGATION at request time against the same + address guard web_fetch uses, and abort the ones it refuses. + + ``check_url`` in ``browser_open_url`` only sees the model-supplied URL. The browser then + follows redirects and runs page JavaScript, and a later navigation (a 30x hop, a + meta-refresh, ``location = …``) to ``127.0.0.1`` / ``169.254.169.254`` is never re-vetted + before ``browser_read_page`` / ``browser_screenshot`` lift its content into the agent — + the gap ``redirect_refusal`` (OPE-124) can only patch after the fact for the single + ``goto``. Gating navigation requests here closes it at the source and narrows the + DNS-rebinding window (the handler sees the concrete request that is about to go out). + + Subresource fetches (images, scripts, XHR) are not gated: they can't pull internal + content back into the agent's context, and resolving DNS for every one would be dozens of + lookups per page. A resolver-pinning proxy is the larger design that would also cover + those; this closes the read-back path. + """ + request = route.request + try: + is_navigation = request.is_navigation_request() + except Exception: + is_navigation = True # unknown shape → treat as a navigation and fail closed + if is_navigation: + try: + blocked = check_url(request.url) + except Exception: + blocked = "address guard error" # a guard failure must fail closed + if blocked: + route.abort("blockedbyclient") + return + route.continue_() + + def _meta( name: str, *, approval: bool = False, capabilities: Optional[list[str]] = None ): @@ -128,6 +161,9 @@ def page(self): self._context = self._browser.new_context( viewport={"width": 1280, "height": 900} ) + # Vet every navigation (redirect / JS / meta-refresh), not just the URL the + # model passed to browser_open_url, before its request goes out. + self._context.route("**/*", _guard_route) self._page = self._context.new_page() self._touch( open=True, status="open", last_action="open browser", last_error="" diff --git a/tests/test_browser_url_guard.py b/tests/test_browser_url_guard.py new file mode 100644 index 000000000..7980500cd --- /dev/null +++ b/tests/test_browser_url_guard.py @@ -0,0 +1,114 @@ +"""browser_open_url's address guard must cover EVERY navigation, not just the model URL. + +check_url vets the URL the model supplied; the browser then follows redirects and runs page +JS, and a later navigation to 127.0.0.1 / 169.254.169.254 was never re-vetted before +browser_read_page / browser_screenshot lifted its content into the agent. The context-level +route guard aborts blocked navigations at request time. These tests exercise that handler +directly (no Playwright needed). +""" + +from __future__ import annotations + +import socket + +import pytest + +from coworker.connectors.browser_automation import _guard_route +from coworker.web import guard + + +def _resolves_to(monkeypatch, ip: str): + monkeypatch.setattr( + guard.socket, + "getaddrinfo", + lambda *a, **k: [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, 80))], + ) + + +class _FakeRequest: + def __init__(self, url: str, is_nav: bool = True): + self.url = url + self._is_nav = is_nav + + def is_navigation_request(self) -> bool: + return self._is_nav + + +class _FakeRoute: + def __init__(self, url: str, is_nav: bool = True): + self.request = _FakeRequest(url, is_nav) + self.aborted = None + self.continued = False + + def abort(self, reason: str | None = None): + self.aborted = reason + + def continue_(self): + self.continued = True + + +@pytest.mark.parametrize( + "url", + [ + "http://127.0.0.1:11434/api/tags", + "http://localhost:8000/", + "http://169.254.169.254/latest/meta-data/", + "http://[::1]:8080/", + ], +) +def test_navigation_to_internal_literal_is_aborted(url): + route = _FakeRoute(url, is_nav=True) + _guard_route(route) + assert route.aborted == "blockedbyclient" + assert route.continued is False + + +def test_navigation_to_host_resolving_to_loopback_is_aborted(monkeypatch): + # DNS rebinding / split-horizon: a public-looking name that resolves to loopback. + _resolves_to(monkeypatch, "127.0.0.1") + route = _FakeRoute("http://sneaky.example.com/", is_nav=True) + _guard_route(route) + assert route.aborted == "blockedbyclient" + + +def test_navigation_to_metadata_host_is_aborted(monkeypatch): + _resolves_to(monkeypatch, "169.254.169.254") + route = _FakeRoute("http://metadata.example.com/", is_nav=True) + _guard_route(route) + assert route.aborted == "blockedbyclient" + + +def test_allowed_public_navigation_continues(monkeypatch): + _resolves_to(monkeypatch, "93.184.216.34") + route = _FakeRoute("https://example.com/docs", is_nav=True) + _guard_route(route) + assert route.aborted is None + assert route.continued is True + + +def test_subresource_request_is_not_gated(): + # A non-navigation request (image/script/XHR) can't read internal content back into the + # agent; gating every one would resolve DNS dozens of times per page. It is let through + # even to a loopback literal — the read-back path is what the guard closes. + route = _FakeRoute("http://127.0.0.1:11434/api/tags", is_nav=False) + _guard_route(route) + assert route.aborted is None + assert route.continued is True + + +def test_guard_fails_closed_when_is_navigation_raises(monkeypatch): + # A request object that errors on inspection is treated as a navigation and vetted. + _resolves_to(monkeypatch, "127.0.0.1") + + class _Boom(_FakeRoute): + def __init__(self): + super().__init__("http://sneaky.example.com/", is_nav=True) + + def boom(): + raise RuntimeError("no such method") + + self.request.is_navigation_request = boom + + route = _Boom() + _guard_route(route) + assert route.aborted == "blockedbyclient"