fix(browser): block Camofox private redirects before auto-snapshot

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,
@@ -171,6 +173,100 @@ 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()
+
+    @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"))

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:
@@ -515,12 +517,18 @@ 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)
+        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.