From 7432f85ce2d17ec2c823228b2fe7f6a6a3385ae0 Mon Sep 17 00:00:00 2001 From: Pratik Rai Date: Mon, 27 Apr 2026 03:38:41 +0530 Subject: [PATCH] fix(security): prevent SSRF bypass for IMDS endpoints in browser routing --- tests/tools/test_browser_ssrf_local.py | 140 +++++++++++++++++++++++++ tools/browser_tool.py | 72 +++++++------ 2 files changed, 182 insertions(+), 30 deletions(-) diff --git a/tests/tools/test_browser_ssrf_local.py b/tests/tools/test_browser_ssrf_local.py index 27b6e3933b69..33124d15f565 100644 --- a/tests/tools/test_browser_ssrf_local.py +++ b/tests/tools/test_browser_ssrf_local.py @@ -235,3 +235,143 @@ def test_cloud_allows_redirect_to_public(self, monkeypatch, _common_patches): assert result["success"] is True assert result["url"] == final + + +# --------------------------------------------------------------------------- +# IMDS blocking with hybrid routing (Issue #16234) +# --------------------------------------------------------------------------- + + +class TestImdsBlockingWithHybridRouting: + """Verify IMDS endpoints are blocked even when hybrid routing is enabled. + + This tests the fix for Issue #16234: the pre-navigation SSRF guard must + run BEFORE the hybrid routing decision, not after. Previously, when + auto_local_for_private_urls was enabled and a cloud provider was configured, + the SSRF check was skipped entirely, allowing access to 169.254.169.254. + """ + + AWS_IMDS = "http://169.254.169.254/latest/meta-data/" + AWS_IMDS_V2 = "http://169.254.169.254/latest/meta-data/iam/security-credentials/" + GCP_IMDS = "http://metadata.google.internal/computeMetadata/v1/instance/hostname" + AZURE_IMDS = "http://169.254.169.253/metadata/instance" + ALIYUN_IMDS = "http://100.100.100.200/latest/meta-data/" + + @pytest.fixture() + def _cloud_mode(self, monkeypatch): + """Configure cloud backend mode.""" + monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: False) + monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: "browserbase") + monkeypatch.setattr(browser_tool, "check_website_access", lambda url: None) + monkeypatch.setattr( + browser_tool, + "_get_session_info", + lambda task_id: { + "session_name": f"s_{task_id}", + "bb_session_id": "bb-123", + "cdp_url": None, + "features": {}, + "_first_nav": False, + }, + ) + + @pytest.fixture() + def _hybrid_routing_enabled(self, monkeypatch): + """Enable hybrid auto-local routing for private URLs.""" + monkeypatch.setattr(browser_tool, "_auto_local_for_private_urls", lambda: True) + + def test_blocks_aws_imds_with_hybrid_routing( + self, monkeypatch, _cloud_mode, _hybrid_routing_enabled + ): + """AWS IMDS is blocked even when hybridRouting is enabled.""" + monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False) + + result = json.loads(browser_tool.browser_navigate(self.AWS_IMDS)) + + assert result["success"] is False + assert "private or internal address" in result["error"] + + def test_blocks_aws_imds_v2_with_hybrid_routing( + self, monkeypatch, _cloud_mode, _hybrid_routing_enabled + ): + """AWS IMDS v2 (169.254.169.254) is blocked.""" + monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False) + + result = json.loads(browser_tool.browser_navigate(self.AWS_IMDS_V2)) + + assert result["success"] is False + + def test_blocks_gcp_imds_with_hybrid_routing( + self, monkeypatch, _cloud_mode, _hybrid_routing_enabled + ): + """GCP IMDS (metadata.google.internal) is blocked.""" + monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False) + + result = json.loads(browser_tool.browser_navigate(self.GCP_IMDS)) + + assert result["success"] is False + assert "private or internal address" in result["error"] + + def test_blocks_azure_imds_with_hybrid_routing( + self, monkeypatch, _cloud_mode, _hybrid_routing_enabled + ): + """Azure IMDS (169.254.169.253) is blocked.""" + monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False) + + result = json.loads(browser_tool.browser_navigate(self.AZURE_IMDS)) + + assert result["success"] is False + + def test_blocks_aliyun_imds_with_hybrid_routing( + self, monkeypatch, _cloud_mode, _hybrid_routing_enabled + ): + """Aliyun IMDS (100.100.100.200) is blocked.""" + monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False) + + result = json.loads(browser_tool.browser_navigate(self.ALIYUN_IMDS)) + + assert result["success"] is False + + def test_blocks_link_local_range_with_hybrid_routing( + self, monkeypatch, _cloud_mode, _hybrid_routing_enabled + ): + """Entire 169.254.0.0/16 range is blocked.""" + monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False) + + result = json.loads( + browser_tool.browser_navigate("http://169.254.42.99/anything") + ) + + assert result["success"] is False + + def test_allows_public_url_with_hybrid_routing( + self, monkeypatch, _cloud_mode, _hybrid_routing_enabled + ): + """Public URLs pass even with hybrid routing enabled.""" + monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: False) + monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: True) + monkeypatch.setattr( + browser_tool, + "_run_browser_command", + lambda *a, **kw: _make_browser_result(), + ) + + result = json.loads(browser_tool.browser_navigate("https://example.com")) + + assert result["success"] is True + + def test_hybrid_routing_still_works_for_legitimate_private( + self, monkeypatch, _cloud_mode, _hybrid_routing_enabled + ): + """Legitimate private URLs (non-IMDS) still route to local when allowed.""" + monkeypatch.setattr(browser_tool, "_allow_private_urls", lambda: True) + monkeypatch.setattr(browser_tool, "_is_safe_url", lambda url: True) + monkeypatch.setattr( + browser_tool, + "_run_browser_command", + lambda *a, **kw: _make_browser_result(), + ) + + result = json.loads(browser_tool.browser_navigate("http://192.168.1.1:8080/")) + + assert result["success"] is True diff --git a/tools/browser_tool.py b/tools/browser_tool.py index aecb2ee7f65e..5d8f6ff2d199 100644 --- a/tools/browser_tool.py +++ b/tools/browser_tool.py @@ -51,6 +51,8 @@ import atexit import functools +import importlib +import ipaddress import json import logging import os @@ -516,6 +518,27 @@ def _auto_local_for_private_urls() -> bool: return _cached_auto_local_for_private_urls +# Explicit private ranges for Python 3.10 compatibility. +# ``ipaddress.ip_address.is_private`` only covers 172.16.0.0/12 in Python ≥3.11 +# (expanded in bpo-40791). These are the complete RFC1918 private ranges +# plus link-local and ULA (fd00::/8). +_PRIVATE_IP_RANGES = ( + ipaddress.ip_network("10.0.0.0/8"), + ipaddress.ip_network("172.16.0.0/12"), + ipaddress.ip_network("192.168.0.0/16"), + ipaddress.ip_network("169.254.0.0/16"), + ipaddress.ip_network("fd00::/8"), +) + + +def _ip_in_private_range(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: + """Check if IP is in RFC1918 private ranges (Python 3.10 compatible). + + Uses explicit ranges instead of ip.is_private for Python 3.10 compatibility. + """ + return any(ip in net for net in _PRIVATE_IP_RANGES) + + def _url_is_private(url: str) -> bool: """Return True when the URL's host resolves to a private/LAN/loopback address. @@ -524,14 +547,10 @@ def _url_is_private(url: str) -> bool: resolution failures are treated as NOT private (fall through to whatever backend is configured, which will surface the DNS error naturally). """ + from urllib.parse import urlparse + import socket + try: - from tools.url_safety import is_safe_url - # is_safe_url returns False for private/loopback/link-local/CGNAT AND - # for DNS failures. We only want the private-network case here, so - # we parse + check the host shape as a DNS-failure sieve first. - from urllib.parse import urlparse - import ipaddress - import socket parsed = urlparse(url) hostname = (parsed.hostname or "").strip().lower().rstrip(".") if not hostname: @@ -540,10 +559,10 @@ def _url_is_private(url: str) -> bool: try: ip = ipaddress.ip_address(hostname) return ( - ip.is_private + _ip_in_private_range(ip) or ip.is_loopback or ip.is_link_local - or ip in ipaddress.ip_network("100.64.0.0/10") + or ip in ipaddress.ip_network("100.64.0.0/10") # CGNAT ) except ValueError: pass @@ -564,10 +583,10 @@ def _url_is_private(url: str) -> bool: except ValueError: continue if ( - ip.is_private + _ip_in_private_range(ip) or ip.is_loopback or ip.is_link_local - or ip in ipaddress.ip_network("100.64.0.0/10") + or ip in ipaddress.ip_network("100.64.0.0/10") # CGNAT ): return True return False @@ -1687,21 +1706,13 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str: "Secrets must not be sent in URLs.", }) - # SSRF protection — block private/internal addresses before navigating. + # SSRF protection — always block dangerous URLs before any navigation. + # This runs FIRST, before routing decision, because safety is absolute. # Skipped for local backends (Camofox, headless Chromium without a cloud - # provider) because the agent already has full local network access via - # the terminal tool. Also skipped when hybrid routing will auto-spawn a - # local Chromium sidecar for this URL (cloud provider configured + - # private URL + ``browser.auto_local_for_private_urls`` enabled) — the - # cloud provider never sees the URL in that case. Can also be opted - # out globally via ``browser.allow_private_urls`` in config. - effective_task_id = task_id or "default" - nav_session_key = _navigation_session_key(effective_task_id, url) - auto_local_this_nav = _is_local_sidecar_key(nav_session_key) - + # provider) because the agent already has full local network access. + # Also skipped when user has explicitly opted out via ``browser.allow_private_urls``. if ( not _is_local_backend() - and not auto_local_this_nav and not _allow_private_urls() and not _is_safe_url(url) ): @@ -1710,6 +1721,11 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str: "error": "Blocked: URL targets a private or internal address", }) + # Routing decision — now separate from safety check. + effective_task_id = task_id or "default" + nav_session_key = _navigation_session_key(effective_task_id, url) + auto_local_this_nav = _is_local_sidecar_key(nav_session_key) + # Website policy check — block before navigating blocked = check_website_access(url) if blocked: @@ -1755,15 +1771,11 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str: title = data.get("title", "") final_url = data.get("url", url) - # Post-redirect SSRF check — if the browser followed a redirect to a - # private/internal address, block the result so the model can't read - # internal content via subsequent browser_snapshot calls. - # Skipped for local backends (same rationale as the pre-nav check), - # and for the hybrid local sidecar (we're already on a local browser - # hitting a private URL by design). + # Post-redirect SSRF check — always block if redirect lands on dangerous URL. + # This runs regardless of routing destination because safety is absolute. + # Skipped for local backends (same rationale as the pre-nav check). if ( not _is_local_backend() - and not auto_local_this_nav and not _allow_private_urls() and final_url and final_url != url and not _is_safe_url(final_url) ):