From 20ea2214e080ca79d0aaeaa761af7bc616861195 Mon Sep 17 00:00:00 2001 From: dsad Date: Fri, 3 Jul 2026 01:52:51 +0300 Subject: [PATCH 1/2] fix(browser): block Camofox private redirects --- tests/tools/test_browser_camofox.py | 51 +++++++++++++++++++++++++++++ tools/browser_camofox.py | 51 +++++++++++++++++++++++++++-- 2 files changed, 100 insertions(+), 2 deletions(-) diff --git a/tests/tools/test_browser_camofox.py b/tests/tools/test_browser_camofox.py index df5bef4aff600..27fdadf18ef69 100644 --- a/tests/tools/test_browser_camofox.py +++ b/tests/tools/test_browser_camofox.py @@ -171,6 +171,57 @@ def test_navigates_existing_tab(self, mock_post, monkeypatch): assert result["success"] is True assert result["url"] == "https://b.com" + @patch("tools.browser_camofox.requests.get") + @patch("tools.browser_camofox.requests.post") + def test_blocks_metadata_redirect_before_auto_snapshot(self, mock_post, mock_get, monkeypatch): + monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") + mock_post.side_effect = [ + _mock_response(json_data={ + "tabId": "tab_meta", + "url": "http://169.254.169.254/latest/meta-data/", + }), + _mock_response(json_data={"ok": True, "url": "about:blank"}), + ] + + def fail_snapshot(*_args, **_kwargs): + raise AssertionError("Camofox auto-snapshot must not run after a blocked redirect") + + mock_get.side_effect = fail_snapshot + + result = json.loads(camofox_navigate("https://example.com/redirect", task_id="t_meta")) + + assert result["success"] is False + assert "cloud metadata endpoint" in result["error"] + assert mock_post.call_args_list[-1].kwargs["json"]["url"] == "about:blank" + mock_get.assert_not_called() + + @patch("tools.browser_camofox.requests.get") + @patch("tools.browser_camofox.requests.post") + def test_blocks_private_redirect_when_guard_active(self, mock_post, mock_get, monkeypatch): + monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") + from tools import browser_tool + + monkeypatch.setattr(browser_tool, "_eval_ssrf_guard_active", lambda task_id: True) + mock_post.side_effect = [ + _mock_response(json_data={ + "tabId": "tab_private", + "url": "http://10.0.0.5/admin", + }), + _mock_response(json_data={"ok": True, "url": "about:blank"}), + ] + + def fail_snapshot(*_args, **_kwargs): + raise AssertionError("Camofox auto-snapshot must not run after a blocked redirect") + + mock_get.side_effect = fail_snapshot + + result = json.loads(camofox_navigate("https://example.com/redirect", task_id="t_private")) + + assert result["success"] is False + assert "private/internal address" in result["error"] + assert mock_post.call_args_list[-1].kwargs["json"]["url"] == "about:blank" + mock_get.assert_not_called() + def test_connection_error_returns_helpful_message(self, monkeypatch): monkeypatch.setenv("CAMOFOX_URL", "http://localhost:19999") result = json.loads(camofox_navigate("https://example.com", task_id="t_err")) diff --git a/tools/browser_camofox.py b/tools/browser_camofox.py index fe11256aa76bb..0c71b7d1188a8 100644 --- a/tools/browser_camofox.py +++ b/tools/browser_camofox.py @@ -414,6 +414,8 @@ def _ensure_tab(task_id: Optional[str], url: str = "about:blank") -> Dict[str, A resp.raise_for_status() data = resp.json() session["tab_id"] = data.get("tabId") + if data.get("url"): + session["_last_url"] = data.get("url") return session @@ -497,7 +499,7 @@ def camofox_navigate(url: str, task_id: Optional[str] = None) -> str: if not session["tab_id"]: # Create tab with the target URL directly session = _ensure_tab(task_id, browser_url) - data = {"ok": True, "url": browser_url} + data = {"ok": True, "url": session.get("_last_url") or browser_url} else: # Navigate existing tab — recover from stale tab 404 try: @@ -518,9 +520,15 @@ def camofox_navigate(url: str, task_id: Optional[str] = None) -> str: data = {"ok": True, "url": browser_url} else: raise + final_url = data.get("url", browser_url) + blocked_redirect = _camofox_blocked_redirect_url(final_url, task_id) + if blocked_redirect: + _camofox_navigate_blank(session) + return blocked_redirect + result = { "success": True, - "url": data.get("url", browser_url), + "url": final_url, "title": data.get("title", ""), } if rewrite_info: @@ -570,6 +578,45 @@ def camofox_navigate(url: str, task_id: Optional[str] = None) -> str: return tool_error(str(e), success=False) +def _camofox_navigate_blank(session: Dict[str, Any]) -> None: + """Best-effort clear of a blocked Camofox tab.""" + try: + if session.get("tab_id"): + _post( + f"/tabs/{session['tab_id']}/navigate", + {"userId": session["user_id"], "url": "about:blank"}, + timeout=10, + ) + except Exception as exc: + logger.debug("Camofox blocked-page clear failed: %s", exc) + + +def _camofox_blocked_redirect_url(final_url: str, task_id: Optional[str]) -> Optional[str]: + """Return a blocked payload when Camofox navigation lands on a private URL.""" + if not final_url: + return None + + from tools.browser_tool import ( + _eval_ssrf_guard_active, + _is_always_blocked_url, + _is_safe_url, + ) + + if _is_always_blocked_url(final_url): + return json.dumps({ + "success": False, + "error": "Blocked: redirect landed on a cloud metadata endpoint", + }, ensure_ascii=False) + + if _eval_ssrf_guard_active(task_id or "default") and not _is_safe_url(final_url): + return json.dumps({ + "success": False, + "error": "Blocked: redirect landed on a private/internal address", + }, ensure_ascii=False) + + return None + + def _camofox_private_page_block(session: Dict[str, Any], task_id: Optional[str], action: str) -> Optional[str]: """Return a blocked payload when the current Camofox page is private/internal. From 1806a5413420ed59585c13d0acf8984011aec00c Mon Sep 17 00:00:00 2001 From: dsad Date: Fri, 3 Jul 2026 06:08:42 +0300 Subject: [PATCH 2/2] fix(browser): guard Camofox stale-tab redirects --- tests/tools/test_browser_camofox.py | 45 +++++++++++++++++++++++++++++ tools/browser_camofox.py | 2 +- 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/tests/tools/test_browser_camofox.py b/tests/tools/test_browser_camofox.py index 27fdadf18ef69..b0e88c9f45275 100644 --- a/tests/tools/test_browser_camofox.py +++ b/tests/tools/test_browser_camofox.py @@ -3,6 +3,8 @@ import json from unittest.mock import MagicMock, patch +import requests + from tools.browser_camofox import ( camofox_back, @@ -222,6 +224,49 @@ def fail_snapshot(*_args, **_kwargs): assert mock_post.call_args_list[-1].kwargs["json"]["url"] == "about:blank" mock_get.assert_not_called() + @patch("tools.browser_camofox.requests.get") + @patch("tools.browser_camofox.requests.post") + def test_blocks_stale_tab_recreate_metadata_redirect_before_auto_snapshot( + self, + mock_post, + mock_get, + monkeypatch, + ): + monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") + stale = _mock_response(status=404, json_data={"error": "gone"}) + stale.raise_for_status.side_effect = requests.HTTPError(response=stale) + mock_post.side_effect = [ + _mock_response(json_data={"tabId": "tab_stale", "url": "https://safe.test"}), + stale, + _mock_response(json_data={ + "tabId": "tab_recreated", + "url": "http://169.254.169.254/latest/meta-data/", + }), + _mock_response(json_data={"ok": True, "url": "about:blank"}), + ] + + def get_side_effect(url, *_args, **_kwargs): + if str(url).endswith("/health"): + return _mock_response(json_data={"ok": True}) + if "/tabs/tab_stale/snapshot" in str(url): + return _mock_response(json_data={"snapshot": "- heading \"Safe\"", "refsCount": 1}) + if "/tabs/tab_recreated/snapshot" in str(url): + raise AssertionError("Camofox auto-snapshot must not run after a blocked redirect") + return _mock_response(json_data={}) + + mock_get.side_effect = get_side_effect + + camofox_navigate("https://safe.test", task_id="t_stale_meta") + result = json.loads(camofox_navigate("https://redirect.test", task_id="t_stale_meta")) + + assert result["success"] is False + assert "cloud metadata endpoint" in result["error"] + assert mock_post.call_args_list[-1].kwargs["json"]["url"] == "about:blank" + assert not any( + "/tabs/tab_recreated/snapshot" in str(call.args[0]) + for call in mock_get.call_args_list + ) + def test_connection_error_returns_helpful_message(self, monkeypatch): monkeypatch.setenv("CAMOFOX_URL", "http://localhost:19999") result = json.loads(camofox_navigate("https://example.com", task_id="t_err")) diff --git a/tools/browser_camofox.py b/tools/browser_camofox.py index 0c71b7d1188a8..e01c5054ebe2d 100644 --- a/tools/browser_camofox.py +++ b/tools/browser_camofox.py @@ -517,7 +517,7 @@ def camofox_navigate(url: str, task_id: Optional[str] = None) -> str: ) session["tab_id"] = None session = _ensure_tab(task_id, browser_url) - data = {"ok": True, "url": browser_url} + data = {"ok": True, "url": session.get("_last_url") or browser_url} else: raise final_url = data.get("url", browser_url)